from flask import Flask, request, jsonify
import xmlrpc.client
import xmltodict
import os
import logging
import re
# Initialize Flask app
app = Flask(__name__)
# Configure logging
logging.basicConfig(level=logging.DEBUG)
# Load environment variables
ECONET_USERNAME = os.environ.get("ECONET_USERNAME")
ECONET_PASSWORD = os.environ.get("ECONET_PASSWORD")
ECONET_API_URL = os.environ.get("ECONET_API_URL",
"https://your_econet_api_url/xmlrpc")
# Validate environment variables
if not ECONET_USERNAME or not ECONET_PASSWORD:
raise ValueError("Econet username and password must be set as environment
variables.")
if not ECONET_API_URL or ECONET_API_URL == "https://your_econet_api_url/xmlrpc":
raise ValueError("ECONET_API_URL must be set to a valid endpoint.")
# Helper function to validate MSISDN format (basic validation)
def validate_msisdn_format(msisdn):
pattern = r"^\+?1?\d{9,15}$" # Adjust pattern as per the MSISDN format
return re.match(pattern, msisdn)
# Common function to handle API request and response validation
def common_api_request(method_name, api_params, required_fields):
"""Handles API request and validates response structure."""
response = call_econet_api(method_name, api_params)
if validate_response(response, required_fields):
return jsonify(response), 200
return jsonify({"error": "Invalid response structure from Econet API"}), 500
def call_econet_api(method_name, params):
"""Calls the Econet XML-RPC API."""
try:
server = xmlrpc.client.ServerProxy(ECONET_API_URL)
api_params = [ECONET_USERNAME, ECONET_PASSWORD]
api_params.append(params)
result = getattr(server, method_name)(*api_params)
if isinstance(result, str):
# Use xmltodict for simplified XML parsing
try:
response_data = xmltodict.parse(result)
return response_data
except Exception as e:
logging.error(f"Failed to parse XML: {e}")
return None
else:
logging.error(f"Unexpected response from Econet API: {result}")
return None
except xmlrpc.client.Fault as e:
logging.error(f"XML-RPC Fault: {e}")
return None
except Exception as e:
logging.error(f"An error occurred: {e}")
return None
def validate_response(response, required_fields):
"""Validates that the response contains the required fields."""
if not response:
return False
for field in required_fields:
if field not in response:
return False
return True
@app.route("/api/airtime/load", methods=["POST"])
def load_airtime():
"""Loads airtime to a specified MSISDN."""
data = request.get_json()
msisdn = data.get("msisdn")
amount = data.get("amount")
reference = data.get("reference")
currency = data.get("currency", 840) # Default to USD (840) if not provided
# Validate required fields
if not msisdn or not amount or not reference:
return jsonify({"error": "Missing msisdn, amount, or reference"}), 400
# Validate MSISDN format
if not validate_msisdn_format(msisdn):
return jsonify({"error": "Invalid MSISDN format"}), 400
# Validate amount and currency
try:
amount = int(amount)
currency = int(currency)
except ValueError:
return jsonify({"error": "Amount and Currency must be valid integers"}),
400
api_params = {
"MSISDN": str(msisdn),
"Amount": amount,
"Reference": str(reference),
"Currency": currency # Defaults to 840 (USD)
}
return common_api_request("load_value", api_params, ["Status", "StatusCode",
"Description"])
@app.route("/api/bundle/load", methods=["POST"])
def load_bundle():
"""Purchases a data bundle for a specified MSISDN."""
data = request.get_json()
msisdn = data.get("msisdn")
provider_code = data.get("provider_code")
amount = data.get("amount")
currency = data.get("currency", 840) # Default to USD (840) if not provided
account_type = data.get("account_type")
quantity = data.get("quantity")
reference = data.get("reference")
product_code = data.get("product_code")
# Validate required fields
if not msisdn or not provider_code or not amount or not currency or not
account_type or not quantity or not reference or not product_code:
return jsonify({"error": "Missing required parameters for load_bundle"}),
400
# Validate MSISDN format
if not validate_msisdn_format(msisdn):
return jsonify({"error": "Invalid MSISDN format"}), 400
# Validate amount, currency, and quantity
try:
amount = int(amount)
currency = int(currency)
quantity = int(quantity)
except ValueError:
return jsonify({"error": "Amount, Currency, and Quantity must be valid
integers"}), 400
api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code),
"Amount": amount,
"Currency": currency, # Defaults to 840 (USD)
"AccountType": int(account_type),
"Quantity": quantity,
"Reference": str(reference),
"ProductCode": str(product_code)
}
return common_api_request("load_bundle", api_params, ["Status", "StatusCode",
"Description"])
@app.route("/api/value/deduct", methods=["POST"])
def deduct_value():
"""Deducts airtime value from a specified MSISDN."""
data = request.get_json()
msisdn = data.get("msisdn")
provider_code = data.get("provider_code")
amount = data.get("amount")
reference = data.get("reference")
# Validate required fields
if not msisdn or not provider_code or not amount or not reference:
return jsonify({"error": "Missing required parameters for deduct_value"}),
400
# Validate MSISDN format
if not validate_msisdn_format(msisdn):
return jsonify({"error": "Invalid MSISDN format"}), 400
# Validate amount
try:
amount = int(amount)
except ValueError:
return jsonify({"error": "Amount must be a valid integer"}), 400
api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code),
"Amount": amount,
"Reference": str(reference)
}
return common_api_request("deduct_value", api_params, ["Status", "StatusCode",
"Description"])
@app.route("/api/balance", methods=["POST"])
def account_balance_enquiry():
"""Enquires the balance of a specified MSISDN."""
data = request.get_json()
msisdn = data.get("msisdn")
provider_code = data.get("provider_code")
# Validate required fields
if not msisdn or not provider_code:
return jsonify({"error": "Missing msisdn and provider_code"}), 400
# Validate MSISDN format
if not validate_msisdn_format(msisdn):
return jsonify({"error": "Invalid MSISDN format"}), 400
api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code)
}
return common_api_request("account_balance_enquiry", api_params,
["AccountType", "Currency", "Amount"])
@app.route("/api/msisdn/validate", methods=["POST"])
def validate_msisdn():
"""Validates a specified MSISDN."""
data = request.get_json()
msisdn = data.get("msisdn")
provider_code = data.get("provider_code")
# Validate required fields
if not msisdn or not provider_code:
return jsonify({"error": "Missing msisdn and provider_code"}), 400
# Validate MSISDN format
if not validate_msisdn_format(msisdn):
return jsonify({"error": "Invalid MSISDN format"}), 400
api_params = {
"MSISDN": str(msisdn),
"ProviderCode": int(provider_code)
}
return common_api_request("validate_msisdn", api_params, ["Status",
"StatusCode", "Description"])
@app.route("/api/transaction/status", methods=["POST"])
def get_transaction_status():
"""Checks the status of a previously submitted transaction."""
data = request.get_json()
reference = data.get("reference")
# Validate required fields
if not reference:
return jsonify({"error": "Missing reference"}), 400
api_params = {
"Reference": str(reference)
}
return common_api_request("get_transaction_status", api_params, ["Status",
"StatusCode", "Description"])
@app.route("/api/account/transfer", methods=["POST"])
def account_transfer():
"""Transfers funds between company accounts."""
data = request.get_json()
from_company_id = data.get("from_company_id")
to_company_id = data.get("to_company_id")
currency = data.get("currency", 840) # Default to USD (840) if not provided
amount = data.get("amount")
reference = data.get("reference")
# Validate required fields
if not from_company_id or not to_company_id or not currency or not amount or
not reference:
return jsonify({"error": "Missing required parameters for
account_transfer"}), 400
# Validate amount and currency
try:
amount = int(amount)
currency = int(currency)
except ValueError:
return jsonify({"error": "Amount and Currency must be valid integers"}),
400
api_params = {
"FromCompanyID": int(from_company_id),
"ToCompanyID": int(to_company_id),
"Currency": currency, # Defaults to 840 (USD)
"Amount": amount,
"Reference": str(reference)
}
return common_api_request("account_transfer", api_params, ["Status",
"StatusCode", "Description"])
if __name__ == "__main__":
app.run(debug=True)
