From 38fd809ca4e3abd2baab3c2bb930f943dfb02fe9 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 30 Aug 2026 19:58:43 +0800 Subject: [PATCH] =?UTF-8?q?type:=20=E6=94=B6=E7=AA=84=E6=96=97=E9=B1=BC?= =?UTF-8?q?=E9=B1=BC=E7=BF=85=E5=85=85=E5=80=BC=E6=89=A7=E8=A1=8C=E5=99=A8?= =?UTF-8?q?=E7=B1=BB=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- web/backend/services/douyu_runner_gold.py | 147 +++++++++++++++++----- 1 file changed, 115 insertions(+), 32 deletions(-) diff --git a/web/backend/services/douyu_runner_gold.py b/web/backend/services/douyu_runner_gold.py index f232d4e..c421f3e 100644 --- a/web/backend/services/douyu_runner_gold.py +++ b/web/backend/services/douyu_runner_gold.py @@ -5,16 +5,41 @@ import re import time from decimal import Decimal from datetime import datetime, timezone +from typing import Any, TYPE_CHECKING, cast from sqlalchemy.orm import Session -from core.douyu import DouyuActivityClient, FishFinRechargeClient, FishFinRechargeConfig, FishFinRechargeError +from core.douyu import ( + DouyuActivityClient, + FishFinRechargeClient, + FishFinRechargeConfig, + FishFinRechargeError, +) from ..models import Account, DouyuTask from .douyu_service import update_account_profile_from_cookie from .douyu_runner_core import DOUYU_PAYMENT_POLL_INTERVAL, DOUYU_PAYMENT_POLL_SECONDS +if TYPE_CHECKING: + from .douyu_runner import DouyuBatchRunner + + class GoldMixin: """鱼翅充值域：扫码充值、供应商直充与到账轮询。""" - def _refresh_account_gold_balance(self, client: DouyuActivityClient, account: Account) -> dict: + + if TYPE_CHECKING: + _stop: Any + + @staticmethod + def _to_int(value: Any) -> int | None: ... + + def _client(self, cookie: str) -> DouyuActivityClient: ... + def _task_payload(self, task: DouyuTask) -> dict: ... + def _push_log(self, level: str, message: str) -> None: ... + def _mark_task(self, *args: Any, **kwargs: Any) -> None: ... + def _update_task_progress(self, *args: Any, **kwargs: Any) -> None: ... + + def _refresh_account_gold_balance( + self, client: DouyuActivityClient, account: Account + ) -> dict: """刷新鱼翅和钱包兑换余额并写回账号表。""" gold = client.gold_account() exchange = client.exchange_balance() @@ -67,7 +92,9 @@ class GoldMixin: if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL): break continue - changed = last_gold is not None and (baseline_gold is None or last_gold != baseline_gold) + changed = last_gold is not None and ( + baseline_gold is None or last_gold != baseline_gold + ) if changed: result["payment_polling"] = False result["gold_recharged"] = True @@ -83,7 +110,9 @@ class GoldMixin: 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) + self._update_task_progress( + db, task, "running", f"等待鱼翅到账: {exc}", result + ) if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL): break result["payment_polling"] = False @@ -94,8 +123,12 @@ class GoldMixin: @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 {} + data_raw = payload.get("data") + data = cast(dict[str, Any], data_raw) if isinstance(data_raw, dict) else {} + result_raw = payload.get("result") + result = ( + cast(dict[str, Any], result_raw) if isinstance(result_raw, dict) else {} + ) for source in (payload, data, result): for key in keys: if source.get(key) is not None: @@ -105,7 +138,11 @@ class GoldMixin: @classmethod def _supplier_order_status(cls, payload: dict) -> int | None: """提取供应商订单状态，文档约定 0-4。""" - return cls._to_int(cls._supplier_value(payload, "order_status", "orderStatus", "supplier_order_status")) + return cls._to_int( + cls._supplier_value( + payload, "order_status", "orderStatus", "supplier_order_status" + ) + ) @classmethod def _supplier_message(cls, payload: dict) -> str: @@ -116,8 +153,12 @@ class GoldMixin: @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 {} + data_raw = payload.get("data") + data = cast(dict[str, Any], data_raw) if isinstance(data_raw, dict) else {} + result_raw = payload.get("result") + response_result = ( + cast(dict[str, Any], result_raw) if isinstance(result_raw, dict) else {} + ) result = { key: value for key, value in {**payload, **data, **response_result}.items() @@ -131,7 +172,10 @@ class GoldMixin: existing = str(task.supplier_out_order_id or "").strip() if existing: return existing - batch_token = re.sub(r"[^A-Za-z0-9]", "", str(task.batch_id or "")).upper()[:16] or "LOCAL" + batch_token = ( + re.sub(r"[^A-Za-z0-9]", "", str(task.batch_id or "")).upper()[:16] + or "LOCAL" + ) return f"DYGF{batch_token}T{task.id}" def _wait_supplier_gold_order( @@ -151,7 +195,9 @@ class GoldMixin: # 回调可能已在另一个数据库会话中结束订单，刷新后直接使用其结果。 db.refresh(task) if task.status in {"success", "failed"}: - callback_result = task.result if isinstance(task.result, dict) else result + callback_result = ( + task.result if isinstance(task.result, dict) else result + ) result.update(callback_result) result["payment_polling"] = False return self._supplier_order_status(callback_result) @@ -159,12 +205,14 @@ class GoldMixin: 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), - }) + 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 @@ -182,21 +230,29 @@ class GoldMixin: 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) + 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 _execute_create_gold_qr(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + 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, cookie, config, amount) + self._execute_create_gold_supplier_order( + db, task, account, cookie, config, amount + ) except FishFinRechargeError as exc: - self._mark_task(db, task, "failed", str(exc), {"recharge_channel": "supplier_api"}) + self._mark_task( + db, task, "failed", str(exc), {"recharge_channel": "supplier_api"} + ) return client = self._client(cookie) baseline_gold = account.gold_balance @@ -206,11 +262,17 @@ class GoldMixin: db.commit() except Exception as exc: self._push_log("warning", f"生成鱼翅码前刷新余额失败: {exc}") - result = client.create_gold_qr(amount=amount, pay_type=int(config["gold_pay_type"])) + result = client.create_gold_qr( + amount=amount, pay_type=int(config["gold_pay_type"]) + ) account.bind_status = "gold_qr_created" account.updated_at = datetime.now(timezone.utc) - self._update_task_progress(db, task, "running", f"鱼翅 {amount} 元支付码已生成，等待到账", result) - recharged = self._wait_gold_balance_after_payment(db, task, account, client, result, baseline_gold) + self._update_task_progress( + db, task, "running", f"鱼翅 {amount} 元支付码已生成，等待到账", result + ) + recharged = self._wait_gold_balance_after_payment( + db, task, account, client, result, baseline_gold + ) if self._stop.is_set(): self._mark_task(db, task, "stopped", "任务已停止", result) return @@ -244,7 +306,9 @@ class GoldMixin: ) -> None: """创建供应商鱼翅直充订单并轮询订单状态。""" product_id = str(config.get("gold_api_product_id") or "").strip() - template_name = str(config.get("gold_api_account_template_name") or "斗鱼昵称").strip() + template_name = str( + config.get("gold_api_account_template_name") or "斗鱼昵称" + ).strip() if not product_id: raise FishFinRechargeError("请先在配置中填写供应商直充商品 ID") # 充值商品按斗鱼昵称识别账号，UID 只能作为审计信息，不能作为充值值。 @@ -283,8 +347,16 @@ class GoldMixin: ) elif stage == "response": status = self._to_int(event.get("order_status")) - status_labels = {0: "待处理", 1: "处理中", 2: "成功", 3: "失败", 4: "异常"} - status_text = status_labels.get(status, "-") + status_labels = { + 0: "待处理", + 1: "处理中", + 2: "成功", + 3: "失败", + 4: "异常", + } + status_text = status_labels.get( + status if status is not None else -1, "-" + ) reason = str(event.get("fail_reason") or event.get("message") or "-") self._push_log( "info", @@ -310,7 +382,9 @@ class GoldMixin: pay_amount=pay_amount, out_order_id=order_no, product_id=product_id, - recharge_arg=[{"templateName": template_name, "templateVal": recharge_account}], + recharge_arg=[ + {"templateName": template_name, "templateVal": recharge_account} + ], order_type=0, notify_url=client.config.notify_url, ) @@ -331,11 +405,19 @@ class GoldMixin: "supplier_order": self._supplier_result(order_payload), } if code != 200: - self._mark_task(db, task, "failed", self._supplier_message(order_payload) or "供应商创建直充订单失败", result) + 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) + 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(): @@ -351,7 +433,9 @@ class GoldMixin: return self._mark_task(db, task, "failed", "供应商直充订单查询超时", result) - def _execute_query_gold_balance(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): + def _execute_query_gold_balance( + self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict + ): client = self._client(cookie) result = self._refresh_account_gold_balance(client, account) account.bind_status = "gold_balance_queried" @@ -363,4 +447,3 @@ class GoldMixin: f"鱼翅余额: {account.gold_balance if account.gold_balance is not None else '-'}", result, ) -