"""WeChat mini-program virtual-payment signing and server API adapter. The AppKey and session_key never leave the backend. The JSON string returned as ``signData`` is exactly the string used for both HMAC signatures. """ from __future__ import annotations import hashlib import hmac import json import os import time import xml.etree.ElementTree as ET from typing import Any import httpx REQUEST_VIRTUAL_PAYMENT_URI = "requestVirtualPayment" PAYMENT_EVENTS = {"xpay_goods_deliver_notify"} REFUND_EVENTS = {"xpay_refund_notify"} class WechatVirtualPaymentError(RuntimeError): pass def json_compact(payload: dict[str, Any]) -> str: return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) def hmac_sha256_hex(key: str, message: str) -> str: return hmac.new(key.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).hexdigest() def virtual_env() -> int: value = os.getenv("WECHAT_VIRTUAL_ENV", "sandbox").strip().lower() return 0 if value in {"0", "prod", "production", "live", "online"} else 1 def _app_key(env: int) -> str: name = "WECHAT_VIRTUAL_APP_KEY" if env == 0 else "WECHAT_VIRTUAL_SANDBOX_APP_KEY" return os.getenv(name, "").strip() def _offer_id() -> str: return os.getenv("WECHAT_VIRTUAL_OFFER_ID", "").strip() def product_id_for_plan(plan) -> str: configured = str(getattr(plan, "virtual_product_id", "") or "").strip() if not configured: configured = os.getenv(f"WECHAT_VIRTUAL_PRODUCT_{plan.id}", "").strip() if not configured: raise WechatVirtualPaymentError(f"套餐 {plan.id} 尚未配置微信虚拟支付商品 ID") if len(configured) > 64 or not all(ch.isalnum() or ch in "_-" for ch in configured): raise WechatVirtualPaymentError("微信虚拟支付商品 ID 格式不正确") return configured def build_payment_params(*, order, plan, session_key: str) -> dict[str, Any]: env = virtual_env() offer_id = _offer_id() app_key = _app_key(env) if not offer_id or not app_key or not session_key: raise WechatVirtualPaymentError("微信小程序虚拟支付配置不完整") sign_data = json_compact({ "offerId": offer_id, "buyQuantity": 1, "env": env, "currencyType": "CNY", "productId": product_id_for_plan(plan), "goodsPrice": int(order.price_cents), "outTradeNo": order.order_no, "attach": json_compact({"orderNo": order.order_no, "planId": order.plan_id}), }) return { "provider": "wechat_virtual", "payment_channel": "virtual", "payment_method": "wechat", "mode": "short_series_goods", "signData": sign_data, "paySig": hmac_sha256_hex(app_key, f"{REQUEST_VIRTUAL_PAYMENT_URI}&{sign_data}"), "signature": hmac_sha256_hex(session_key, sign_data), "env": env, "offerId": offer_id, "outTradeNo": order.order_no, } def exchange_code(code: str) -> dict[str, str]: app_id = os.getenv("WECHAT_MP_APP_ID", "").strip() app_secret = os.getenv("WECHAT_MP_APP_SECRET", "").strip() if not app_id or not app_secret: raise WechatVirtualPaymentError("微信小程序登录配置不完整") try: response = httpx.get( "https://api.weixin.qq.com/sns/jscode2session", params={ "appid": app_id, "secret": app_secret, "js_code": code, "grant_type": "authorization_code", }, timeout=15, ) data = response.json() except (httpx.HTTPError, ValueError) as exc: raise WechatVirtualPaymentError("微信登录态交换失败，请稍后重试") from exc if response.status_code >= 400 or data.get("errcode"): raise WechatVirtualPaymentError(data.get("errmsg") or "微信登录态交换失败") openid = str(data.get("openid") or "").strip() session_key = str(data.get("session_key") or "").strip() if not openid or not session_key: raise WechatVirtualPaymentError("微信未返回完整登录态") return {"openid": openid, "session_key": session_key} def verify_callback_signature(signature: str, timestamp: str, nonce: str) -> bool: token = os.getenv("WECHAT_VIRTUAL_CALLBACK_TOKEN", "").strip() if not token or not signature or not timestamp or not nonce: return False source = "".join(sorted([token, timestamp, nonce])) expected = hashlib.sha1(source.encode("utf-8")).hexdigest() return hmac.compare_digest(signature, expected) def _xml_value(element: ET.Element) -> Any: children = list(element) if not children: return element.text or "" return {child.tag: _xml_value(child) for child in children} def parse_callback_body(body: bytes) -> dict[str, Any]: text = body.decode("utf-8", errors="replace").strip() if not text: return {} try: payload = json.loads(text) if isinstance(payload, dict): return payload except json.JSONDecodeError: pass try: parsed = _xml_value(ET.fromstring(text)) except ET.ParseError as exc: raise WechatVirtualPaymentError("微信虚拟支付回调格式不正确") from exc return parsed if isinstance(parsed, dict) else {} def case_get(payload: Any, key: str) -> Any: if not isinstance(payload, dict): return None lowered = key.lower() for current, value in payload.items(): if str(current).lower() == lowered: return value return None def callback_value(payload: dict[str, Any], *path: str) -> Any: current: Any = payload for key in path: current = case_get(current, key) if current is None: break return current _access_token_cache: tuple[str, float] = ("", 0) def _access_token() -> str: global _access_token_cache token, expires_at = _access_token_cache if token and expires_at > time.monotonic() + 60: return token app_id = os.getenv("WECHAT_MP_APP_ID", "").strip() app_secret = os.getenv("WECHAT_MP_APP_SECRET", "").strip() if not app_id or not app_secret: raise WechatVirtualPaymentError("微信小程序服务端配置不完整") try: response = httpx.get( "https://api.weixin.qq.com/cgi-bin/token", params={"grant_type": "client_credential", "appid": app_id, "secret": app_secret}, timeout=15, ) data = response.json() except (httpx.HTTPError, ValueError) as exc: raise WechatVirtualPaymentError("微信 access_token 获取失败") from exc if response.status_code >= 400 or data.get("errcode"): raise WechatVirtualPaymentError(data.get("errmsg") or "微信 access_token 获取失败") token = str(data.get("access_token") or "") if not token: raise WechatVirtualPaymentError("微信未返回 access_token") _access_token_cache = (token, time.monotonic() + int(data.get("expires_in") or 7200)) return token def call_xpay(uri: str, payload: dict[str, Any]) -> dict[str, Any]: env = int(payload.get("env", virtual_env())) app_key = _app_key(env) if not app_key: raise WechatVirtualPaymentError("微信虚拟支付 AppKey 未配置") body = json_compact(payload) pay_sig = hmac_sha256_hex(app_key, f"{uri}&{body}") try: response = httpx.post( f"https://api.weixin.qq.com{uri}", params={"access_token": _access_token(), "pay_sig": pay_sig}, content=body.encode("utf-8"), headers={"Content-Type": "application/json"}, timeout=20, ) data = response.json() except (httpx.HTTPError, ValueError) as exc: raise WechatVirtualPaymentError("微信虚拟支付服务暂时不可用") from exc if response.status_code >= 400 or data.get("errcode") not in (None, 0): raise WechatVirtualPaymentError(data.get("errmsg") or "微信虚拟支付请求失败") return data def request_refund(*, openid: str, order_no: str, refund_no: str, amount_cents: int, reason: int = 3) -> dict[str, Any]: return call_xpay("/xpay/refund_order", { "openid": openid, "order_id": order_no, "refund_order_id": refund_no, "left_fee": amount_cents, "refund_fee": amount_cents, "biz_meta": json_compact({"orderNo": order_no}), "refund_reason": int(reason), "req_from": 1, "env": virtual_env(), }) def query_order(*, openid: str, order_no: str) -> dict[str, Any]: return call_xpay("/xpay/query_order", { "openid": openid, "order_id": order_no, "env": virtual_env(), })