"""Signed client for Huihui's production payment-v3 service.""" import hashlib import random import string from datetime import datetime, timedelta, timezone from typing import Any import httpx _CN_TZ = timezone(timedelta(hours=8)) class HuihuiPaymentError(RuntimeError): pass class HuihuiPaymentClient: def __init__(self, config: dict): self.base_url = config.get( "HUIHUI_PAYMENT_BASE_URL", "https://open.99hui.com/api/payment-v3" ).rstrip("/") self.app_id = config.get("HUIHUI_APP_ID", "") self.access_id = config.get("HUIHUI_ACCESS_ID", "") self.access_secret = config.get("HUIHUI_ACCESS_SECRET", "") self.timeout = float(config.get("HUIHUI_PAYMENT_TIMEOUT_SECONDS", 30)) @property def configured(self) -> bool: return bool(self.base_url and self.app_id and self.access_id and self.access_secret) def _signed_params(self, user_id: str) -> dict: params = { "appId": self.app_id, "accessId": self.access_id, "nonce": "".join(random.choices(string.ascii_lowercase + string.digits, k=12)), "timestamp": datetime.now(_CN_TZ).strftime("%Y%m%d%H%M%S"), "signType": "MD5", "signVersion": "1.0", "userId": user_id, } source = "&".join( f"{key}={params[key]}" for key in sorted(params) if params[key] not in (None, "", []) ) source += f"&accessSecret={self.access_secret}" params["signature"] = hashlib.md5(source.encode("utf-8")).hexdigest().upper() return params @staticmethod def _json(response: httpx.Response) -> dict: try: payload = response.json() except ValueError as exc: raise HuihuiPaymentError("会会支付返回了无效响应") from exc if not isinstance(payload, dict): raise HuihuiPaymentError("会会支付返回格式不正确") return payload def create_payment( self, *, huihui_token: str, huihui_user_id: str, real_name: str, order_no: str, amount: str, points_amount: int, pay_type: str, pay_way: str, callback_url: str, ) -> dict[str, Any]: if not self.configured: raise HuihuiPaymentError("会会支付服务未配置") if not huihui_token or not huihui_user_id: raise HuihuiPaymentError("当前会会登录凭证无法发起支付") now = datetime.now(_CN_TZ) body = { "appId": self.app_id, "callbackUrl": callback_url, "chargeType": 4, "currency": "cny", "description": f"充值 {points_amount} 积分", "expend": {}, "masterOrderAmt": amount, "masterOrderNo": order_no, "memberId": huihui_user_id, "orderDesc": "数字分身积分充值", "orderTime": now.isoformat(), "orderTitle": "数字分身积分充值", "payAmt": float(amount), "payType": pay_type, "payWay": pay_way, "realName": real_name or "会会用户", "timeExpire": (now + timedelta(hours=2)).strftime("%Y%m%d%H%M%S"), } headers = { "Authorization": f"Bearer {huihui_token}", "appId": self.app_id, "windowAppId": self.app_id, } try: response = httpx.post( f"{self.base_url}/payment/pay", headers=headers, params=self._signed_params(huihui_user_id), json=body, timeout=self.timeout, follow_redirects=True, ) except httpx.HTTPError as exc: raise HuihuiPaymentError("会会支付连接失败,请稍后重试") from exc payload = self._json(response) code = payload.get("code") if response.status_code >= 400 or code not in (0, 200, "0", "200"): raise HuihuiPaymentError(payload.get("message") or "会会支付下单失败") data = payload.get("data") or {} if not isinstance(data, dict): raise HuihuiPaymentError("会会支付未返回订单信息") return data