# recharge/services/goterpay_api.py from __future__ import annotations import logging from dataclasses import dataclass from typing import Any, Dict, Optional from requests.adapters import HTTPAdapter from urllib3.util import Retry import requests logger = logging.getLogger(__name__) API_BASE = "https://api.goterpay.com" DASHBOARD_BASE = "https://dashboard.goterpay.com/api" DEFAULT_TIMEOUT = 20 # seconds class GoterPayError(Exception): def __init__(self, message: str, *, status_code: Optional[int] = None, payload: Any = None): super().__init__(message) self.status_code = status_code self.payload = payload @dataclass class GoterPayConfig: mid: str mkey: str subwallet: Optional[str] = None # required for recharge/billpay per docs timeout: int = DEFAULT_TIMEOUT class GoterPayAPI: """ Exact wrapper for the documented GoterPay endpoints. Only implements the endpoints you provided. """ def __init__(self, cfg: GoterPayConfig): self.cfg = cfg self.session = requests.Session() self.session.headers.update({"Accept": "application/json", "User-Agent": "ViralPe/1.0"}) retry = Retry(total=3, backoff_factor=0.3, status_forcelist=[429, 500, 502, 503, 504]) self.session.mount("https://", HTTPAdapter(max_retries=retry)) # --------------------- # Helpers # --------------------- def _get(self, base: str, path: str, params: Dict[str, Any]) -> Dict[str, Any]: q = {"mid": self.cfg.mid, "mkey": self.cfg.mkey} q.update({k: v for k, v in params.items() if v is not None}) url = f"{base.rstrip('/')}/{path.lstrip('/')}" r = self.session.get(url, params=q, timeout=self.cfg.timeout) if r.status_code != 200: raise GoterPayError(f"HTTP {r.status_code} from GoterPay", status_code=r.status_code, payload=r.text) try: return r.json() except Exception: # Some endpoints may return non-JSON (plans can be text on some providers). # Fall back to raw text so caller can inspect. return {"raw": r.text} @staticmethod def normalize_status(value: Optional[str]) -> str: """ Map provider statuses to SUCCESS | PENDING | FAILED | ERROR (simple). """ if not value: return "ERROR" v = str(value).strip().upper() if v in {"SUCCESS", "SUCESS"}: return "SUCCESS" if v in {"PENDING", "PROCESSING"}: return "PENDING" if v in {"FAILED", "FAIL", "ERROR"}: return "FAILED" if v == "FAILED" else "ERROR" # MobileInfo uses "success" lower-case if v == "SUCCESS".upper(): return "SUCCESS" return value.upper() # --------------------- # Endpoints (exact) # --------------------- # Mobile Info Api # GET https://api.goterpay.com/mobileinfo?mid=...&mkey=...&mobile=... def mobile_info(self, mobile: str) -> Dict[str, Any]: return self._get(API_BASE, "mobileinfo", {"mobile": mobile}) # Recharge Plan Api # GET https://api.goterpay.com/Rplan?mid=...&mkey=...&operator=...&circle=... def recharge_plan(self, operator_code: str, circle_code: str) -> Dict[str, Any]: return self._get(API_BASE, "Rplan", {"operator": operator_code, "circle": circle_code}) # R Offer Api # GET https://api.goterpay.com/Roffer?mid=...&mkey=...&operator=...&number=... def r_offer(self, operator_code: str, number: str) -> Dict[str, Any]: return self._get(API_BASE, "Roffer", {"operator": operator_code, "number": number}) # DTH Info Api # GET https://api.goterpay.com/dthinfo?mid=...&mkey=...&operator=...&number=... def dth_info(self, operator_code: str, number: str) -> Dict[str, Any]: return self._get(API_BASE, "dthinfo", {"operator": operator_code, "number": number}) # Mobile Recharge API # GET https://dashboard.goterpay.com/api/Recharge?mid=...&mkey=...&subwallet=...&txnid=...&number=...&amount=...&operator=...&circle=... def mobile_recharge( self, *, txnid: str, number: str, amount: str | int | float, operator_code: str, circle_code: str, subwallet: Optional[str] = None, ) -> Dict[str, Any]: sw = subwallet or self.cfg.subwallet if not sw: raise GoterPayError("subwallet is required for Mobile Recharge API") return self._get( DASHBOARD_BASE, "Recharge", { "subwallet": sw, "txnid": txnid, "number": str(number), "amount": str(amount), "operator": operator_code, "circle": circle_code, }, ) # BillFetch API # GET https://dashboard.goterpay.com/api/BillFetch?mid=...&mkey=...&number=...&operator=...&txnid=...&optional1=... def bill_fetch( self, *, number: str, operator_code: str, txnid: str, optional1: Optional[str] = None ) -> Dict[str, Any]: return self._get( DASHBOARD_BASE, "BillFetch", {"number": number, "operator": operator_code, "txnid": txnid, "optional1": optional1}, ) # BillPay API # GET https://dashboard.goterpay.com/api/BillPay?mid=...&mkey=...&subwallet=...&txnid=...&number=...&amount=...&operator=...&optional1=... def bill_pay( self, *, txnid: str, number: str, amount: str | int | float, operator_code: str, optional1: Optional[str] = None, subwallet: Optional[str] = None, ) -> Dict[str, Any]: sw = subwallet or self.cfg.subwallet if not sw: raise GoterPayError("subwallet is required for BillPay API") return self._get( DASHBOARD_BASE, "BillPay", { "subwallet": sw, "txnid": txnid, "number": str(number), "amount": str(amount), "operator": operator_code, "optional1": optional1, }, ) # Status API # GET https://dashboard.goterpay.com/api/Status?mid=...&mkey=...&txnid=... def status(self, txnid: str) -> Dict[str, Any]: return self._get(DASHBOARD_BASE, "Status", {"txnid": txnid}) # Complaint API # GET https://dashboard.goterpay.com/api/Complaint?mid=...&mkey=...&txnid=...&remark=... def complaint(self, *, txnid: str, remark: str) -> Dict[str, Any]: return self._get(DASHBOARD_BASE, "Complaint", {"txnid": txnid, "remark": remark}) # Notes (kept tight) # Uses only GET with query params exactly as in your samples. # Splits bases: api.goterpay.com vs dashboard.goterpay.com/api. # Requires subwallet for Mobile Recharge and BillPay (throws a clear error if missing). # No speculative endpoints for “Operator Code” or “Circle Code” since no URL/spec was provided—share those when ready and I’ll add them.