"""斗鱼活动任务路由。""" from __future__ import annotations import asyncio import threading from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect from loguru import logger from sqlalchemy import or_ from sqlalchemy.orm import Session, joinedload 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 ..permissions import user_has_permission from ..schemas import ( DouyuConfigOut, DouyuConfigUpdate, DouyuGoodsOut, DouyuTaskAccountOut, DouyuTaskBatchRequest, DouyuTaskOut, DouyuWorkbenchAccountsUpdate, DouyuXpdGoodsOut, ) from ..services.douyu_runner import DouyuBatchRunner, douyu_batch_registry from ..services.audit_service import record_audit from ..services.douyu_service import ( DOUYU_CONFIG_FIELDS, DOUYU_HANDBOOK_TASK_TYPES, SUPPORTED_DOUYU_TASK_TYPES, apply_douyu_config_defaults, cleanup_orphan_douyu_tasks, cookie_account_ids_query, create_douyu_planned_tasks, douyu_config_value, douyu_task_payload, ensure_douyu_config, ) from core.douyu import FishFinRechargeClient, FishFinRechargeConfig, FishFinRechargeConfigError 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 {} for source in (payload, data, result): for key in keys: if source.get(key) is not None: return source[key] return None @router.post("/supplier-recharge/callback") async def supplier_recharge_callback(request: Request, db: Session = Depends(get_db)): """接收供应商直充终态通知，验签后按 out_order_id 幂等更新任务。""" try: payload = await request.json() except ValueError as exc: raise HTTPException(status_code=400, detail="供应商回调不是 JSON") from exc if not isinstance(payload, dict): raise HTTPException(status_code=400, detail="供应商回调格式无效") try: client = FishFinRechargeClient(FishFinRechargeConfig.from_env()) except FishFinRechargeConfigError as exc: logger.error("[douyu] 供应商直充回调配置无效: {}", exc) raise HTTPException(status_code=503, detail="供应商直充回调未配置") from exc if not client.verify_response_sign(payload, "POST"): logger.warning("[douyu] 供应商直充回调验签失败: keys={}", sorted(payload)) raise HTTPException(status_code=401, detail="供应商回调签名无效") out_order_id = str(_supplier_response_value(payload, "out_order_id") or "").strip() if not out_order_id: 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") .first() ) if not task: raise HTTPException(status_code=404, detail="供应商回调订单不存在") 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, }) if task.status not in {"success", "failed", "stopped"}: if status == 2: task.status = "success" task.message = "供应商直充成功（异步通知）" task.finished_at = datetime.now(timezone.utc) if task.account: 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 "供应商直充失败") 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.result = result callback_success = status not in {3, 4} record_audit( db, None, action="recharge:douyu:callback", target=f"supplier_order:{out_order_id}", detail={ "out_order_id": out_order_id, "task_id": task.id, "order_id": result.get("order_id"), "supplier_order_status": status, "task_status": task.status, "message": task.message, }, success=callback_success, ) db.commit() logger.info("[douyu] 供应商直充回调已处理: out_order_id={} status={}", out_order_id, status) acknowledgement = {"code": 200, "message": "success"} acknowledgement["sign"] = client.sign(acknowledgement, "POST") return acknowledgement def _can_view_all(user: User) -> bool: return user_has_permission(user, "account:view_all") def _visible_task_accounts_query(db: Session, current: User): """返回当前用户可用于斗鱼任务的账号查询。""" cookie_ids = cookie_account_ids_query(db).subquery() query = ( db.query(Account) .options(joinedload(Account.assigned_user)) .filter(Account.id.in_(cookie_ids)) ) if _can_view_all(current): return query if user_has_permission(current, "account:view_assigned"): return query.filter(Account.assigned_to == current.id) raise HTTPException(status_code=403, detail="无权查看斗鱼账号") def _visible_tasks_query(db: Session, current: User): """返回当前用户可查看的斗鱼任务查询。""" query = db.query(DouyuTask).options(joinedload(DouyuTask.account)) if _can_view_all(current): return query if user_has_permission(current, "account:view_assigned"): return query.join(DouyuTask.account).filter(Account.assigned_to == current.id) raise HTTPException(status_code=403, detail="无权查看斗鱼任务") 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)) if not _can_view_all(current): 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()} if allowed_ids != requested_ids: raise HTTPException(status_code=403, detail="包含无权操作的斗鱼账号") def _require_batch_owner(db: Session, current: User, batch_id: str) -> None: """客服只能停止或订阅自己创建的任务批次。""" if _can_view_all(current): return exists = ( db.query(DouyuTask.id) .filter(DouyuTask.batch_id == batch_id, DouyuTask.created_by == current.id) .first() ) if not exists: raise HTTPException(status_code=403, detail="无权操作该斗鱼任务批次") def _account_out(account: Account) -> DouyuTaskAccountOut: return DouyuTaskAccountOut( id=account.id, username=account.username, uid=account.uid or "", nickname=account.nickname or "", tag=account.tag or "", points=account.points, game_name=account.game_name or "", game_channel=account.game_channel or "", gold_balance=account.gold_balance, exchange_balance=account.exchange_balance, bind_status=account.bind_status or "", change_role_wait_time=account.change_role_wait_time, esports_points=account.esports_points, esports_game_name=account.esports_game_name or "", esports_game_channel=account.esports_game_channel or "", esports_bind_status=account.esports_bind_status or "", esports_change_role_wait_time=account.esports_change_role_wait_time, esports_can_change_time=account.esports_can_change_time, xpd_game_name=account.xpd_game_name or "", xpd_openid=account.xpd_openid or "", xpd_role_id=account.xpd_role_id or "", xpd_plat_id=account.xpd_plat_id, xpd_area_id=account.xpd_area_id, xpd_balance=account.xpd_balance, 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, ) def _task_out(task: DouyuTask, *, include_detail: bool = False) -> DouyuTaskOut: return DouyuTaskOut(**douyu_task_payload(task, include_detail=include_detail)) def _config_out(config: DouyuConfig) -> DouyuConfigOut: 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), updated_at=config.updated_at, ) @router.get("/task-types") def task_types(current: User = Depends(require_permission("douyu:task"))): """返回斗鱼任务类型。""" return SUPPORTED_DOUYU_TASK_TYPES @router.get("/accounts") def list_task_accounts( search: str = Query(""), tag: str = Query(""), ids: str = Query(""), page: int | None = Query(None, ge=1), page_size: int = Query(50, ge=1, le=200), db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """查看可执行斗鱼任务的账号（必须有成功 Cookie）。 tag 按账号标签精确筛选；ids 为逗号分隔的账号 ID，用于工作台按已导入账号过滤；为空时返回全部。 """ query = _visible_task_accounts_query(db, current) id_list = [int(x) for x in ids.split(",") if x.strip().isdigit()] if id_list: query = query.filter(Account.id.in_(id_list)) tag_text = (tag or "").strip() if tag_text: query = query.filter(Account.tag == tag_text) 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), )) total = None if page is not None: total = query.order_by(None).count() query = query.order_by(Account.id.desc()) if page is not None: query = query.offset((page - 1) * page_size).limit(page_size) else: 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 result @router.get("/accounts/ids") def list_task_account_ids( search: str = Query(""), tag: str = Query(""), db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """返回当前用户可导入工作台的全部账号 ID，不受列表分页限制。""" query = _visible_task_accounts_query(db, current) tag_text = (tag or "").strip() if tag_text: query = query.filter(Account.tag == tag_text) 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), )) account_ids = [ account_id 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)} @router.get("/workbench-accounts") def list_workbench_accounts( handbook_scope: str = Query(..., pattern="^(elite|esports|peace)$"), db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """返回当前用户在指定工作台启用的账号，供不同浏览器同步。""" rows = ( db.query(DouyuWorkbenchAccount.account_id) .join(Account, Account.id == DouyuWorkbenchAccount.account_id) .filter( DouyuWorkbenchAccount.user_id == current.id, DouyuWorkbenchAccount.handbook_scope == handbook_scope, ) ) 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 return {"account_ids": account_ids, "configured": configured} @router.put("/workbench-accounts") def update_workbench_accounts( req: DouyuWorkbenchAccountsUpdate, db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """用当前完整账号集合覆盖一个工作台，作为跨浏览器的同步状态。""" account_ids = sorted(set(req.account_ids)) if any(account_id < 1 for account_id in account_ids): 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() if workbench is None: db.add(DouyuWorkbench(user_id=current.id, handbook_scope=req.handbook_scope)) else: workbench.updated_at = datetime.now(timezone.utc) db.query(DouyuWorkbenchAccount).filter( 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.commit() return {"account_ids": account_ids, "success": True} @router.get("/config", response_model=DouyuConfigOut) def get_config( db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:config")), ): """获取斗鱼活动配置。""" return _config_out(ensure_douyu_config(db)) @router.get("/recharge-channel") def get_recharge_channel( db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """返回当前鱼翅充值渠道，不暴露供应商商品等配置细节。""" config = ensure_douyu_config(db) return { "gold_recharge_channel": douyu_config_value( "gold_recharge_channel", config.gold_recharge_channel, ), } @router.put("/config", response_model=DouyuConfigOut) def update_config( req: DouyuConfigUpdate, db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:config")), ): """更新斗鱼活动配置。""" config = ensure_douyu_config(db) changed_fields = [] for field in DOUYU_CONFIG_FIELDS: value = getattr(req, field) if value is None: continue setattr(config, field, value.strip() if isinstance(value, str) else value) changed_fields.append(field) apply_douyu_config_defaults(config) config.updated_at = datetime.now(timezone.utc) 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", detail={"changed_fields": recharge_fields}, ) db.commit() db.refresh(config) return _config_out(config) @router.get("/goods", response_model=list[DouyuGoodsOut]) def list_goods( db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """查看已缓存的斗鱼商品快照。""" rows = db.query(DouyuGoodsSnapshot).order_by(DouyuGoodsSnapshot.id.asc()).all() return rows @router.get("/esports-goods", response_model=list[DouyuGoodsOut]) def list_esports_goods( db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """查看已缓存的电竞手册皮肤商城商品。""" rows = db.query(DouyuEsportsGoodsSnapshot).order_by(DouyuEsportsGoodsSnapshot.id.asc()).all() return rows @router.get("/xpd-goods", response_model=list[DouyuXpdGoodsOut]) def list_xpd_goods( db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """查看已缓存的和平小店商品。""" rows = db.query(DouyuXpdGoodsSnapshot).order_by(DouyuXpdGoodsSnapshot.id.asc()).all() return rows @router.post("/tasks/batch") async def create_task_batch( req: DouyuTaskBatchRequest, db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """创建斗鱼任务记录并启动后台执行器。""" if not req.account_ids: raise HTTPException(status_code=400, detail="请选择斗鱼账号") _require_task_account_access(db, current, req.account_ids) cleanup_orphan_douyu_tasks( db, active_batch_ids=douyu_batch_registry.active_ids(), statuses=("pending", "running"), message="任务已中断（无执行器接管）", ) try: batch_id, count = create_douyu_planned_tasks( db, req.account_ids, req.task_type, req.handbook_scope, current.id, req.payload, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc if count == 0: 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 "微信扫码支付" 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") .order_by(DouyuTask.id.asc()) .limit(100) .all() ) record_audit( 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, "recharge_channel": recharge_channel, "payment_method": payment_method, # 仅记录充值身份，禁止在审计中写入 Cookie、密码等凭据。 "recharge_accounts": [ { "task_id": task.id, "username": account.username, "douyu_uid": account.uid or "", "douyu_nickname": account.nickname or "", } for task, account in recharge_accounts ], "recharge_accounts_truncated": count > len(recharge_accounts), }, ) db.commit() log_queue = asyncio.Queue() loop = asyncio.get_running_loop() thread_db = SessionLocal() runner = DouyuBatchRunner( db=thread_db, batch_id=batch_id, task_type=req.task_type, payload=req.payload, log_queue=log_queue, loop=loop, concurrency=req.concurrency, ) douyu_batch_registry.register(batch_id, log_queue, loop, runner) thread = threading.Thread(target=runner.run, daemon=True) thread.start() return {"batch_id": batch_id, "count": count, "success": True} @router.get("/tasks") 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）"), page: int | None = Query(None, ge=1), page_size: int = Query(100, ge=1, le=500), db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """查看斗鱼任务记录，默认只返回最近 100 条。""" query = _visible_tasks_query(db, current) if batch_id: 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]), )) total = None if page is not None: total = query.enable_eagerloads(False).order_by(None).count() query = query.order_by(DouyuTask.id.desc()) if page is not None: query = query.offset((page - 1) * page_size).limit(page_size) else: 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 result @router.post("/tasks/cleanup-orphans") def cleanup_orphan_tasks( db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """手动清理没有内存执行器接管的斗鱼任务。""" cleaned = cleanup_orphan_douyu_tasks( db, active_batch_ids=douyu_batch_registry.active_ids(), statuses=("pending", "running"), message="任务已中断（无执行器接管）", ) return {"message": f"已清理 {cleaned} 个残留任务", "cleaned": cleaned, "success": True} @router.get("/tasks/{task_id}", response_model=DouyuTaskOut) def get_task( task_id: int, db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """获取单条斗鱼任务详情。""" task = _visible_tasks_query(db, current).filter(DouyuTask.id == task_id).first() if not task: raise HTTPException(status_code=404, detail="任务不存在") return _task_out(task, include_detail=True) @router.post("/stop/{batch_id}") def stop_batch( batch_id: str, db: Session = Depends(get_db), current: User = Depends(require_permission("douyu:task")), ): """停止正在运行的斗鱼批次。""" _require_batch_owner(db, current, batch_id) is_recharge_batch = ( db.query(DouyuTask.id) .filter(DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr") .first() is not None ) batch = douyu_batch_registry.get(batch_id) if batch: if batch.get("finished"): douyu_batch_registry.pop(batch_id) cleaned = cleanup_orphan_douyu_tasks(db, batch_id=batch_id, message="批次已结束") if cleaned: 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}", detail={"batch_id": batch_id, "mode": "running"}, ) db.commit() return {"message": "已发送停止信号", "success": True} 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.commit() return {"message": f"已清理 {cleaned} 个残留任务", "success": True} raise HTTPException(status_code=404, detail="批次不存在或已结束") @router.websocket("/ws/{batch_id}") async def ws_douyu_logs(websocket: WebSocket, batch_id: str): """斗鱼实时日志推送通道。""" user = authenticate_websocket(websocket) if not user: await websocket.close(code=1008, reason="未授权") return if not user_has_permission(user, "douyu:task"): await websocket.close(code=1008, reason="无权限") return db = SessionLocal() try: _require_batch_owner(db, user, batch_id) except HTTPException: await websocket.close(code=1008, reason="无权访问该任务批次") return finally: db.close() await websocket.accept() batch = douyu_batch_registry.get(batch_id) if not batch: await websocket.send_json({"level": "error", "message": "批次不存在或已结束"}) await websocket.close() return log_queue: asyncio.Queue = batch["log_queue"] try: while True: try: msg = await asyncio.wait_for(log_queue.get(), timeout=30) await websocket.send_json(msg) if msg.get("level") == "result": await asyncio.sleep(0.1) break except asyncio.TimeoutError: await websocket.send_json({"level": "heartbeat", "message": ""}) except WebSocketDisconnect: pass finally: latest = douyu_batch_registry.get(batch_id) if latest and latest.get("finished"): douyu_batch_registry.pop(batch_id)