import json import logging import hashlib from typing import Dict, Any, Optional from urllib.parse import urlencode import httpx from app.config import settings from app.services.payment_gateway import PaymentGateway logger = logging.getLogger(__name__) PINGPONG_HOST_SANDBOX = "https://sandbox-acquirer-payment.pingpongx.com" PINGPONG_HOST_PROD_EU = "https://acquirer-payment.pingpongx.com" PINGPONG_HOST_PROD_US = "https://acquirer-payment-checkout-us.pingpongx.com" class PingPongCheckoutService(PaymentGateway): name = "pingpong" supported_types = ["pingpong", "card"] def __init__(self): self.client_id = settings.PINGPONG_CLIENT_ID or "" self.acc_id = settings.PINGPONG_ACC_ID or "" self.secret_key = settings.PINGPONG_SECRET_KEY or "" self.sandbox = settings.PINGPONG_SANDBOX if self.sandbox: self._base_url = PINGPONG_HOST_SANDBOX else: region = (settings.PINGPONG_REGION or "EU").upper() if region == "US": self._base_url = PINGPONG_HOST_PROD_US else: self._base_url = PINGPONG_HOST_PROD_EU def _is_configured(self) -> bool: return bool(self.client_id and self.acc_id and self.secret_key) def _sign(self, params: dict) -> str: sorted_keys = sorted(k for k in params if k != "sign") sign_str = "&".join(f"{k}={params[k]}" for k in sorted_keys) raw = sign_str + self.secret_key return hashlib.sha256(raw.encode("utf-8")).hexdigest().upper() def _verify_sign(self, params: dict) -> bool: if "sign" not in params: return False expected = self._sign(params) return expected == params["sign"].upper() async def _request(self, path: str, biz_data: dict) -> dict: biz_json = json.dumps(biz_data, separators=(",", ":")) body = { "accId": self.acc_id, "clientId": self.client_id, "signType": "SHA256", "version": "1.0", "bizContent": biz_json, } body["sign"] = self._sign(body) async with httpx.AsyncClient() as client: resp = await client.post( f"{self._base_url}{path}", json=body, headers={"Content-Type": "application/json"}, timeout=30, ) if resp.status_code != 200: raise ValueError(f"PingPong request failed: {resp.status_code} {resp.text}") data = resp.json() if data.get("code") != "000000": raise ValueError(f"PingPong error: {data.get('code')} - {data.get('description', '')}") return data async def create_order(self, order_no: str, amount: int, description: str, **kwargs) -> Dict[str, Any]: if not self._is_configured(): raise ValueError("PingPong 未配置") usd_amount = round(amount / 100, 2) usd_str = f"{usd_amount:.2f}" success_url = kwargs.get("success_url", "https://trade.yuzhiran.com/workspace/credits?pingpong=success") cancel_url = kwargs.get("cancel_url", "https://trade.yuzhiran.com/workspace/credits?pingpong=cancel") notification_url = kwargs.get("notification_url", "https://trade.yuzhiran.com/api/v1/payment/pingpong-webhook") shopper_ip = kwargs.get("shopper_ip", "127.0.0.1") biz_data = { "merchantTransactionId": order_no, "amount": usd_str, "currency": "USD", "paymentType": "SALE", "shopperIP": shopper_ip, "notificationUrl": notification_url, "payResultUrl": success_url, "cancelUrl": cancel_url, "goods": [{ "name": description[:127] or "Credit Package", "unitPrice": usd_str, "number": "1", }], } result = await self._request("/v4/payment/prePay", biz_data) bc = result.get("bizContent", {}) return { "gateway_order_id": bc.get("transactionId", ""), "merchant_order_id": order_no, "session_url": bc.get("paymentUrl", ""), "session_id": bc.get("token", ""), "amount": usd_amount, "status": "pending", } async def query_order(self, order_no: str) -> Dict[str, Any]: if not self._is_configured(): return {"status": "unknown"} biz_data = {"merchantTransactionId": order_no} try: result = await self._request("/v4/payment/query", biz_data) bc = result.get("bizContent", {}) return { "status": bc.get("status", "unknown"), "payment_status": "completed" if bc.get("status") == "SUCCESS" else bc.get("status", ""), "amount": float(bc.get("amount", 0)), "currency": bc.get("currency", "USD"), } except Exception as e: logger.error(f"PingPong query failed: {e}") return {"status": "unknown"} async def refund(self, order_no: str, amount: int, reason: str = "") -> Dict[str, Any]: usd_amount = round(amount / 100, 2) biz_data = { "merchantTransactionId": order_no, "amount": f"{usd_amount:.2f}", "currency": "USD", "reason": reason or "Refund", } try: result = await self._request("/v4/refund", biz_data) return {"status": "COMPLETED", "refund_id": result.get("bizContent", {}).get("refundId", "")} except Exception as e: raise ValueError(f"PingPong refund failed: {e}") async def query_refund(self, order_no: str) -> Dict[str, Any]: biz_data = {"merchantTransactionId": order_no} try: result = await self._request("/v4/refund/query", biz_data) return {"status": result.get("bizContent", {}).get("status", "unknown")} except Exception: return {"status": "not_implemented"} def verify_callback(self, headers: dict, body: str) -> bool: try: data = json.loads(body) return self._verify_sign(data) except (json.JSONDecodeError, KeyError) as e: logger.error(f"PingPong callback verify failed: {e}") return False def parse_callback(self, body: str, headers: dict) -> Dict[str, Any]: data = json.loads(body) bc_raw = data.get("bizContent", "{}") if isinstance(bc_raw, str): try: bc = json.loads(bc_raw) except (json.JSONDecodeError, TypeError): bc = {} else: bc = bc_raw status = bc.get("status", "") return { "event": status, "order_no": bc.get("merchantTransactionId", ""), "gateway_order_id": bc.get("transactionId", ""), "gateway_order_no": bc.get("transactionId", ""), "amount": float(bc.get("amount", 0)), "success": status == "SUCCESS", "raw": bc, } async def close_order(self, order_no: str) -> Dict[str, Any]: return {"status": "ok"}