110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
import hashlib
|
||
import requests
|
||
from typing import Dict, Any, Optional
|
||
|
||
class Upay:
|
||
def __init__(self):
|
||
# self.app_id="4NgfCm1e"
|
||
# self.app_secret="4gDTZDXfpKQBboT6"
|
||
# self.base_url = "https://api-test.upay.ink"
|
||
|
||
self.app_id="E7c4dss9"
|
||
self.app_secret="Hwc56INsabRau2yn"
|
||
self.base_url = "https://api.upay.ink"
|
||
|
||
def _generate_signature(self, params: Dict[str, Any]) -> str:
|
||
"""
|
||
生成签名
|
||
根据 UPay API 签名算法生成签名
|
||
|
||
步骤：
|
||
1. 将需要加签的参数按照参数名 ASCII 码从小到大排序（字典序）
|
||
2. 使用 URL 键值对的格式拼接成字符串 stringA
|
||
3. 在 stringA 最后拼接上 &appSecret=密钥 得到 stringSignTemp
|
||
4. 对 stringSignTemp 进行 MD5 运算，再转换为大写
|
||
"""
|
||
# 过滤空值和签名字段（signature 参数不参与签名）
|
||
filtered_params = {k: v for k, v in params.items() if v is not None and k != 'signature'}
|
||
|
||
# 按键名升序排序（字典序）
|
||
sorted_params = sorted(filtered_params.items())
|
||
|
||
# 拼接字符串 stringA
|
||
string_a = '&'.join([f"{k}={v}" for k, v in sorted_params])
|
||
|
||
# 拼接 appSecret 得到 stringSignTemp
|
||
string_sign_temp = f"{string_a}&appSecret={self.app_secret}"
|
||
|
||
# MD5 运算并转为大写
|
||
signature = hashlib.md5(string_sign_temp.encode('utf-8')).hexdigest().upper()
|
||
|
||
return signature
|
||
|
||
def create_order(self,
|
||
merchant_order_no: str,
|
||
chain_type: str,
|
||
fiat_amount: str,
|
||
fiat_currency: str,
|
||
notify_url: str,
|
||
attach: Optional[str] = None,
|
||
product_name: Optional[str] = None,
|
||
redirect_url: Optional[str] = None) -> Dict[str, Any]:
|
||
"""
|
||
创建订单
|
||
|
||
Args:
|
||
merchant_order_no: 商户端自主生成的订单号，在商户端要保证唯一性
|
||
chain_type: 链路类型（1: 波场TRC20, 2: 以太坊ERC20, 3: PayPal PYUSD）
|
||
fiat_amount: 法币金额，精确到小数点后4位
|
||
fiat_currency: 法币类型（USD, CNY, INR, JPY, KRW, PHP, EUR, GBP, CHF, TWD, HKD, MOP, SGD, NZD, THB, CAD, ZAR, BRL）
|
||
notify_url: 接收异步通知的回调地址
|
||
attach: 用户自定义数据（可选）
|
||
product_name: 商品名称（可选）
|
||
redirect_url: 支付成功后的前端重定向地址（可选）
|
||
|
||
Returns:
|
||
Dict: API 响应数据
|
||
"""
|
||
# 构建请求参数
|
||
sign_params = {
|
||
'appId': self.app_id,
|
||
'merchantOrderNo': merchant_order_no,
|
||
'chainType': chain_type,
|
||
'fiatAmount': fiat_amount,
|
||
'fiatCurrency': fiat_currency,
|
||
'notifyUrl': notify_url,
|
||
}
|
||
|
||
params = sign_params.copy()
|
||
|
||
# 添加可选参数
|
||
if attach is not None:
|
||
params['attach'] = attach
|
||
if product_name is not None:
|
||
params['productName'] = product_name
|
||
if redirect_url is not None:
|
||
params['redirectUrl'] = redirect_url
|
||
|
||
print("请求参数:", params)
|
||
|
||
# 生成签名
|
||
signature = self._generate_signature(sign_params)
|
||
params['signature'] = signature
|
||
|
||
# 发送请求
|
||
url = f"{self.base_url}/v1/api/open/order/apply"
|
||
|
||
try:
|
||
response = requests.post(url, json=params)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
except requests.exceptions.RequestException as e:
|
||
return {
|
||
'success': False,
|
||
'error': f'请求失败: {str(e)}'
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
'success': False,
|
||
'error': f'处理异常: {str(e)}'
|
||
} |