diff --git a/web/backend/routers/douyu.py b/web/backend/routers/douyu.py index 699f8d1..b91e664 100644 --- a/web/backend/routers/douyu.py +++ b/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({ - "out_order_id": out_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, - }) + result.update( + { + "out_order_id": out_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,15 +344,17 @@ def list_task_accounts( search_text = (search or "").strip() if search_text: pattern = f"%{search_text}%" - query = query.filter(or_( - Account.username.ilike(pattern), - Account.uid.ilike(pattern), - Account.nickname.ilike(pattern), - Account.tag.ilike(pattern), - Account.game_name.ilike(pattern), - Account.esports_game_name.ilike(pattern), - Account.xpd_game_name.ilike(pattern), - )) + query = query.filter( + or_( + Account.username.ilike(pattern), + Account.uid.ilike(pattern), + Account.nickname.ilike(pattern), + Account.tag.ilike(pattern), + 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,18 +389,23 @@ def list_task_account_ids( search_text = (search or "").strip() if search_text: pattern = f"%{search_text}%" - query = query.filter(or_( - Account.username.ilike(pattern), - Account.uid.ilike(pattern), - Account.nickname.ilike(pattern), - Account.tag.ilike(pattern), - Account.game_name.ilike(pattern), - Account.esports_game_name.ilike(pattern), - Account.xpd_game_name.ilike(pattern), - )) + query = query.filter( + or_( + Account.username.ilike(pattern), + Account.uid.ilike(pattern), + Account.nickname.ilike(pattern), + Account.tag.ilike(pattern), + 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( - DouyuWorkbench.user_id == current.id, - DouyuWorkbench.handbook_scope == handbook_scope, - ).first() is not None + 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 + ) 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( - DouyuWorkbench.user_id == current.id, - DouyuWorkbench.handbook_scope == req.handbook_scope, - ).first() + workbench = ( + db.query(DouyuWorkbench) + .filter( + DouyuWorkbench.user_id == current.id, + DouyuWorkbench.handbook_scope == req.handbook_scope, + ) + .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([ - DouyuWorkbenchAccount( - user_id=current.id, - handbook_scope=req.handbook_scope, - account_id=account_id, - ) - for account_id in account_ids - ]) + 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_( - DouyuTask.handbook_scope == handbook_scope, - (DouyuTask.handbook_scope == "legacy") - & DouyuTask.task_type.in_(DOUYU_HANDBOOK_TASK_TYPES[handbook_scope]), - )) + 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}