246 lines
8.5 KiB
Python
246 lines
8.5 KiB
Python
"""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(),
|
|
})
|