125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
"""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
|