type: 收窄斗鱼任务路由类型

web/backend/routers/douyu.py

@@ -5,15 +5,35 @@ from __future__ import annotations
 import asyncio
 import threading
 from datetime import datetime, timezone
+from typing import Any, cast
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect
+from fastapi import (
+    APIRouter,
+    Depends,
+    HTTPException,
+    Query,
+    Request,
+    WebSocket,
+    WebSocketDisconnect,
+)
 from loguru import logger
-from sqlalchemy import func, or_
+from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session, joinedload, load_only
 
 from ..database import SessionLocal, get_db
 from ..deps import authenticate_websocket, get_current_user, require_permission
-from ..models import Account, DouyuConfig, DouyuEsportsGoodsSnapshot, DouyuGoodsSnapshot, DouyuTask, DouyuWorkbench, DouyuWorkbenchAccount, DouyuXpdGoodsSnapshot, User
+from ..models import (
+    Account,
+    DouyuConfig,
+    DouyuEsportsGoodsSnapshot,
+    DouyuGoodsSnapshot,
+    DouyuTask,
+    DouyuWorkbench,
+    DouyuWorkbenchAccount,
+    DouyuXpdGoodsSnapshot,
+    LoginTask,
+    User,
+)
 from ..permissions import user_has_permission
 from ..schemas import (
     DouyuConfigOut,
@@ -39,7 +59,11 @@ from ..services.douyu_service import (
     douyu_task_payload,
     ensure_douyu_config,
 )
-from core.douyu import FishFinRechargeClient, FishFinRechargeConfig, FishFinRechargeConfigError
+from core.douyu import (
+    FishFinRechargeClient,
+    FishFinRechargeConfig,
+    FishFinRechargeConfigError,
+)
 
 
 router = APIRouter(prefix="/api/douyu", tags=["斗鱼活动"])
@@ -47,8 +71,10 @@ router = APIRouter(prefix="/api/douyu", tags=["斗鱼活动"])
 
 def _supplier_response_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:
@@ -79,21 +105,29 @@ async def supplier_recharge_callback(request: Request, db: Session = Depends(get
         raise HTTPException(status_code=400, detail="供应商回调缺少 out_order_id")
     task = (
         db.query(DouyuTask)
-        .filter(DouyuTask.supplier_out_order_id == out_order_id, DouyuTask.task_type == "create_gold_qr")
+        .filter(
+            DouyuTask.supplier_out_order_id == out_order_id,
+            DouyuTask.task_type == "create_gold_qr",
+        )
         .first()
     )
     if not task:
         raise HTTPException(status_code=404, detail="供应商回调订单不存在")
 
-    status = DouyuBatchRunner._to_int(_supplier_response_value(payload, "order_status", "orderStatus"))
+    status = DouyuBatchRunner._to_int(
+        _supplier_response_value(payload, "order_status", "orderStatus")
+    )
     result = dict(task.result) if isinstance(task.result, dict) else {}
-    result.update({
+    result.update(
+        {
             "out_order_id": out_order_id,
-        "order_id": _supplier_response_value(payload, "order_id", "orderId") or result.get("order_id"),
+            "order_id": _supplier_response_value(payload, "order_id", "orderId")
+            or result.get("order_id"),
             "supplier_order_status": status,
             "supplier_order": DouyuBatchRunner._supplier_result(payload),
             "supplier_callback_received": True,
-    })
+        }
+    )
     if task.status not in {"success", "failed", "stopped"}:
         if status == 2:
             task.status = "success"
@@ -103,17 +137,25 @@ async def supplier_recharge_callback(request: Request, db: Session = Depends(get
                 task.account.bind_status = "gold_recharged"
                 task.account.updated_at = datetime.now(timezone.utc)
         elif status in {3, 4}:
-            reason = str(_supplier_response_value(payload, "fail_reason", "message", "msg") or "供应商直充失败")
+            reason = str(
+                _supplier_response_value(payload, "fail_reason", "message", "msg")
+                or "供应商直充失败"
+            )
             task.status = "failed"
             task.message = reason[:512]
             task.finished_at = datetime.now(timezone.utc)
         else:
             task.status = "running"
-            task.message = f"供应商直充订单处理中（状态 {status if status is not None else '-'}）"
+            task.message = (
+                f"供应商直充订单处理中（状态 {status if status is not None else '-'}）"
+            )
     task.result = result
     callback_success = status not in {3, 4}
     record_audit(
-        db, None, action="recharge:douyu:callback", target=f"supplier_order:{out_order_id}",
+        db,
+        None,
+        action="recharge:douyu:callback",
+        target=f"supplier_order:{out_order_id}",
         detail={
             "out_order_id": out_order_id,
             "task_id": task.id,
@@ -125,7 +167,9 @@ async def supplier_recharge_callback(request: Request, db: Session = Depends(get
         success=callback_success,
     )
     db.commit()
-    logger.info("[douyu] 供应商直充回调已处理: out_order_id={} status={}", out_order_id, status)
+    logger.info(
+        "[douyu] 供应商直充回调已处理: out_order_id={} status={}", out_order_id, status
+    )
     acknowledgement = {"code": 200, "message": "success"}
     acknowledgement["sign"] = client.sign(acknowledgement, "POST")
     return acknowledgement
@@ -137,7 +181,11 @@ def _can_view_all(user: User) -> bool:
 
 def _visible_task_accounts_query(db: Session, current: User):
     """返回当前用户可用于斗鱼任务的账号查询。"""
-    cookie_ids = cookie_account_ids_query(db).subquery()
+    cookie_ids = (
+        select(LoginTask.account_id)
+        .where(LoginTask.status == "success", LoginTask.cookie != "")
+        .distinct()
+    )
     query = (
         db.query(Account)
         .options(joinedload(Account.assigned_user))
@@ -168,7 +216,9 @@ def _visible_tasks_query(db: Session, current: User):
     raise HTTPException(status_code=403, detail="无权查看斗鱼任务")
 
 
-def _require_task_account_access(db: Session, current: User, account_ids: list[int]) -> None:
+def _require_task_account_access(
+    db: Session, current: User, account_ids: list[int]
+) -> None:
     """确保任务只会提交到当前用户可操作的账号。"""
     requested_ids = set(account_ids)
     query = db.query(Account.id).filter(Account.id.in_(requested_ids))
@@ -176,7 +226,7 @@ def _require_task_account_access(db: Session, current: User, account_ids: list[i
         if not user_has_permission(current, "account:view_assigned"):
             raise HTTPException(status_code=403, detail="无权操作斗鱼账号")
         query = query.filter(Account.assigned_to == current.id)
-    allowed_ids = {account_id for account_id, in query.all()}
+    allowed_ids = {account_id for (account_id,) in query.all()}
     if allowed_ids != requested_ids:
         raise HTTPException(status_code=403, detail="包含无权操作的斗鱼账号")
 
@@ -223,7 +273,9 @@ def _account_out(account: Account) -> DouyuTaskAccountOut:
         xpd_fragments=account.xpd_fragments,
         xpd_bind_status=account.xpd_bind_status or "",
         assigned_to=account.assigned_to,
-        assigned_username=account.assigned_user.username if account.assigned_user else None,
+        assigned_username=account.assigned_user.username
+        if account.assigned_user
+        else None,
     )
 
 
@@ -232,30 +284,32 @@ def _task_out(task: DouyuTask, *, include_detail: bool = False) -> DouyuTaskOut:
 
 
 def _config_out(config: DouyuConfig) -> DouyuConfigOut:
+    text = lambda field: str(douyu_config_value(field, getattr(config, field)))
+    number = lambda field: int(douyu_config_value(field, getattr(config, field)))
     return DouyuConfigOut(
-        manual_id=douyu_config_value("manual_id", config.manual_id),
-        rid=douyu_config_value("rid", config.rid),
-        bind_act_alias=douyu_config_value("bind_act_alias", config.bind_act_alias),
-        confirm_act_alias=douyu_config_value("confirm_act_alias", config.confirm_act_alias),
-        legacy_act_alias=douyu_config_value("legacy_act_alias", config.legacy_act_alias),
-        room_id=douyu_config_value("room_id", config.room_id),
-        elite_amount=douyu_config_value("elite_amount", config.elite_amount),
-        esports_manual_id=douyu_config_value("esports_manual_id", config.esports_manual_id),
-        esports_act_alias=douyu_config_value("esports_act_alias", config.esports_act_alias),
-        esports_amount=douyu_config_value("esports_amount", config.esports_amount),
-        esports_chicken_gift_id=douyu_config_value("esports_chicken_gift_id", config.esports_chicken_gift_id),
-        esports_chicken_skin_id=douyu_config_value("esports_chicken_skin_id", config.esports_chicken_skin_id),
-        esports_firework_gift_id=douyu_config_value("esports_firework_gift_id", config.esports_firework_gift_id),
-        esports_firework_skin_id=douyu_config_value("esports_firework_skin_id", config.esports_firework_skin_id),
-        xpd_act_alias=douyu_config_value("xpd_act_alias", config.xpd_act_alias),
-        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),
+        manual_id=text("manual_id"),
+        rid=text("rid"),
+        bind_act_alias=text("bind_act_alias"),
+        confirm_act_alias=text("confirm_act_alias"),
+        legacy_act_alias=text("legacy_act_alias"),
+        room_id=text("room_id"),
+        elite_amount=number("elite_amount"),
+        esports_manual_id=text("esports_manual_id"),
+        esports_act_alias=text("esports_act_alias"),
+        esports_amount=number("esports_amount"),
+        esports_chicken_gift_id=text("esports_chicken_gift_id"),
+        esports_chicken_skin_id=text("esports_chicken_skin_id"),
+        esports_firework_gift_id=text("esports_firework_gift_id"),
+        esports_firework_skin_id=text("esports_firework_skin_id"),
+        xpd_act_alias=text("xpd_act_alias"),
+        xpd_act_id=text("xpd_act_id"),
+        xpd_rid=text("xpd_rid"),
+        gold_pay_type=number("gold_pay_type"),
+        gold_recharge_channel=text("gold_recharge_channel"),
+        gold_api_product_id=text("gold_api_product_id"),
+        gold_api_account_template_name=text("gold_api_account_template_name"),
+        gift_id=text("gift_id"),
+        skin_id=text("skin_id"),
         updated_at=config.updated_at,
     )
 
@@ -290,7 +344,8 @@ def list_task_accounts(
     search_text = (search or "").strip()
     if search_text:
         pattern = f"%{search_text}%"
-        query = query.filter(or_(
+        query = query.filter(
+            or_(
                 Account.username.ilike(pattern),
                 Account.uid.ilike(pattern),
                 Account.nickname.ilike(pattern),
@@ -298,7 +353,8 @@ def list_task_accounts(
                 Account.game_name.ilike(pattern),
                 Account.esports_game_name.ilike(pattern),
                 Account.xpd_game_name.ilike(pattern),
-        ))
+            )
+        )
     total = None
     if page is not None:
         total = query.order_by(None).count()
@@ -309,7 +365,12 @@ def list_task_accounts(
         query = query.limit(500)
     result = [_account_out(account) for account in query.all()]
     if page is not None:
-        return {"items": result, "total": total or 0, "page": page, "page_size": page_size}
+        return {
+            "items": result,
+            "total": total or 0,
+            "page": page,
+            "page_size": page_size,
+        }
     return result
 
 
@@ -328,7 +389,8 @@ def list_task_account_ids(
     search_text = (search or "").strip()
     if search_text:
         pattern = f"%{search_text}%"
-        query = query.filter(or_(
+        query = query.filter(
+            or_(
                 Account.username.ilike(pattern),
                 Account.uid.ilike(pattern),
                 Account.nickname.ilike(pattern),
@@ -336,10 +398,14 @@ def list_task_account_ids(
                 Account.game_name.ilike(pattern),
                 Account.esports_game_name.ilike(pattern),
                 Account.xpd_game_name.ilike(pattern),
-        ))
+            )
+        )
     account_ids = [
         account_id
-        for account_id, in query.enable_eagerloads(False).order_by(Account.id.desc()).with_entities(Account.id).all()
+        for (account_id,) in query.enable_eagerloads(False)
+        .order_by(Account.id.desc())
+        .with_entities(Account.id)
+        .all()
     ]
     return {"account_ids": account_ids, "total": len(account_ids)}
 
@@ -361,11 +427,19 @@ def list_workbench_accounts(
     )
     if not _can_view_all(current):
         rows = rows.filter(Account.assigned_to == current.id)
-    account_ids = [account_id for account_id, in rows.order_by(DouyuWorkbenchAccount.id.asc()).all()]
-    configured = db.query(DouyuWorkbench.id).filter(
+    account_ids = [
+        account_id
+        for (account_id,) in rows.order_by(DouyuWorkbenchAccount.id.asc()).all()
+    ]
+    configured = (
+        db.query(DouyuWorkbench.id)
+        .filter(
             DouyuWorkbench.user_id == current.id,
             DouyuWorkbench.handbook_scope == handbook_scope,
-    ).first() is not None
+        )
+        .first()
+        is not None
+    )
     return {"account_ids": account_ids, "configured": configured}
 
 
@@ -381,10 +455,14 @@ def update_workbench_accounts(
         raise HTTPException(status_code=400, detail="无效的账号ID")
     if account_ids:
         _require_task_account_access(db, current, account_ids)
-    workbench = db.query(DouyuWorkbench).filter(
+    workbench = (
+        db.query(DouyuWorkbench)
+        .filter(
             DouyuWorkbench.user_id == current.id,
             DouyuWorkbench.handbook_scope == req.handbook_scope,
-    ).first()
+        )
+        .first()
+    )
     if workbench is None:
         db.add(DouyuWorkbench(user_id=current.id, handbook_scope=req.handbook_scope))
     else:
@@ -393,14 +471,16 @@ def update_workbench_accounts(
         DouyuWorkbenchAccount.user_id == current.id,
         DouyuWorkbenchAccount.handbook_scope == req.handbook_scope,
     ).delete(synchronize_session=False)
-    db.add_all([
+    db.add_all(
+        [
             DouyuWorkbenchAccount(
                 user_id=current.id,
                 handbook_scope=req.handbook_scope,
                 account_id=account_id,
             )
             for account_id in account_ids
-    ])
+        ]
+    )
     db.commit()
     return {"account_ids": account_ids, "success": True}
 
@@ -423,7 +503,8 @@ def get_recharge_channel(
     config = ensure_douyu_config(db)
     return {
         "gold_recharge_channel": douyu_config_value(
-            "gold_recharge_channel", config.gold_recharge_channel,
+            "gold_recharge_channel",
+            config.gold_recharge_channel,
         ),
     }
 
@@ -448,7 +529,10 @@ def update_config(
     recharge_fields = [field for field in changed_fields if field.startswith("gold_")]
     if recharge_fields:
         record_audit(
-            db, current, action="recharge:douyu:config", target="douyu_config",
+            db,
+            current,
+            action="recharge:douyu:config",
+            target="douyu_config",
             detail={"changed_fields": recharge_fields},
         )
     db.commit()
@@ -472,7 +556,11 @@ def list_esports_goods(
     current: User = Depends(require_permission("douyu:task")),
 ):
     """查看已缓存的电竞手册皮肤商城商品。"""
-    rows = db.query(DouyuEsportsGoodsSnapshot).order_by(DouyuEsportsGoodsSnapshot.id.asc()).all()
+    rows = (
+        db.query(DouyuEsportsGoodsSnapshot)
+        .order_by(DouyuEsportsGoodsSnapshot.id.asc())
+        .all()
+    )
     return rows
 
 
@@ -482,7 +570,9 @@ def list_xpd_goods(
     current: User = Depends(require_permission("douyu:task")),
 ):
     """查看已缓存的和平小店商品。"""
-    rows = db.query(DouyuXpdGoodsSnapshot).order_by(DouyuXpdGoodsSnapshot.id.asc()).all()
+    rows = (
+        db.query(DouyuXpdGoodsSnapshot).order_by(DouyuXpdGoodsSnapshot.id.asc()).all()
+    )
     return rows
 
 
@@ -516,27 +606,39 @@ async def create_task_batch(
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc
     if count == 0:
-        raise HTTPException(status_code=400, detail="没有可执行的斗鱼账号，请先登录获取 Cookie")
+        raise HTTPException(
+            status_code=400, detail="没有可执行的斗鱼账号，请先登录获取 Cookie"
+        )
 
     if req.task_type == "create_gold_qr":
         config = ensure_douyu_config(db)
         recharge_channel = str(
             douyu_config_value("gold_recharge_channel", config.gold_recharge_channel)
         ).strip()
-        payment_method = "API 直充支付" if recharge_channel == "supplier_api" else "微信扫码支付"
+        payment_method = (
+            "API 直充支付" if recharge_channel == "supplier_api" else "微信扫码支付"
+        )
         recharge_accounts = (
             db.query(DouyuTask, Account)
             .join(Account, Account.id == DouyuTask.account_id)
-            .filter(DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr")
+            .filter(
+                DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr"
+            )
             .order_by(DouyuTask.id.asc())
             .limit(100)
             .all()
         )
         record_audit(
-            db, current, action="recharge:douyu:create", target=f"douyu_batch:{batch_id}",
+            db,
+            current,
+            action="recharge:douyu:create",
+            target=f"douyu_batch:{batch_id}",
             detail={
-                "batch_id": batch_id, "task_type": req.task_type, "count": count,
-                "account_count": len(req.account_ids), "handbook_scope": req.handbook_scope,
+                "batch_id": batch_id,
+                "task_type": req.task_type,
+                "count": count,
+                "account_count": len(req.account_ids),
+                "handbook_scope": req.handbook_scope,
                 "recharge_channel": recharge_channel,
                 "payment_method": payment_method,
                 # 仅记录充值身份，禁止在审计中写入 Cookie、密码等凭据。
@@ -578,7 +680,9 @@ async def create_task_batch(
 def list_tasks(
     batch_id: str | None = None,
     handbook_scope: str | None = Query(None, pattern="^(elite|esports|peace)$"),
-    include_detail: bool = Query(False, description="是否返回完整任务结果（默认否，轮询请保持 false）"),
+    include_detail: bool = Query(
+        False, description="是否返回完整任务结果（默认否，轮询请保持 false）"
+    ),
     page: int | None = Query(None, ge=1),
     page_size: int = Query(100, ge=1, le=500),
     db: Session = Depends(get_db),
@@ -590,11 +694,13 @@ def list_tasks(
         query = query.filter(DouyuTask.batch_id == batch_id)
     if handbook_scope:
         # 旧任务没有归属字段，按历史任务类型继续展示，但绝不会触发自动二维码弹窗。
-        query = query.filter(or_(
+        query = query.filter(
+            or_(
                 DouyuTask.handbook_scope == handbook_scope,
                 (DouyuTask.handbook_scope == "legacy")
                 & DouyuTask.task_type.in_(DOUYU_HANDBOOK_TASK_TYPES[handbook_scope]),
-        ))
+            )
+        )
     total = None
     if page is not None:
         total = (
@@ -611,7 +717,12 @@ def list_tasks(
         query = query.limit(page_size)
     result = [_task_out(task, include_detail=include_detail) for task in query.all()]
     if page is not None:
-        return {"items": result, "total": total or 0, "page": page, "page_size": page_size}
+        return {
+            "items": result,
+            "total": total or 0,
+            "page": page,
+            "page_size": page_size,
+        }
     return result
 
 
@@ -627,7 +738,11 @@ def cleanup_orphan_tasks(
         statuses=("pending", "running"),
         message="任务已中断（无执行器接管）",
     )
-    return {"message": f"已清理 {cleaned} 个残留任务", "cleaned": cleaned, "success": True}
+    return {
+        "message": f"已清理 {cleaned} 个残留任务",
+        "cleaned": cleaned,
+        "success": True,
+    }
 
 
 @router.get("/tasks/{task_id}", response_model=DouyuTaskOut)
@@ -661,25 +776,42 @@ def stop_batch(
     if batch:
         if batch.get("finished"):
             douyu_batch_registry.pop(batch_id)
-            cleaned = cleanup_orphan_douyu_tasks(db, batch_id=batch_id, message="批次已结束")
+            cleaned = cleanup_orphan_douyu_tasks(
+                db, batch_id=batch_id, message="批次已结束"
+            )
             if cleaned:
-                return {"message": f"批次已结束，已清理 {cleaned} 个残留任务", "success": True}
+                return {
+                    "message": f"批次已结束，已清理 {cleaned} 个残留任务",
+                    "success": True,
+                }
             raise HTTPException(status_code=404, detail="批次已结束")
         batch["runner"].stop()
         if is_recharge_batch:
             record_audit(
-                db, current, action="recharge:douyu:stop", target=f"douyu_batch:{batch_id}",
+                db,
+                current,
+                action="recharge:douyu:stop",
+                target=f"douyu_batch:{batch_id}",
                 detail={"batch_id": batch_id, "mode": "running"},
             )
             db.commit()
         return {"message": "已发送停止信号", "success": True}
 
-    cleaned = cleanup_orphan_douyu_tasks(db, batch_id=batch_id, message="任务已停止（批次不存在）")
+    cleaned = cleanup_orphan_douyu_tasks(
+        db, batch_id=batch_id, message="任务已停止（批次不存在）"
+    )
     if cleaned:
         if is_recharge_batch:
             record_audit(
-                db, current, action="recharge:douyu:stop", target=f"douyu_batch:{batch_id}",
-                detail={"batch_id": batch_id, "mode": "orphan_cleanup", "cleaned": cleaned},
+                db,
+                current,
+                action="recharge:douyu:stop",
+                target=f"douyu_batch:{batch_id}",
+                detail={
+                    "batch_id": batch_id,
+                    "mode": "orphan_cleanup",
+                    "cleaned": cleaned,
+                },
             )
             db.commit()
         return {"message": f"已清理 {cleaned} 个残留任务", "success": True}