154 lines
5.1 KiB
Python
154 lines
5.1 KiB
Python
"""
|
||
ZPAY 支付渠道实现
|
||
文档: https://z-pay.cn/doc.html
|
||
支持：页面跳转支付、API 接口支付、异步通知验签
|
||
"""
|
||
import hashlib
|
||
from typing import Any, Dict, Optional
|
||
|
||
import requests
|
||
|
||
from .base import PaymentProvider
|
||
|
||
|
||
class ZPayProvider(PaymentProvider):
|
||
"""ZPAY 支付（易支付兼容接口）"""
|
||
|
||
# API 地址
|
||
SUBMIT_URL = "https://zpayz.cn/submit.php" # 页面跳转
|
||
MAPI_URL = "https://zpayz.cn/mapi.php" # API 接口
|
||
|
||
def __init__(self, pid: str, key: str):
|
||
self.pid = pid
|
||
self.key = key
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "zpay"
|
||
|
||
def _md5_sign(self, params: Dict[str, Any], exclude: Optional[set] = None) -> str:
|
||
"""
|
||
ZPAY MD5 签名算法：
|
||
1. 参数按 ASCII 排序，sign/sign_type/空值不参与
|
||
2. 拼接为 a=b&c=d
|
||
3. sign = md5(拼接串 + KEY)，小写
|
||
"""
|
||
exclude = exclude or {"sign", "sign_type"}
|
||
filtered = {k: v for k, v in params.items() if k not in exclude and v is not None and v != ""}
|
||
sorted_keys = sorted(filtered.keys())
|
||
parts = [f"{k}={filtered[k]}" for k in sorted_keys]
|
||
raw = "&".join(parts) + self.key
|
||
return hashlib.md5(raw.encode("utf-8")).hexdigest().lower()
|
||
|
||
def create_order(
|
||
self,
|
||
*,
|
||
order_id: str,
|
||
name: str,
|
||
total_fee_yuan: str,
|
||
pay_type: str,
|
||
notify_url: str,
|
||
return_url: Optional[str] = None,
|
||
client_ip: Optional[str] = None,
|
||
extra: Optional[Dict[str, Any]] = None,
|
||
) -> Dict[str, Any]:
|
||
# ZPAY: alipay / wxpay（pay_type 可为业务类型或支付方式，此处按支付方式解析）
|
||
channel_type = "wxpay" if str(pay_type).lower() in ("wx", "wechat", "wxpay") else "alipay"
|
||
params = {
|
||
"pid": self.pid,
|
||
"type": channel_type,
|
||
"out_trade_no": order_id,
|
||
"notify_url": notify_url,
|
||
"name": name,
|
||
"money": total_fee_yuan,
|
||
"return_url": return_url or notify_url,
|
||
}
|
||
if client_ip:
|
||
params["clientip"] = client_ip
|
||
if extra and extra.get("param"):
|
||
params["param"] = str(extra["param"])
|
||
params["sign_type"] = "MD5"
|
||
params["sign"] = self._md5_sign(params)
|
||
|
||
# 页面跳转方式：返回 form 提交 URL 和参数
|
||
return {
|
||
"status": "ok",
|
||
"provider": self.name,
|
||
"params": params,
|
||
"pay_url": self.SUBMIT_URL,
|
||
"submit_method": "POST",
|
||
}
|
||
|
||
def create_order_api(
|
||
self,
|
||
*,
|
||
order_id: str,
|
||
name: str,
|
||
total_fee_yuan: str,
|
||
pay_type: str,
|
||
notify_url: str,
|
||
return_url: Optional[str] = None,
|
||
client_ip: str = "127.0.0.1",
|
||
device: str = "pc",
|
||
extra: Optional[Dict[str, Any]] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
ZPAY API 接口支付（返回 payurl/qrcode 等）
|
||
适用于需要二维码或 H5 跳转的场景
|
||
"""
|
||
channel_type = "wxpay" if pay_type.lower() in ("wx", "wechat", "wxpay") else "alipay"
|
||
params = {
|
||
"pid": self.pid,
|
||
"type": channel_type,
|
||
"out_trade_no": order_id,
|
||
"notify_url": notify_url,
|
||
"name": name,
|
||
"money": total_fee_yuan,
|
||
"clientip": client_ip or "127.0.0.1",
|
||
"device": device,
|
||
}
|
||
if return_url:
|
||
params["return_url"] = return_url
|
||
if extra and extra.get("param"):
|
||
params["param"] = str(extra["param"])
|
||
params["sign_type"] = "MD5"
|
||
params["sign"] = self._md5_sign(params)
|
||
|
||
resp = requests.post(self.MAPI_URL, data=params, timeout=15)
|
||
result = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
|
||
if isinstance(result, dict) and result.get("code") == 1:
|
||
return {
|
||
"status": "ok",
|
||
"provider": self.name,
|
||
"payurl": result.get("payurl"),
|
||
"payurl2": result.get("payurl2"), # 微信 H5
|
||
"qrcode": result.get("qrcode"),
|
||
"img": result.get("img"),
|
||
"O_id": result.get("O_id"),
|
||
"trade_no": result.get("trade_no"),
|
||
}
|
||
return {
|
||
"status": "error",
|
||
"provider": self.name,
|
||
"msg": result.get("msg", resp.text),
|
||
}
|
||
|
||
def verify_notify(self, data: Dict[str, Any]) -> bool:
|
||
received = data.get("sign", "")
|
||
calculated = self._md5_sign(data)
|
||
return calculated == received
|
||
|
||
def parse_notify(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
# ZPAY 通知：trade_status 为 TRADE_SUCCESS 表示成功
|
||
status = "success" if data.get("trade_status") == "TRADE_SUCCESS" else "pending"
|
||
return {
|
||
"order_id": data.get("out_trade_no", ""),
|
||
"trade_no": data.get("trade_no", ""),
|
||
"pay_price": data.get("money", ""),
|
||
"pay_time": "", # ZPAY 通知未明确返回
|
||
"trade_status": status,
|
||
}
|
||
|
||
def success_response(self) -> str:
|
||
return "success"
|