109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""
|
|
XorPay 支付渠道实现
|
|
文档: https://xorpay.com
|
|
"""
|
|
import hashlib
|
|
from typing import Any, Dict, Optional
|
|
|
|
import requests
|
|
|
|
from .base import PaymentProvider
|
|
|
|
|
|
class XorPayProvider(PaymentProvider):
|
|
"""XorPay 支付"""
|
|
|
|
def __init__(self, aid: str, secret: str, cashier_url: str = "https://xorpay.com/api/cashier"):
|
|
self.aid = aid
|
|
self.secret = secret
|
|
self.cashier_url = cashier_url.rstrip("/")
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "xorpay"
|
|
|
|
def _sign(self, name: str, pay_type: str, price: str, order_id: str, notify_url: str) -> str:
|
|
raw = name + pay_type + price + order_id + notify_url + self.secret
|
|
return hashlib.md5(raw.encode("utf-8")).hexdigest().lower()
|
|
|
|
def _query2_sign(self, order_id: str) -> str:
|
|
raw = order_id + self.secret
|
|
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]:
|
|
# XorPay 使用 jsapi 表示 H5/JSAPI 支付
|
|
channel_pay_type = "jsapi"
|
|
params = {
|
|
"name": name,
|
|
"pay_type": channel_pay_type,
|
|
"price": total_fee_yuan,
|
|
"order_id": order_id,
|
|
"notify_url": notify_url,
|
|
}
|
|
params["sign"] = self._sign(
|
|
params["name"],
|
|
params["pay_type"],
|
|
params["price"],
|
|
params["order_id"],
|
|
params["notify_url"],
|
|
)
|
|
return {
|
|
"status": "ok",
|
|
"provider": self.name,
|
|
"params": params,
|
|
"pay_url": f"{self.cashier_url}/{self.aid}",
|
|
}
|
|
|
|
def verify_notify(self, data: Dict[str, Any]) -> bool:
|
|
received_sign = data.get("sign", "")
|
|
aoid = data.get("aoid", "")
|
|
order_id = data.get("order_id", "")
|
|
pay_price = data.get("pay_price", "")
|
|
pay_time = data.get("pay_time", "")
|
|
raw = aoid + order_id + pay_price + pay_time + self.secret
|
|
calculated = hashlib.md5(raw.encode("utf-8")).hexdigest().lower()
|
|
return calculated == received_sign
|
|
|
|
def parse_notify(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
|
return {
|
|
"order_id": data.get("order_id", ""),
|
|
"trade_no": data.get("aoid", ""),
|
|
"pay_price": data.get("pay_price", ""),
|
|
"pay_time": data.get("pay_time", ""),
|
|
"trade_status": "success", # XorPay 验签通过即视为成功
|
|
}
|
|
|
|
def success_response(self) -> str:
|
|
return "ok"
|
|
|
|
def query_order_status(self, order_id: str, timeout: int = 10) -> Dict[str, Any]:
|
|
url = f"https://xorpay.com/api/query2/{self.aid}"
|
|
params = {
|
|
"order_id": order_id,
|
|
"sign": self._query2_sign(order_id),
|
|
}
|
|
try:
|
|
response = requests.get(url, params=params, timeout=timeout)
|
|
response.raise_for_status()
|
|
data = response.json() or {}
|
|
status = str(data.get("status", "")).strip().lower()
|
|
return {
|
|
"paid": status in ("payed", "success"),
|
|
"status": status,
|
|
"pay_price": data.get("pay_price") or data.get("price") or "",
|
|
"raw": data,
|
|
}
|
|
except Exception as error:
|
|
return {"paid": False, "error": str(error)}
|