"""顽兔支付 Python SDK（单文件版）。 把本文件复制进你的项目即可使用，只依赖 Python 标准库（3.9+）。 接口说明见顽兔提供的接口文档；FastAPI 集成示例见 examples/fastapi_demo.py。 商户侧只有三件事：调接口、验 webhook、幂等入账。 """ from __future__ import annotations import hashlib import hmac import json import re import secrets import time import urllib.error import urllib.request from typing import Any, Callable, Mapping, Optional, Tuple, Union class WantuPayError(Exception): """SDK 错误基类。""" class WantuPayConfigError(WantuPayError): """本地入参或配置错误，请求未发出。""" class WantuPayHttpError(WantuPayError): """网络失败或响应异常。查询类接口可重试。""" def __init__(self, message: str, status: Optional[int] = None, body: Optional[str] = None): super().__init__(message) self.status = status self.body = body class WantuPayApiError(WantuPayError): """网关返回业务失败（code != 0），错误码见接口文档第 7 节。""" def __init__( self, code: int, message: str, http_status: int, channel_code: Optional[str] = None, channel_message: Optional[str] = None, ): super().__init__(f"顽兔支付网关业务失败 {code}: {message}") self.code = code self.http_status = http_status # 渠道原因仅用于排障展示，业务逻辑不要依赖 self.channel_code = channel_code self.channel_message = channel_message class WantuWebhookVerifyError(WantuPayError): """webhook 验签失败：不要处理业务，返回非 2xx 让网关重试。""" _OUT_NO_RE = re.compile(r"^[A-Za-z0-9_]{1,64}$") _AMOUNT_RE = re.compile(r"^(0|[1-9]\d*)(\.\d{1,2})?$") # 三行签名串：timestamp + "\n" + nonce + "\n" + 请求体原文，HMAC-SHA256，hex 小写 def _sign(secret: str, timestamp: str, nonce: str, body: str) -> str: message = f"{timestamp}\n{nonce}\n{body}" return hmac.new(secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).hexdigest() def _header(headers: Mapping[str, Any], name: str) -> Optional[str]: """大小写不敏感取请求头，兼容 FastAPI 的 Headers 与普通 dict。""" getter = getattr(headers, "get", None) if callable(getter): value = getter(name) or getter(name.lower()) if value: return str(value) for key in headers: if str(key).lower() == name.lower(): return str(headers[key]) return None # 可注入的传输函数（测试用）：(url, headers, body) -> (status, text) Transport = Callable[[str, Mapping[str, str], str], Tuple[int, str]] class WantuPayClient: def __init__( self, base_url: str, merchant_id: str, api_secret: str, webhook_secret: Optional[str] = None, timeout: float = 10.0, transport: Optional[Transport] = None, ): if not base_url or not base_url.strip(): raise WantuPayConfigError("base_url 不能为空") if not merchant_id or not api_secret: raise WantuPayConfigError("merchant_id 与 api_secret 不能为空") self.base_url = base_url.strip().rstrip("/") self.merchant_id = merchant_id.strip() self.api_secret = api_secret self.webhook_secret = webhook_secret self.timeout = timeout self._transport = transport or self._http_post # ---------- 四个接口 ---------- def create_order( self, out_trade_no: str, amount: str, subject: str, pay_mode: str, channel: Optional[str] = None, qrcode_width: Optional[int] = None, return_url: Optional[str] = None, expire_minutes: Optional[int] = None, attach: Optional[str] = None, ) -> dict: """创建支付单。同单号同参数幂等。 channel 选支付渠道：不传或 'alipay'＝支付宝（pay_url 是支付页地址，qrcode 模式 嵌 iframe、redirect 模式整页跳转）；'chinaums'＝银商，形态由 pay_mode 决定： 'qrcode'＝聚合码（一张码微信/支付宝/云闪付都能扫，pay_url 是二维码内容，前端要 渲染成码图、不能 iframe）；'redirect'＝公众号支付（微信/支付宝/云闪付 App 内打开 的 H5 页面把用户 302 到 pay_url，银商收银台唤起支付，商户无需自有公众号）。 return_url 是 redirect 模式支付完成后的回跳地址（http/https，≤255 字符）， 仅作页面回跳展示，入账一律以 webhook / 查单为准。 """ if not _OUT_NO_RE.match(out_trade_no): raise WantuPayConfigError("out_trade_no 需为 1-64 位字母/数字/下划线") if not isinstance(amount, str) or not _AMOUNT_RE.match(amount): raise WantuPayConfigError("amount 必须是字符串金额（元，最多两位小数），如 '100.00'") if pay_mode not in ("qrcode", "redirect"): raise WantuPayConfigError("pay_mode 只能是 'qrcode' 或 'redirect'") if channel is not None and channel not in ("alipay", "chinaums"): raise WantuPayConfigError("channel 只能是 'alipay'（支付宝）或 'chinaums'（银商）") if return_url is not None and pay_mode != "redirect": raise WantuPayConfigError("return_url 仅在 pay_mode='redirect' 下有效（qrcode 场景请用轮询）") return self._call( "/v1/orders", { "out_trade_no": out_trade_no, "amount": amount, "subject": subject, "channel": channel, "pay_mode": pay_mode, "qrcode_width": qrcode_width, "return_url": return_url, "expire_minutes": expire_minutes, "attach": attach, }, ) def query_order(self, out_trade_no: Optional[str] = None, order_id: Optional[str] = None) -> dict: """查询支付单（out_trade_no 与 order_id 二选一）。回调之外的兜底手段。""" if not out_trade_no and not order_id: raise WantuPayConfigError("out_trade_no 与 order_id 至少传一个") return self._call("/v1/orders/query", {"out_trade_no": out_trade_no, "order_id": order_id}) def create_refund( self, out_trade_no: str, out_refund_no: str, amount: str, reason: Optional[str] = None, ) -> dict: """申请退款。out_refund_no 是幂等键，网络失败后用同一单号重试是安全的。""" if not _OUT_NO_RE.match(out_refund_no): raise WantuPayConfigError("out_refund_no 需为 1-64 位字母/数字/下划线") if not isinstance(amount, str) or not _AMOUNT_RE.match(amount): raise WantuPayConfigError("amount 必须是字符串金额（元，最多两位小数）") return self._call( "/v1/refunds", { "out_trade_no": out_trade_no, "out_refund_no": out_refund_no, "amount": amount, "reason": reason, }, ) def query_refund(self, out_trade_no: str, out_refund_no: str) -> dict: """查询退款。""" return self._call( "/v1/refunds/query", {"out_trade_no": out_trade_no, "out_refund_no": out_refund_no} ) # ---------- webhook 验签 ---------- def verify_webhook( self, headers: Mapping[str, Any], raw_body: Union[bytes, str], tolerance_sec: int = 300, ) -> dict: """验签并解析 webhook 事件。 raw_body 必须是请求体原文（FastAPI 用 await request.body()）， 任何 JSON 解析要放在验签之后。验签失败抛 WantuWebhookVerifyError， 此时不要处理业务，返回非 2xx 状态即可，网关会自动重试。 返回事件 dict：event_type 为 'order.paid'、'refund.succeeded' 或 'refund.failed'（退款终态失败，资金未动），业务数据在 event['data'] （字段与接口文档一致）。 入账前务必：核对 data['amount'] 与本地订单一致 + 以 out_trade_no 幂等。 """ body_str = raw_body.decode("utf-8") if isinstance(raw_body, (bytes, bytearray)) else raw_body timestamp = _header(headers, "X-Wantu-Timestamp") nonce = _header(headers, "X-Wantu-Nonce") signature = _header(headers, "X-Wantu-Signature") if not timestamp or not nonce or not signature: raise WantuWebhookVerifyError("缺少 X-Wantu-Timestamp / Nonce / Signature 请求头") try: ts = int(timestamp) except ValueError: raise WantuWebhookVerifyError("时间戳格式不正确") if abs(int(time.time()) - ts) > tolerance_sec: raise WantuWebhookVerifyError("webhook 时间戳超窗") # 配置了独立 webhook_secret 用它，否则与请求签名共用 api_secret secret = self.webhook_secret or self.api_secret expected = _sign(secret, timestamp, nonce, body_str) if not hmac.compare_digest(expected, signature): raise WantuWebhookVerifyError("webhook 验签失败：报文不可信") try: event = json.loads(body_str) except ValueError: raise WantuWebhookVerifyError("webhook 请求体不是合法 JSON") if not isinstance(event, dict): raise WantuWebhookVerifyError("webhook 请求体不是 JSON 对象") return event # ---------- 内部 ---------- def _call(self, path: str, payload: dict) -> dict: body = json.dumps( {k: v for k, v in payload.items() if v is not None}, ensure_ascii=False, separators=(",", ":"), ) timestamp = str(int(time.time())) nonce = secrets.token_hex(12) headers = { "Content-Type": "application/json; charset=utf-8", "X-Wantu-Merchant-Id": self.merchant_id, "X-Wantu-Timestamp": timestamp, "X-Wantu-Nonce": nonce, "X-Wantu-Signature": _sign(self.api_secret, timestamp, nonce, body), } status, text = self._transport(self.base_url + path, headers, body) try: parsed = json.loads(text) except ValueError: raise WantuPayHttpError(f"网关响应不是 JSON（HTTP {status}）", status=status, body=text[:300]) code = parsed.get("code") if not isinstance(code, int): raise WantuPayHttpError("网关响应缺少 code 字段", status=status, body=text[:300]) if code != 0: raise WantuPayApiError( code=code, message=str(parsed.get("message", "未知错误")), http_status=status, channel_code=parsed.get("channel_code"), channel_message=parsed.get("channel_message"), ) data = parsed.get("data") return data if isinstance(data, dict) else {} def _http_post(self, url: str, headers: Mapping[str, str], body: str) -> Tuple[int, str]: request = urllib.request.Request(url, data=body.encode("utf-8"), headers=dict(headers), method="POST") try: with urllib.request.urlopen(request, timeout=self.timeout) as response: return response.status, response.read().decode("utf-8") except urllib.error.HTTPError as err: # 4xx/5xx 的响应体里是网关的错误外壳，照常读出来交给上层解析 return err.code, err.read().decode("utf-8") except urllib.error.URLError as err: raise WantuPayHttpError(f"请求顽兔支付网关失败: {err.reason}")