190 lines
7.7 KiB
Python
190 lines
7.7 KiB
Python
"""鱼翅直充供应商 API 客户端。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import Any, Mapping
|
|
|
|
import requests
|
|
|
|
|
|
class FishFinRechargeError(RuntimeError):
|
|
"""鱼翅直充供应商接口调用失败。"""
|
|
|
|
|
|
class FishFinRechargeConfigError(FishFinRechargeError):
|
|
"""鱼翅直充供应商配置不完整。"""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FishFinRechargeConfig:
|
|
"""供应商接入配置；敏感值仅从运行环境读取。"""
|
|
|
|
base_url: str
|
|
app_id: str
|
|
app_secret: str
|
|
app_key: str = ""
|
|
timeout: tuple[float, float] = (8, 20)
|
|
|
|
@classmethod
|
|
def from_env(cls) -> "FishFinRechargeConfig":
|
|
"""从环境变量读取配置，不在代码或数据库中保存商户密钥。"""
|
|
timeout = float(os.getenv("FISH_FIN_RECHARGE_TIMEOUT", "20"))
|
|
return cls(
|
|
base_url=os.getenv("FISH_FIN_RECHARGE_BASE_URL", "").strip(),
|
|
app_id=os.getenv("FISH_FIN_RECHARGE_APP_ID", "").strip(),
|
|
app_secret=os.getenv("FISH_FIN_RECHARGE_APP_SECRET", "").strip(),
|
|
app_key=os.getenv("FISH_FIN_RECHARGE_APP_KEY", "").strip(),
|
|
timeout=(8, timeout),
|
|
)
|
|
|
|
def validate(self) -> None:
|
|
"""在实际发起请求前校验必要配置。"""
|
|
missing = [
|
|
name
|
|
for name, value in (
|
|
("FISH_FIN_RECHARGE_BASE_URL", self.base_url),
|
|
("FISH_FIN_RECHARGE_APP_ID", self.app_id),
|
|
("FISH_FIN_RECHARGE_APP_SECRET", self.app_secret),
|
|
)
|
|
if not value
|
|
]
|
|
if missing:
|
|
raise FishFinRechargeConfigError(f"缺少鱼翅直充配置: {', '.join(missing)}")
|
|
|
|
|
|
class FishFinRechargeClient:
|
|
"""实现供应商 createOrderV2、queryOrderV2 与 userInfoV2 协议。"""
|
|
|
|
CREATE_ORDER_PATH = "/adapter-apiaccess/open/api/createOrderV2"
|
|
QUERY_ORDER_PATH = "/adapter-apiaccess/open/api/queryOrderV2"
|
|
USER_INFO_PATH = "/adapter-apiaccess/open/api/userInfoV2"
|
|
|
|
def __init__(self, config: FishFinRechargeConfig, *, session: requests.Session | None = None):
|
|
config.validate()
|
|
self.config = config
|
|
self.session = session or requests.Session()
|
|
self.session.trust_env = False
|
|
|
|
@staticmethod
|
|
def _sign_value(value: Any) -> str:
|
|
"""将参数转为待签名文本；对象按稳定紧凑 JSON 表示。"""
|
|
if isinstance(value, (dict, list, tuple)):
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
return str(value)
|
|
|
|
@classmethod
|
|
def normalized_params(cls, params: Mapping[str, Any]) -> dict[str, str]:
|
|
"""按供应商规则去除首尾空白与空值，保留可签名参数。"""
|
|
normalized: dict[str, str] = {}
|
|
for raw_key, raw_value in params.items():
|
|
key = str(raw_key).strip()
|
|
if not key or key == "sign" or raw_value is None:
|
|
continue
|
|
value = cls._sign_value(raw_value).strip()
|
|
if value:
|
|
normalized[key] = value
|
|
return normalized
|
|
|
|
def sign(self, params: Mapping[str, Any], method: str) -> str:
|
|
"""生成小写 MD5 签名。"""
|
|
normalized = self.normalized_params(params)
|
|
query = "&".join(f"{key}={normalized[key]}" for key in sorted(normalized))
|
|
raw = f"{query}{method.upper()}{self.config.app_secret}"
|
|
return hashlib.md5(raw.encode("utf-8")).hexdigest()
|
|
|
|
def verify_response_sign(self, payload: Mapping[str, Any], method: str) -> bool:
|
|
"""校验带 sign 的响应；供应商未返回 sign 时由调用方决定是否接受。"""
|
|
received = str(payload.get("sign") or "").strip().lower()
|
|
return bool(received) and received == self.sign(payload, method)
|
|
|
|
def _request(self, method: str, path: str, params: Mapping[str, Any] | None = None) -> dict[str, Any]:
|
|
"""补齐公共参数、签名并执行一次 JSON 请求。"""
|
|
request_params: dict[str, Any] = {
|
|
"app_id": self.config.app_id,
|
|
"timestamp": int(time.time()),
|
|
**(params or {}),
|
|
}
|
|
request_params["sign"] = self.sign(request_params, method)
|
|
url = f"{self.config.base_url.rstrip('/')}{path}"
|
|
try:
|
|
if method.upper() == "GET":
|
|
response = self.session.get(url, params=request_params, timeout=self.config.timeout)
|
|
else:
|
|
response = self.session.post(url, json=request_params, timeout=self.config.timeout)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
except requests.RequestException as exc:
|
|
raise FishFinRechargeError(f"供应商请求失败: {exc}") from exc
|
|
except ValueError as exc:
|
|
raise FishFinRechargeError("供应商响应不是 JSON") from exc
|
|
if not isinstance(payload, dict):
|
|
raise FishFinRechargeError("供应商响应格式无效")
|
|
return payload
|
|
|
|
@staticmethod
|
|
def _price(value: Decimal | int | float | str) -> str:
|
|
"""规范化金额，避免浮点数表达式进入签名或订单请求。"""
|
|
try:
|
|
price = Decimal(str(value))
|
|
except (InvalidOperation, ValueError) as exc:
|
|
raise ValueError("customer_price 必须是有效金额") from exc
|
|
if not price.is_finite() or price <= 0:
|
|
raise ValueError("customer_price 必须大于 0")
|
|
return format(price.normalize(), "f")
|
|
|
|
def create_order(
|
|
self,
|
|
*,
|
|
charge_account: str,
|
|
buy_num: int,
|
|
customer_price: Decimal | int | float | str,
|
|
customer_order_no: str,
|
|
product_id: str,
|
|
recharge_arg: list[dict[str, Any]],
|
|
notify_url: str = "",
|
|
ext_arg: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""创建直充订单；成功后应使用 query_order 或供应商回调确认最终状态。"""
|
|
if not str(charge_account).strip():
|
|
raise ValueError("charge_account 不能为空")
|
|
if not str(customer_order_no).strip():
|
|
raise ValueError("customer_order_no 不能为空")
|
|
if not str(product_id).strip():
|
|
raise ValueError("product_id 不能为空")
|
|
if not isinstance(buy_num, int) or isinstance(buy_num, bool) or buy_num < 1:
|
|
raise ValueError("buy_num 必须是不小于 1 的整数")
|
|
if not isinstance(recharge_arg, list) or not recharge_arg:
|
|
raise ValueError("recharge_arg 必须是非空数组")
|
|
return self._request("POST", self.CREATE_ORDER_PATH, {
|
|
"charge_account": str(charge_account).strip(),
|
|
"buy_num": buy_num,
|
|
"customer_price": self._price(customer_price),
|
|
"customer_order_no": str(customer_order_no).strip(),
|
|
"product_id": str(product_id).strip(),
|
|
"recharge_arg": recharge_arg,
|
|
"notify_url": str(notify_url).strip(),
|
|
"ext_arg": ext_arg or {},
|
|
})
|
|
|
|
def query_order(self, **query: Any) -> dict[str, Any]:
|
|
"""查询订单。
|
|
|
|
文档未列出查询字段，调用方必须显式传入供应商确认的订单标识，例如
|
|
``customer_order_no`` 或供应商订单号。
|
|
"""
|
|
if not query or not any(str(value).strip() for value in query.values() if value is not None):
|
|
raise ValueError("查询订单至少需要一个非空订单标识")
|
|
return self._request("GET", self.QUERY_ORDER_PATH, query)
|
|
|
|
def account_info(self) -> dict[str, Any]:
|
|
"""查询商户账户信息。"""
|
|
return self._request("GET", self.USER_INFO_PATH)
|