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.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)}' }