功能: 接入鱼翅供应商直充渠道

2026-08-13 21:11:20 +08:00
parent 420ee3d95c
commit 902e669614
14 changed files with 586 additions and 15 deletions
@@ -0,0 +1,46 @@
+"""增加斗鱼鱼翅充值渠道配置
+
+Revision ID: 20260813_0026
+Revises: 20260813_0025
+Create Date: 2026-08-13
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "20260813_0026"
+down_revision: Union[str, None] = "20260813_0025"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _columns(bind) -> set[str]:
+    return {column["name"] for column in sa.inspect(bind).get_columns("douyu_config")}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if not sa.inspect(bind).has_table("douyu_config"):
+        return
+    columns = _columns(bind)
+    additions = (
+        ("gold_recharge_channel", sa.Column("gold_recharge_channel", sa.String(length=16), nullable=True, server_default="wechat_qr")),
+        ("gold_api_product_id", sa.Column("gold_api_product_id", sa.String(length=128), nullable=True, server_default="111570")),
+        ("gold_api_account_template_name", sa.Column("gold_api_account_template_name", sa.String(length=64), nullable=True, server_default="斗鱼UID")),
+    )
+    for name, column in additions:
+        if name not in columns:
+            op.add_column("douyu_config", column)
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if not sa.inspect(bind).has_table("douyu_config"):
+        return
+    columns = _columns(bind)
+    for name in ("gold_api_account_template_name", "gold_api_product_id", "gold_recharge_channel"):
+        if name in columns:
+            op.drop_column("douyu_config", name)
@@ -214,6 +214,9 @@ class DouyuConfig(Base):
    xpd_act_id = Column(String(64), default="46195")
    xpd_rid = Column(String(64), default="9263298")
    gold_pay_type = Column(Integer, default=1)
+    gold_recharge_channel = Column(String(16), default="wechat_qr")
+    gold_api_product_id = Column(String(128), default="111570")
+    gold_api_account_template_name = Column(String(64), default="斗鱼UID")
    gift_id = Column(String(64), default="23643")
    skin_id = Column(String(64), default="2942")
    updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow)
@@ -154,6 +154,9 @@ def _config_out(config: DouyuConfig) -> DouyuConfigOut:
        xpd_act_id=douyu_config_value("xpd_act_id", config.xpd_act_id),
        xpd_rid=douyu_config_value("xpd_rid", config.xpd_rid),
        gold_pay_type=douyu_config_value("gold_pay_type", config.gold_pay_type),
+        gold_recharge_channel=douyu_config_value("gold_recharge_channel", config.gold_recharge_channel),
+        gold_api_product_id=douyu_config_value("gold_api_product_id", config.gold_api_product_id),
+        gold_api_account_template_name=douyu_config_value("gold_api_account_template_name", config.gold_api_account_template_name),
        gift_id=douyu_config_value("gift_id", config.gift_id),
        skin_id=douyu_config_value("skin_id", config.skin_id),
        updated_at=config.updated_at,
@@ -585,6 +585,9 @@ class DouyuConfigOut(BaseModel):
    xpd_act_id: str = "46195"
    xpd_rid: str = "9263298"
    gold_pay_type: int = 1
+    gold_recharge_channel: str = "wechat_qr"
+    gold_api_product_id: str = "111570"
+    gold_api_account_template_name: str = "斗鱼UID"
    gift_id: str = "23643"
    skin_id: str = "2942"
    updated_at: Optional[datetime] = None
@@ -610,6 +613,9 @@ class DouyuConfigOut(BaseModel):
            "xpd_act_id": self.xpd_act_id,
            "xpd_rid": self.xpd_rid,
            "gold_pay_type": self.gold_pay_type,
+            "gold_recharge_channel": self.gold_recharge_channel,
+            "gold_api_product_id": self.gold_api_product_id,
+            "gold_api_account_template_name": self.gold_api_account_template_name,
            "gift_id": self.gift_id,
            "skin_id": self.skin_id,
            "updated_at": _ensure_tz(self.updated_at).isoformat() if self.updated_at else None,
@@ -635,6 +641,9 @@ class DouyuConfigUpdate(BaseModel):
    xpd_act_id: Optional[str] = None
    xpd_rid: Optional[str] = None
    gold_pay_type: Optional[int] = Field(None, ge=1, le=9)
+    gold_recharge_channel: Optional[str] = Field(None, pattern="^(wechat_qr|supplier_api)$")
+    gold_api_product_id: Optional[str] = Field(None, max_length=128)
+    gold_api_account_template_name: Optional[str] = Field(None, max_length=64)
    gift_id: Optional[str] = None
    skin_id: Optional[str] = None

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import threading
 import time
+from decimal import Decimal
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from typing import Optional
@@ -12,7 +13,13 @@ from typing import Optional
 from loguru import logger
 from sqlalchemy.orm import Session, joinedload

-from core.douyu import DouyuActivityClient, DouyuActivityError
+from core.douyu import (
+    DouyuActivityClient,
+    DouyuActivityError,
+    FishFinRechargeClient,
+    FishFinRechargeConfig,
+    FishFinRechargeError,
+)

 from ..database import SessionLocal
 from ..models import Account, DouyuEsportsGoodsSnapshot, DouyuGoodsSnapshot, DouyuTask, DouyuXpdGoodsSnapshot
@@ -719,6 +726,87 @@ class DouyuBatchRunner:
        result["gold_balance"] = last_gold
        return False

+    @staticmethod
+    def _supplier_value(payload: dict, *keys: str):
+        """兼容供应商将订单字段放在响应根节点、data 或 result 节点。"""
+        data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
+        result = payload.get("result") if isinstance(payload.get("result"), dict) else {}
+        for source in (payload, data, result):
+            for key in keys:
+                if source.get(key) is not None:
+                    return source[key]
+        return None
+
+    @classmethod
+    def _supplier_order_status(cls, payload: dict) -> int | None:
+        """提取供应商订单状态，文档约定 0-4。"""
+        return cls._to_int(cls._supplier_value(payload, "order_status", "orderStatus"))
+
+    @classmethod
+    def _supplier_message(cls, payload: dict) -> str:
+        """提取供应商可展示的业务消息。"""
+        value = cls._supplier_value(payload, "msg", "message", "error_msg")
+        return str(value or "")[:256]
+
+    @staticmethod
+    def _supplier_result(payload: dict) -> dict:
+        """保存必要订单状态，避免把完整供应商响应或签名暴露到任务结果。"""
+        data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
+        response_result = payload.get("result") if isinstance(payload.get("result"), dict) else {}
+        result = {
+            key: value
+            for key, value in {**payload, **data, **response_result}.items()
+            if key not in {"sign", "cards", "card_no", "card_pwd", "recharge_arg"}
+        }
+        return result
+
+    def _wait_supplier_gold_order(
+        self,
+        db: Session,
+        task: DouyuTask,
+        client: FishFinRechargeClient,
+        result: dict,
+    ) -> int | None:
+        """轮询供应商直充订单至结束状态。"""
+        order_no = str(result["customer_order_no"])
+        deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
+        poll_count = 0
+        result["payment_polling"] = True
+        while not self._stop.is_set() and time.monotonic() <= deadline:
+            try:
+                payload = client.query_order(customer_order_no=order_no)
+                code = self._to_int(self._supplier_value(payload, "code"))
+                status = self._supplier_order_status(payload)
+                poll_count += 1
+                result.update({
+                    "payment_poll_count": poll_count,
+                    "supplier_code": code,
+                    "supplier_order_status": status,
+                    "supplier_order": self._supplier_result(payload),
+                })
+                if code != 200:
+                    result["payment_polling"] = False
+                    return status if status in {2, 3, 4} else 4
+                if status in {2, 3, 4}:
+                    result["payment_polling"] = False
+                    return status
+                self._update_task_progress(
+                    db,
+                    task,
+                    "running",
+                    f"供应商直充订单处理中（状态 {status if status is not None else '-'}）",
+                    result,
+                )
+            except FishFinRechargeError as exc:
+                poll_count += 1
+                result["payment_poll_count"] = poll_count
+                result["payment_poll_error"] = str(exc)
+                self._update_task_progress(db, task, "running", f"查询供应商订单失败: {exc}", result)
+            if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
+                break
+        result["payment_polling"] = False
+        return None
+
    def _refresh_points_after_elite_gift(
        self,
        db: Session,
@@ -2358,6 +2446,13 @@ class DouyuBatchRunner:
    def _execute_create_gold_qr(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict):
        payload = self._task_payload(task)
        amount = int(payload.get("amount") or payload.get("gold_amount") or 1)
+        channel = str(config.get("gold_recharge_channel") or "wechat_qr")
+        if channel == "supplier_api":
+            try:
+                self._execute_create_gold_supplier_order(db, task, account, config, amount)
+            except FishFinRechargeError as exc:
+                self._mark_task(db, task, "failed", str(exc), {"recharge_channel": "supplier_api"})
+            return
        client = self._client(cookie)
        baseline_gold = account.gold_balance
        try:
@@ -2393,6 +2488,109 @@ class DouyuBatchRunner:
            result,
        )

+    def _execute_create_gold_supplier_order(
+        self,
+        db: Session,
+        task: DouyuTask,
+        account: Account,
+        config: dict,
+        amount: int,
+    ) -> None:
+        """创建供应商鱼翅直充订单并轮询订单状态。"""
+        product_id = str(config.get("gold_api_product_id") or "").strip()
+        template_name = str(config.get("gold_api_account_template_name") or "斗鱼UID").strip()
+        if not product_id:
+            raise FishFinRechargeError("请先在配置中填写供应商直充商品 ID")
+        charge_account = str(account.uid or "").strip()
+        if not charge_account:
+            raise FishFinRechargeError("账号缺少斗鱼 UID，无法发起供应商直充")
+
+        # task.id 是唯一且稳定的商户单号来源，重试时不会创建不同的供应商订单。
+        order_no = f"DYGF{task.id}"
+        # customer_price 是用户选择的充值面值；goodsFaceValue=0.993 是供货成本，不能作为支付金额。
+        customer_price = Decimal(amount)
+        def trace(event: dict) -> None:
+            """将脱敏供应商协议信息输出到任务日志，便于线上联调。"""
+            stage = event.get("stage")
+            if stage == "request":
+                params = event.get("params") or {}
+                self._push_log(
+                    "info",
+                    "供应商直充请求 "
+                    f"path={event.get('path')} uid={params.get('charge_account')} "
+                    f"buy_num={params.get('buy_num')} customer_price={params.get('customer_price')} "
+                    f"product_id={params.get('product_id')} types={event.get('parameter_types')} "
+                    f"sign_digest={event.get('sign_digest')}",
+                )
+                if event.get("json_body"):
+                    self._push_log(
+                        "info",
+                        "供应商直充完整调试请求 "
+                        f"url={event.get('url')} content_type={event.get('content_type')} "
+                        f"body={FishFinRechargeClient._json_text(event['json_body'])} "
+                        f"sign_params={FishFinRechargeClient._json_text(event.get('sign_params') or {})} "
+                        f"sign_source={event.get('sign_source')}",
+                    )
+            elif stage == "response":
+                self._push_log(
+                    "info",
+                    "供应商直充响应 "
+                    f"http={event.get('http_status')} code={event.get('code')} "
+                    f"message={event.get('message') or '-'} keys={event.get('keys')} "
+                    f"data_keys={event.get('data_keys')} result_keys={event.get('result_keys')}",
+                )
+                if event.get("response_body"):
+                    self._push_log(
+                        "info",
+                        "供应商直充完整调试响应 "
+                        f"headers={FishFinRechargeClient._json_text(event.get('response_headers') or {})} "
+                        f"body={FishFinRechargeClient._json_text(event['response_body'])} "
+                        f"raw={event.get('response_text')}",
+                    )
+
+        client = FishFinRechargeClient(FishFinRechargeConfig.from_env(), trace=trace)
+        order_payload = client.create_order(
+            charge_account=charge_account,
+            buy_num=amount,
+            customer_price=customer_price,
+            customer_order_no=order_no,
+            product_id=product_id,
+            recharge_arg=[{"templateName": template_name, "templateVal": charge_account}],
+        )
+        code = self._to_int(self._supplier_value(order_payload, "code"))
+        status = self._supplier_order_status(order_payload)
+        result = {
+            "recharge_channel": "supplier_api",
+            "customer_order_no": order_no,
+            "charge_account": charge_account,
+            "buy_num": amount,
+            "product_id": product_id,
+            "customer_price": format(customer_price.normalize(), "f"),
+            "supplier_code": code,
+            "supplier_order_status": status,
+            "supplier_order": self._supplier_result(order_payload),
+        }
+        if code != 200:
+            self._mark_task(db, task, "failed", self._supplier_message(order_payload) or "供应商创建直充订单失败", result)
+            return
+        account.bind_status = "gold_api_order_created"
+        account.updated_at = datetime.now(timezone.utc)
+        self._update_task_progress(db, task, "running", "供应商直充订单已创建，等待到账", result)
+        if status not in {2, 3, 4}:
+            status = self._wait_supplier_gold_order(db, task, client, result)
+        if self._stop.is_set():
+            self._mark_task(db, task, "stopped", "任务已停止", result)
+            return
+        if status == 2:
+            account.bind_status = "gold_recharged"
+            account.updated_at = datetime.now(timezone.utc)
+            self._mark_task(db, task, "success", "供应商直充成功", result)
+            return
+        if status in {3, 4}:
+            self._mark_task(db, task, "failed", "供应商直充失败", result)
+            return
+        self._mark_task(db, task, "failed", "供应商直充订单查询超时", result)
+
    def _execute_donate_elite_gift(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict):
        payload = self._task_payload(task)
        gift_count = int(payload.get("gift_count") or payload.get("count") or 1)
@@ -85,6 +85,9 @@ DOUYU_CONFIG_DEFAULTS = {    "manual_id": "G4KA4Qnz4LDp7",
    "esports_firework_gift_id": "24767",
    "esports_firework_skin_id": "3850",
    "gold_pay_type": 1,
+    "gold_recharge_channel": "wechat_qr",
+    "gold_api_product_id": "111570",
+    "gold_api_account_template_name": "斗鱼UID",
    "gift_id": "23643",
    "skin_id": "2942",
    "xpd_act_alias": "20260623KDQFH",
@@ -117,6 +120,8 @@ def apply_douyu_config_defaults(config: DouyuConfig) -> bool:
            changed = True
            continue
        normalized = douyu_config_value(field, getattr(config, field, None))
+        if field == "gold_recharge_channel" and normalized not in {"wechat_qr", "supplier_api"}:
+            normalized = DOUYU_CONFIG_DEFAULTS[field]
        if getattr(config, field, None) != normalized:
            setattr(config, field, normalized)
            changed = True