"""斗鱼活动任务批次执行器。""" from __future__ import annotations import asyncio import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from typing import Optional from loguru import logger from sqlalchemy.orm import Session, joinedload from core.douyu import DouyuActivityClient, DouyuActivityError from ..database import SessionLocal from ..models import Account, DouyuGoodsSnapshot, DouyuTask from .douyu_service import ( DOUYU_CONFIG_FIELDS, account_uid, douyu_config_value, ensure_douyu_config, latest_success_cookie, update_account_profile_from_cookie, ) DOUYU_LEGACY_BIND_ACT_ALIAS = "20250213NQCYX" DOUYU_BIND_ROLE_POLL_SECONDS = 65 DOUYU_BIND_ROLE_POLL_INTERVAL = 5 class DouyuBatchRunner: """批量执行斗鱼活动任务，通过队列推送实时日志。""" def __init__( self, db: Session, batch_id: str, task_type: str, payload: Optional[dict] = None, log_queue: Optional[asyncio.Queue] = None, loop: Optional[asyncio.AbstractEventLoop] = None, concurrency: int = 3, ): self.db = db self.batch_id = batch_id self.task_type = task_type self.payload = payload or {} self.log_queue = log_queue self.loop = loop self.concurrency = max(1, min(concurrency, 10)) self._stop = threading.Event() self._counter_lock = threading.Lock() self._started = 0 def stop(self): self._stop.set() def _push_log(self, level: str, message: str): if level == "result": try: douyu_batch_registry.mark_finished(self.batch_id) except NameError: pass if level != "result" and message: log_func = getattr(logger, level, logger.info) log_func(f"[douyu] {message}") if self.log_queue and self.loop: asyncio.run_coroutine_threadsafe( self.log_queue.put({"level": level, "message": message}), self.loop, ) @staticmethod def _account_name(account: Account) -> str: return account.nickname or account.username or account.uid or f"#{account.id}" @staticmethod def _to_int(value) -> int | None: if value is None: return None try: return int(value) except (TypeError, ValueError): return None @staticmethod def _format_wait_time(seconds: int | None) -> str: if seconds is None: return "" seconds = max(0, int(seconds)) days, rem = divmod(seconds, 86400) hours, rem = divmod(rem, 3600) minutes, sec = divmod(rem, 60) if days: return f"{days}天{hours}小时{minutes}分" if hours: return f"{hours}小时{minutes}分{sec}秒" return f"{minutes}分{sec}秒" @staticmethod def _bind_qr_act_alias(config: dict) -> str: """生成绑定二维码用的活动 alias。""" bind_alias = str(config.get("bind_act_alias") or "").strip() confirm_alias = str(config.get("confirm_act_alias") or "").strip() if not bind_alias or bind_alias == DOUYU_LEGACY_BIND_ACT_ALIAS: return confirm_alias or bind_alias return bind_alias @staticmethod def _query_bind_act_aliases(config: dict) -> list[str]: """查询/轮询角色用的 alias 列表。 现网最新绑定信息在 legacy(cjm)；活动 alias 可能仍用于扫码/确认， 所以按优先级去重返回多个，轮询时取“更像新扫码结果”的那个。 """ ordered = [ str(config.get("legacy_act_alias") or "").strip(), str(config.get("confirm_act_alias") or "").strip(), str(config.get("bind_act_alias") or "").strip(), ] aliases: list[str] = [] for alias in ordered: if alias and alias not in aliases: aliases.append(alias) return aliases @staticmethod def _confirm_act_alias(config: dict) -> str: """确认绑定接口用的活动 alias。""" confirm_alias = str(config.get("confirm_act_alias") or "").strip() if confirm_alias: return confirm_alias bind_alias = str(config.get("bind_act_alias") or "").strip() if bind_alias and bind_alias != DOUYU_LEGACY_BIND_ACT_ALIAS: return bind_alias return str(config.get("legacy_act_alias") or "").strip() # 兼容旧调用名 @classmethod def _current_bind_act_alias(cls, config: dict) -> str: aliases = cls._query_bind_act_aliases(config) return aliases[0] if aliases else cls._bind_qr_act_alias(config) @staticmethod def _role_channel(bind_info: dict) -> str: return " / ".join( part for part in [bind_info.get("area_name"), bind_info.get("plat_name")] if part ) @staticmethod def _is_truthy_flag(value) -> bool: if value is True: return True if value is False or value is None: return False text = str(value).strip().lower() return text in {"1", "true", "yes", "y"} @classmethod def _is_bound_act(cls, bind_info: dict | None) -> bool: if not bind_info: return False return cls._is_truthy_flag(bind_info.get("is_bound_act")) @classmethod def _can_change_role(cls, bind_info: dict | None) -> bool: """综合 can_change_role 与换绑倒计时判断是否允许换绑。""" if not bind_info: return True wait_time = cls._to_int(bind_info.get("change_role_wait_time")) if wait_time is not None and wait_time > 0: return False if bind_info.get("can_change_role") is not None: return cls._is_truthy_flag(bind_info.get("can_change_role")) # 已绑定但接口没给倒计时/开关时，默认允许（避免误杀首次绑定） return True @classmethod def _is_change_cooling(cls, bind_info: dict | None) -> bool: """是否处于换绑冷却：已有绑定角色且当前不可换绑。""" if not bind_info: return False role_name = str(bind_info.get("role_name") or "").strip() is_bound = cls._is_bound_act(bind_info) or bool(role_name) if not is_bound: return False return not cls._can_change_role(bind_info) @classmethod def _is_pending_role( cls, bind_info: dict | None, *, baseline_role_name: str = "", baseline_is_bound_act: bool = False, ) -> bool: """判断是否出现了可确认的扫码角色。 规则： 1. 必须有角色名 2. 已绑定同一角色（或无 baseline 的已绑定）不算 pending 3. 角色名相对 baseline 变化，或从已绑定变成待确认，算 pending 4. need_bind_act / need_bind_role 且角色相对 baseline 有变化，算 pending 5. 无 baseline 且未绑定但有角色，视为待确认残留 """ if not bind_info: return False role_name = str(bind_info.get("role_name") or "").strip() if not role_name: return False is_bound_act = cls._is_bound_act(bind_info) need_bind_act = cls._is_truthy_flag(bind_info.get("need_bind_act")) need_bind_role = cls._is_truthy_flag(bind_info.get("need_bind_role")) role_changed = bool(baseline_role_name) and role_name != baseline_role_name if is_bound_act: # 已绑定：仅当相对 baseline 角色发生变化时才视为新扫码结果 return role_changed if need_bind_act or need_bind_role: if not baseline_role_name: return True return role_changed or baseline_is_bound_act if not baseline_role_name: return True if role_changed: return True # 同一角色从已绑定变为未绑定 return baseline_is_bound_act def _bind_snapshot(self, bind_info: dict | None) -> dict: info = bind_info or {} role_name = str(info.get("role_name") or "") return { "role_name": role_name, "area_name": info.get("area_name") or "", "plat_name": info.get("plat_name") or "", "nickname": info.get("nickname") or "", "is_bound_act": self._is_bound_act(info), "is_bound_role": self._is_truthy_flag(info.get("is_bound_role")), "is_bound_account": self._is_truthy_flag(info.get("is_bound_account")), "need_bind_act": self._is_truthy_flag(info.get("need_bind_act")), "need_bind_role": self._is_truthy_flag(info.get("need_bind_role")), "change_role_wait_time": self._to_int(info.get("change_role_wait_time")), "can_change_role": self._can_change_role(info), "bind_info": info, } def _format_bind_summary(self, bind_info: dict | None, *, pending: bool | None = None) -> str: snap = self._bind_snapshot(bind_info) role = snap["role_name"] or "-" area = snap["area_name"] or "-" plat = snap["plat_name"] or "-" pending_text = "" if pending is not None: pending_text = f" pending={1 if pending else 0}" return ( f"role={role} area={area} plat={plat}" f" bound_act={1 if snap['is_bound_act'] else 0}" f" bound_role={1 if snap['is_bound_role'] else 0}" f" need_act={1 if snap['need_bind_act'] else 0}" f" need_role={1 if snap['need_bind_role'] else 0}" f" wait={snap['change_role_wait_time'] if snap['change_role_wait_time'] is not None else '-'}" f"{pending_text}" ) def _apply_bind_info_to_account(self, account: Account, bind_info: dict, status: str) -> None: role_name = str(bind_info.get("role_name") or "") account.game_name = role_name or account.game_name account.game_channel = self._role_channel(bind_info) or account.game_channel account.change_role_wait_time = self._to_int(bind_info.get("change_role_wait_time")) account.bind_status = status account.updated_at = datetime.now(timezone.utc) def _mark_task( self, db: Session, task: DouyuTask, status: str, message: str, result: dict | None = None, ) -> None: task.status = status task.message = message[:512] if result is not None: task.result = result task.finished_at = datetime.now(timezone.utc) db.commit() def _update_task_progress( self, db: Session, task: DouyuTask, status: str, message: str, result: dict | None = None, ) -> None: task.status = status task.message = message[:512] if result is not None: task.result = result db.commit() def _upsert_goods(self, db: Session, goods: list[dict]) -> None: now = datetime.now(timezone.utc) for raw in goods: commodity_id = str(raw.get("commodityId") or raw.get("commodity_id") or "") if not commodity_id: continue row = ( db.query(DouyuGoodsSnapshot) .filter(DouyuGoodsSnapshot.commodity_id == commodity_id) .first() ) score = self._to_int(raw.get("score")) if row is None: row = DouyuGoodsSnapshot(commodity_id=commodity_id) db.add(row) row.name = str(raw.get("commodityName") or raw.get("name") or "") row.score = score row.status = str(raw.get("status") or "") row.raw = raw row.updated_at = now db.commit() def _config_info(self, db: Session) -> dict: config = ensure_douyu_config(db) return {field: douyu_config_value(field, getattr(config, field, None)) for field in DOUYU_CONFIG_FIELDS} def _task_payload(self, task: DouyuTask) -> dict: result = task.result if isinstance(task.result, dict) else {} payload = result.get("payload") if isinstance(result.get("payload"), dict) else {} return {**payload, **self.payload} def _client(self, cookie: str) -> DouyuActivityClient: return DouyuActivityClient(cookie, logger=lambda msg: self._push_log("debug", msg)) def _fetch_bind_info_candidates( self, client: DouyuActivityClient, aliases: list[str], ) -> list[dict]: """按多个 actAlias 查询绑定信息，保留成功结果。""" results: list[dict] = [] for alias in aliases: if not alias: continue try: info = client.bind_info(alias, v2=True) except DouyuActivityError as exc: self._push_log("warning", f"查询绑定信息失败 act={alias}: {exc}") continue info = {**info, "act_alias": alias} results.append(info) return results def _pick_bind_info( self, candidates: list[dict], *, baseline_role_name: str = "", baseline_is_bound_act: bool = False, prefer_pending: bool = True, prefer_aliases: list[str] | None = None, ) -> dict | None: """从多个 alias 结果里挑最有用的绑定信息。 - prefer_pending=True：优先选“待确认/新扫码角色”（通常来自 cjm） - prefer_aliases：在同等条件下优先指定 alias（如活动当前绑定） """ if not candidates: return None def _alias_rank(info: dict) -> int: alias = str(info.get("act_alias") or "") if not prefer_aliases: return 0 try: return prefer_aliases.index(alias) except ValueError: return len(prefer_aliases) + 1 ranked = sorted(enumerate(candidates), key=lambda item: (_alias_rank(item[1]), item[0])) ordered = [item[1] for item in ranked] if prefer_pending: for info in ordered: if self._is_pending_role( info, baseline_role_name=baseline_role_name, baseline_is_bound_act=baseline_is_bound_act, ): return info for info in ordered: if str(info.get("role_name") or "").strip(): return info return ordered[0] def _pick_baseline_bind_info( self, candidates: list[dict], config: dict, ) -> dict | None: """选“扫码前当前已绑定角色”作为 baseline。 活动 alias（20260120QYOOB）只反映当前已绑定； cjm 才是换绑最新态。baseline 应优先活动 alias 的已绑定结果， 避免把 cjm 上的待确认新角色误当成扫码前旧角色。 """ if not candidates: return None prefer = [] for key in ("confirm_act_alias", "bind_act_alias", "legacy_act_alias"): alias = str(config.get(key) or "").strip() if alias and alias not in prefer: prefer.append(alias) # 先找“已绑定且有角色”的活动结果 for alias in prefer: for info in candidates: if str(info.get("act_alias") or "") != alias: continue if self._is_bound_act(info) and str(info.get("role_name") or "").strip(): return info # 再退回任意有角色的结果（仍按活动 alias 优先） return self._pick_bind_info( candidates, prefer_pending=False, prefer_aliases=prefer, ) def _wait_bind_role_result( self, client: DouyuActivityClient, db: Session, task: DouyuTask, account: Account, query_aliases: list[str], result: dict, baseline_role_name: str = "", baseline_is_bound_act: bool = False, ) -> tuple[str, dict]: """生成二维码后轮询绑定信息，直到识别到待确认角色、超时或停止。""" deadline = time.monotonic() + DOUYU_BIND_ROLE_POLL_SECONDS result["bind_polling"] = True result["bind_phase"] = result.get("bind_phase") or "waiting_scan" result["bind_ready_for_confirm"] = False result["bind_confirmed"] = False result["baseline_role_name"] = baseline_role_name result["baseline_is_bound_act"] = baseline_is_bound_act result["query_act_aliases"] = query_aliases self._push_log( "info", "开始轮询绑定角色 " f"aliases={','.join(query_aliases) or '-'} " f"baseline={baseline_role_name or '-'} bound={1 if baseline_is_bound_act else 0}", ) self._update_task_progress(db, task, "running", "已生成绑定二维码，等待扫码绑定", result) poll_count = 0 last_summary = "" while not self._stop.is_set() and time.monotonic() < deadline: if self._stop.wait(DOUYU_BIND_ROLE_POLL_INTERVAL): break candidates = self._fetch_bind_info_candidates(client, query_aliases) if not candidates: result["bind_poll_error"] = "所有 actAlias 查询绑定信息失败" self._update_task_progress(db, task, "running", "等待绑定角色同步: 查询失败", result) continue poll_count += 1 bind_info = self._pick_bind_info( candidates, baseline_role_name=baseline_role_name, baseline_is_bound_act=baseline_is_bound_act, prefer_pending=True, prefer_aliases=query_aliases, ) or candidates[0] snapshot = self._bind_snapshot(bind_info) is_pending_role = self._is_pending_role( bind_info, baseline_role_name=baseline_role_name, baseline_is_bound_act=baseline_is_bound_act, ) query_alias = str(bind_info.get("act_alias") or "") summary = ( f"act={query_alias or '-'} " f"{self._format_bind_summary(bind_info, pending=is_pending_role)}" ) # 字段变化或每 3 次打印一次，避免刷屏但仍能看到过程 if summary != last_summary or poll_count == 1 or poll_count % 3 == 0: self._push_log("info", f"轮询绑定#{poll_count}: {summary}") last_summary = summary # 注意：未识别到新角色时，不要把当前已绑定角色写进 role_name， # 否则前端会把旧角色误当成“待确认角色/查询结果”。 if is_pending_role: result.update({ **snapshot, "act_alias": query_alias, "query_act_alias": query_alias, "bind_ready_for_confirm": True, "bind_confirmed": False, "bind_phase": "role_ready", "bind_polling": False, "poll_count": poll_count, "bind_summary": summary, "bind_candidates": [ { "act_alias": item.get("act_alias"), "role_name": item.get("role_name"), "is_bound_act": self._is_bound_act(item), } for item in candidates ], }) role_name = snapshot["role_name"] self._apply_bind_info_to_account(account, bind_info, "game_queried") self._push_log("success", f"识别到待确认角色: {role_name} (act={query_alias})") self._update_task_progress( db, task, "running", f"已识别角色: {role_name}，待确认绑定", result, ) return role_name, result if snapshot["role_name"] and snapshot["is_bound_act"]: result["current_role_name"] = snapshot["role_name"] result["current_area_name"] = snapshot["area_name"] result["current_plat_name"] = snapshot["plat_name"] result.update({ "bind_info": bind_info, "act_alias": query_alias, "query_act_alias": query_alias, "is_bound_act": snapshot["is_bound_act"], "is_bound_role": snapshot["is_bound_role"], "is_bound_account": snapshot["is_bound_account"], "need_bind_act": snapshot["need_bind_act"], "need_bind_role": snapshot["need_bind_role"], "change_role_wait_time": snapshot["change_role_wait_time"], "can_change_role": snapshot["can_change_role"], "role_name": "", "area_name": "", "plat_name": "", "bind_ready_for_confirm": False, "bind_confirmed": False, "bind_phase": "waiting_scan", "bind_polling": True, "poll_count": poll_count, "bind_summary": summary, }) self._update_task_progress(db, task, "running", f"等待扫码绑定 ({summary})", result) result["bind_polling"] = False result["bind_ready_for_confirm"] = False result["bind_phase"] = "stopped" if self._stop.is_set() else "role_timeout" result["poll_count"] = poll_count if last_summary: result["bind_summary"] = last_summary self._push_log( "warning", f"轮询结束 phase={result['bind_phase']} polls={poll_count} last={last_summary or '-'}", ) return "", result def _execute_refresh_goods(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) result = client.list_goods(manual_id=config["manual_id"], rid=config["rid"]) goods = result["goods"] self._upsert_goods(db, goods) account.bind_status = account.bind_status or "active" account.updated_at = datetime.now(timezone.utc) self._mark_task(db, task, "success", f"已刷新商品 {len(goods)} 个", {"goods_count": len(goods), "goods": goods}) def _execute_get_bind_qr(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) qr_act_alias = self._bind_qr_act_alias(config) query_aliases = self._query_bind_act_aliases(config) if not qr_act_alias and not query_aliases: self._mark_task(db, task, "failed", "请先配置绑定活动 actAlias") return if qr_act_alias and qr_act_alias not in query_aliases: query_aliases = [qr_act_alias, *query_aliases] before_candidates = self._fetch_bind_info_candidates(client, query_aliases) if not before_candidates: self._mark_task(db, task, "failed", "查询绑定信息失败，请检查 Cookie 或 actAlias") return # baseline 用活动 alias 的“当前已绑定”；pending 检测优先 cjm 的换绑最新态 before = self._pick_baseline_bind_info(before_candidates, config) or before_candidates[0] # 换绑冷却必须看活动当前绑定（QYOOB），不要用 cjm cooldown_info = self._pick_change_wait_bind_info(before_candidates, config) or before pending_before = self._pick_bind_info( before_candidates, baseline_role_name=str(before.get("role_name") or ""), baseline_is_bound_act=self._is_bound_act(before), prefer_pending=True, prefer_aliases=query_aliases, ) or before before_snapshot = self._bind_snapshot(before) cooldown_snapshot = self._bind_snapshot(cooldown_info) current_role_name = before_snapshot["role_name"] or cooldown_snapshot["role_name"] wait_time = cooldown_snapshot["change_role_wait_time"] self._push_log( "info", "绑定前状态 " f"qr_act={qr_act_alias or '-'} query={','.join(query_aliases)} " f"baseline_hit={before.get('act_alias') or '-'} {self._format_bind_summary(before)} " f"cooldown_hit={cooldown_info.get('act_alias') or '-'} " f"{self._format_bind_summary(cooldown_info)} " f"pending_hit={pending_before.get('act_alias') or '-'} " f"{self._format_bind_summary(pending_before)}", ) # 生成二维码前强制检查换绑冷却：冷却中直接失败，绝不发码 if self._is_change_cooling(cooldown_info): role_label = current_role_name or cooldown_snapshot["role_name"] or "当前角色" wait_text = self._format_wait_time(wait_time) or "冷却中" self._push_log( "warning", f"换绑冷却中，跳过生成二维码 role={role_label} wait={wait_time if wait_time is not None else '-'} " f"can={cooldown_info.get('can_change_role')}", ) result = { "act_alias": qr_act_alias or cooldown_info.get("act_alias") or before.get("act_alias"), "query_act_alias": cooldown_info.get("act_alias") or before.get("act_alias"), "query_act_aliases": query_aliases, "before_bind_info": before, "cooldown_bind_info": cooldown_info, **cooldown_snapshot, "current_role_name": role_label if role_label != "当前角色" else current_role_name, "current_area_name": cooldown_snapshot["area_name"] or before_snapshot["area_name"], "current_plat_name": cooldown_snapshot["plat_name"] or before_snapshot["plat_name"], "change_role_wait_time": wait_time, "change_role_wait_text": self._format_wait_time(wait_time), "bind_ready_for_confirm": False, "bind_confirmed": bool(cooldown_snapshot["is_bound_act"] or before_snapshot["is_bound_act"]), "bind_phase": "change_waiting", "bind_polling": False, } self._apply_bind_info_to_account( account, cooldown_info if cooldown_snapshot["role_name"] else before, "bind_confirmed" if result["bind_confirmed"] else "game_queried", ) self._mark_task( db, task, "failed", f"{role_label} 暂不能换绑，剩余 {wait_text}", result, ) return if not qr_act_alias: self._mark_task(db, task, "failed", "请先配置绑定二维码活动 actAlias") return self._push_log( "info", f"换绑校验通过，开始生成二维码 act={qr_act_alias} " f"role={current_role_name or '-'} wait={wait_time if wait_time is not None else 0}", ) qr_result = client.get_bind_qr(qr_act_alias) result = { **qr_result, "act_alias": qr_act_alias, "query_act_aliases": query_aliases, "before_bind_info": before, "current_role_name": current_role_name, "current_area_name": before_snapshot["area_name"], "current_plat_name": before_snapshot["plat_name"], "role_name": "", "area_name": "", "plat_name": "", "bind_ready_for_confirm": False, "bind_confirmed": False, "bind_phase": "waiting_scan", "bind_polling": True, } account.bind_status = "bind_qr_generated" account.updated_at = datetime.now(timezone.utc) # 关键：先把二维码 progress 出去，前端 running 期间即可弹窗扫码。 self._update_task_progress(db, task, "running", "已生成绑定二维码，等待扫码绑定", result) role_name, result = self._wait_bind_role_result( client, db, task, account, query_aliases, result, baseline_role_name=current_role_name, baseline_is_bound_act=before_snapshot["is_bound_act"], ) if role_name: self._mark_task(db, task, "success", f"已识别角色: {role_name}，待确认绑定", result) return if result.get("bind_phase") == "stopped": self._mark_task(db, task, "stopped", "任务已停止", result) return last_summary = str(result.get("bind_summary") or "") timeout_msg = "已生成绑定二维码，未检测到新扫码角色" if current_role_name: timeout_msg = f"{timeout_msg}（当前仍是 {current_role_name}）" if last_summary: timeout_msg = f"{timeout_msg} | {last_summary}" self._mark_task(db, task, "success", timeout_msg, result) def _execute_confirm_bind(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) confirm_alias = self._confirm_act_alias(config) query_aliases = self._query_bind_act_aliases(config) if confirm_alias and confirm_alias not in query_aliases: query_aliases = [confirm_alias, *query_aliases] if not query_aliases: self._mark_task(db, task, "failed", "请先配置绑定活动 actAlias") return before_candidates = self._fetch_bind_info_candidates(client, query_aliases) if not before_candidates: self._mark_task(db, task, "failed", "查询绑定信息失败，请检查 Cookie 或 actAlias") return before = self._pick_bind_info( before_candidates, baseline_role_name="", baseline_is_bound_act=False, prefer_pending=True, prefer_aliases=query_aliases, ) or before_candidates[0] before_snapshot = self._bind_snapshot(before) role_name = before_snapshot["role_name"] query_alias = str(before.get("act_alias") or "") self._push_log( "info", f"确认前状态 confirm_act={confirm_alias or '-'} hit={query_alias or '-'} " f"{self._format_bind_summary(before)}", ) if not role_name: self._mark_task( db, task, "failed", "尚未识别到待确认角色，请先扫码完成绑定", { "act_alias": confirm_alias or query_alias, "query_act_alias": query_alias, "query_act_aliases": query_aliases, "before_bind_info": before, **before_snapshot, "bind_ready_for_confirm": False, "bind_confirmed": False, "bind_phase": "waiting_role", }, ) return if before_snapshot["is_bound_act"]: self._apply_bind_info_to_account(account, before, "bind_confirmed") self._mark_task( db, task, "success", f"已绑定: {role_name}", { "act_alias": confirm_alias or query_alias, "query_act_alias": query_alias, "query_act_aliases": query_aliases, "before_bind_info": before, **before_snapshot, "bind_ready_for_confirm": True, "bind_confirmed": True, "bind_phase": "confirmed", }, ) return # 确认接口优先用配置的确认 alias；没有则回退到命中查询的 alias use_confirm_alias = confirm_alias or query_alias confirm_result = client.confirm_bind(use_confirm_alias) try: after_candidates = self._fetch_bind_info_candidates(client, query_aliases) after = self._pick_bind_info( after_candidates, baseline_role_name="", baseline_is_bound_act=False, prefer_pending=False, ) if after_candidates else None if after is None: raise DouyuActivityError("确认后回查绑定信息失败") except DouyuActivityError as exc: # 确认接口已成功时，回查失败仍按确认成功处理，但保留错误信息。 self._apply_bind_info_to_account(account, before, "bind_confirmed") self._mark_task( db, task, "success", f"绑定成功: {role_name}", { "act_alias": use_confirm_alias, "query_act_alias": query_alias, "query_act_aliases": query_aliases, "before_bind_info": before, "confirm": confirm_result, **before_snapshot, "bind_ready_for_confirm": True, "bind_confirmed": True, "bind_phase": "confirmed", "refresh_error": str(exc), }, ) return after_snapshot = self._bind_snapshot(after) final_info = after if after_snapshot["role_name"] else before final_snapshot = after_snapshot if after_snapshot["role_name"] else before_snapshot final_role_name = final_snapshot["role_name"] or role_name if not after_snapshot["is_bound_act"]: self._apply_bind_info_to_account(account, final_info, "game_queried") self._mark_task( db, task, "failed", f"确认绑定未生效: {final_role_name}", { "act_alias": use_confirm_alias, "query_act_alias": after.get("act_alias") or query_alias, "query_act_aliases": query_aliases, "before_bind_info": before, "confirm": confirm_result, **final_snapshot, "bind_ready_for_confirm": True, "bind_confirmed": False, "bind_phase": "confirm_failed", "after_bind_info": after, }, ) return self._apply_bind_info_to_account(account, final_info, "bind_confirmed") self._mark_task( db, task, "success", f"绑定成功: {final_role_name}", { "act_alias": use_confirm_alias, "query_act_alias": after.get("act_alias") or query_alias, "query_act_aliases": query_aliases, "before_bind_info": before, "confirm": confirm_result, "after_bind_info": after, **final_snapshot, "bind_ready_for_confirm": True, "bind_confirmed": True, "bind_phase": "confirmed", }, ) def _execute_create_elite_qr(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) ctn = str(self._task_payload(task).get("ctn") or "") if not ctn: ctn = client.acf_ccn(refresh_subscribe=True) result = client.create_elite_qr( ctn=ctn, act_alias=self._current_bind_act_alias(config) or str(config["confirm_act_alias"]), amount=int(config["elite_amount"]), room_id=str(config["room_id"]), ) account.bind_status = "elite_qr_created" account.updated_at = datetime.now(timezone.utc) self._mark_task(db, task, "success", "精英宝典支付码已生成", result) 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) client = self._client(cookie) 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._mark_task(db, task, "success", f"鱼翅 {amount} 元支付码已生成", 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) client = self._client(cookie) result = client.donate_elite_gift( gift_count=gift_count, room_id=str(payload.get("room_id") or config["room_id"]), gift_id=str(payload.get("gift_id") or config["gift_id"]), skin_id=str(payload.get("skin_id") or config["skin_id"]), ) account.bind_status = "gift_donated" account.updated_at = datetime.now(timezone.utc) self._mark_task(db, task, "success", f"赠送精英令成功: {gift_count}", result) def _execute_query_points(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) ctn = client.acf_ccn(refresh_subscribe=False) uid = account_uid(account, cookie) if not uid: self._mark_task(db, task, "failed", "Cookie 中没有 acf_uid，无法查询积分") return result = client.query_points(uid=uid, ctn=ctn) points = self._to_int(result.get("points")) account.uid = uid account.points = points update_account_profile_from_cookie(account, cookie) account.bind_status = "points_queried" account.updated_at = datetime.now(timezone.utc) self._mark_task(db, task, "success", f"积分: {points if points is not None else '-'}", result) def _execute_exchange_goods(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): import time as time_mod payload = self._task_payload(task) commodity_id = str(payload.get("commodity_id") or payload.get("commodityId") or "").strip() if not commodity_id: self._mark_task(db, task, "failed", "请选择兑换商品") return client = self._client(cookie) ctn = client.acf_ccn(refresh_subscribe=False) result = None last_error = "" for attempt in range(8 + 1): if self._stop.is_set(): self._mark_task(db, task, "stopped", "任务已停止") return try: result = client.exchange_goods( manual_id=str(config["manual_id"]), rid=str(config["rid"]), commodity_id=commodity_id, ctn=ctn, ) break except DouyuActivityError as exc: last_error = str(exc) if attempt >= 8: self._mark_task(db, task, "failed", f"兑换失败(已重试{attempt}次): {last_error}") return error_lower = last_error.lower() if any(kw in error_lower for kw in ("无效", "太快", "csrf")): self._push_log("info", f" 重试 {attempt + 1}/8: {last_error}，刷新 csrf_token...") try: token = client.csrf_token() self._push_log("debug", f" csrf_token 已刷新: {token[:12]}...") except Exception: pass else: self._push_log("info", f" 重试 {attempt + 1}/8: {last_error}") time_mod.sleep(0.3) if result is None: self._mark_task(db, task, "failed", f"兑换失败: {last_error}") return goods = ( db.query(DouyuGoodsSnapshot) .filter(DouyuGoodsSnapshot.commodity_id == commodity_id) .first() ) account.bind_status = "goods_exchanged" account.updated_at = datetime.now(timezone.utc) self._mark_task( db, task, "success", f"兑换成功: {(goods.name if goods else '') or commodity_id}", {"goods": goods.raw if goods else None, **result}, ) def _execute_query_game_name(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) query_aliases = self._query_bind_act_aliases(config) if not query_aliases: self._mark_task(db, task, "failed", "请先配置绑定活动 actAlias") return candidates = self._fetch_bind_info_candidates(client, query_aliases) if not candidates: self._mark_task(db, task, "failed", "查询绑定信息失败，请检查 Cookie 或 actAlias") return bind_info = self._pick_bind_info( candidates, prefer_pending=True, prefer_aliases=query_aliases, ) or candidates[0] snapshot = self._bind_snapshot(bind_info) role_name = snapshot["role_name"] source_alias = str(bind_info.get("act_alias") or "") summary = f"act={source_alias or '-'} {self._format_bind_summary(bind_info)}" self._push_log( "info", "查询角色 " f"aliases={','.join(query_aliases)} hit={source_alias or '-'} " f"{self._format_bind_summary(bind_info)}", ) self._apply_bind_info_to_account(account, bind_info, "game_queried" if role_name else "game_not_bound") pending = bool(role_name) and not snapshot["is_bound_act"] result = { "act_alias": source_alias, "query_act_alias": source_alias, "query_act_aliases": query_aliases, **snapshot, "bind_ready_for_confirm": pending, "bind_confirmed": snapshot["is_bound_act"], "bind_phase": ( "confirmed" if snapshot["is_bound_act"] else ("role_ready" if role_name else "waiting_role") ), "bind_summary": summary, "bind_candidates": [ { "act_alias": item.get("act_alias"), "role_name": item.get("role_name"), "is_bound_act": self._is_bound_act(item), } for item in candidates ], } if not role_name: message = f"未获取到游戏名 | {summary}" elif snapshot["is_bound_act"]: message = f"当前已绑定: {role_name} | {summary}" else: message = f"待确认角色: {role_name} | {summary}" self._mark_task(db, task, "success" if role_name else "failed", message, result) def _pick_change_wait_bind_info(self, candidates: list[dict], config: dict) -> dict | None: """换绑倒计时优先看活动当前绑定（QYOOB），不是 cjm 换绑最新态。""" if not candidates: return None # 1) 优先活动 alias 上有角色的结果 baseline = self._pick_baseline_bind_info(candidates, config) if baseline is not None: wait = self._to_int(baseline.get("change_role_wait_time")) if wait is not None: return baseline # 2) 任意带 wait_time 的结果里取 wait 最大的 with_wait = [] for info in candidates: wait = self._to_int(info.get("change_role_wait_time")) if wait is not None: with_wait.append((wait, info)) if with_wait: with_wait.sort(key=lambda item: item[0], reverse=True) return with_wait[0][1] # 3) 回退 baseline / 首个有角色 return baseline or self._pick_bind_info( candidates, prefer_pending=False, prefer_aliases=[ str(config.get("confirm_act_alias") or "").strip(), str(config.get("bind_act_alias") or "").strip(), str(config.get("legacy_act_alias") or "").strip(), ], ) def _execute_query_change_bind_time(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) query_aliases = self._query_bind_act_aliases(config) if not query_aliases: self._mark_task(db, task, "failed", "请先配置绑定活动 actAlias") return candidates = self._fetch_bind_info_candidates(client, query_aliases) if not candidates: self._mark_task(db, task, "failed", "查询绑定信息失败，请检查 Cookie 或 actAlias") return # 换绑时间看“当前已绑定”活动态，不要优先 cjm（cjm 常无 changeRoleWaitTime） bind_info = self._pick_change_wait_bind_info(candidates, config) or candidates[0] snapshot = self._bind_snapshot(bind_info) wait_time = snapshot["change_role_wait_time"] account.change_role_wait_time = wait_time account.bind_status = "change_time_queried" account.updated_at = datetime.now(timezone.utc) source_alias = str(bind_info.get("act_alias") or "") wait_text = self._format_wait_time(wait_time) can_change = snapshot["can_change_role"] role_name = snapshot["role_name"] or "-" if wait_time is None: if can_change is False: status_text = "不可换绑（接口未返回倒计时）" elif can_change is True: status_text = "可换绑" else: status_text = "未返回换绑倒计时" elif wait_time <= 0: status_text = "可换绑" wait_text = "0" else: status_text = f"剩余 {wait_text}" result = { "act_alias": source_alias, "query_act_alias": source_alias, "query_act_aliases": query_aliases, **snapshot, "change_role_wait_text": wait_text, "bind_ready_for_confirm": bool(snapshot["role_name"]) and not snapshot["is_bound_act"], "bind_confirmed": snapshot["is_bound_act"], "bind_summary": f"act={source_alias or '-'} {self._format_bind_summary(bind_info)}", "bind_candidates": [ { "act_alias": item.get("act_alias"), "role_name": item.get("role_name"), "is_bound_act": self._is_bound_act(item), "change_role_wait_time": self._to_int(item.get("change_role_wait_time")), "can_change_role": item.get("can_change_role"), } for item in candidates ], } self._push_log( "info", f"查询换绑时间 hit={source_alias or '-'} role={role_name} " f"wait={wait_time if wait_time is not None else '-'} can={can_change}", ) self._mark_task( db, task, "success", f"{role_name} {status_text} | act={source_alias or '-'}", result, ) def _execute_query_limited_goods(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) result = client.query_limited_goods(manual_id=str(config["manual_id"]), rid=str(config["rid"])) limited = result["limited_goods"] names = [str(item.get("commodityName") or "") for item in limited if item.get("commodityName")] message = "无限制商品" if not names else f"限兑 {len(names)} 个: {', '.join(names[:5])}" account.bind_status = "limited_goods_queried" account.updated_at = datetime.now(timezone.utc) self._mark_task(db, task, "success", message, {"limited_count": len(limited), "limited_goods": limited}) def _execute_query_gold_balance(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) gold = client.gold_account() exchange = client.exchange_balance() account.gold_balance = self._to_int(gold.get("gold")) account.exchange_balance = self._to_int(exchange.get("count")) account.bind_status = "gold_balance_queried" account.updated_at = datetime.now(timezone.utc) self._mark_task( db, task, "success", f"鱼翅余额: {account.gold_balance if account.gold_balance is not None else '-'}", {"gold": gold, "exchange_balance": exchange}, ) def _execute_query_exchange_records(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) result = client.exchange_records(manual_id=str(config["manual_id"])) records = result["records"] account.bind_status = "exchange_records_queried" account.updated_at = datetime.now(timezone.utc) self._mark_task(db, task, "success", f"兑换记录 {len(records)} 条" if records else "暂无兑换记录", result) def _execute_prefetch_csrf_token(self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict): client = self._client(cookie) token = client.csrf_token() account.bind_status = "csrf_token_ready" account.updated_at = datetime.now(timezone.utc) self._mark_task(db, task, "success", "获取 csrf_token 成功", {"csrf_token": token, "cookie": client.cookie}) def _execute_one(self, task_id: int, config: dict, total: int): worker_db = SessionLocal() try: task = ( worker_db.query(DouyuTask) .options(joinedload(DouyuTask.account)) .filter(DouyuTask.id == task_id) .first() ) if not task or self._stop.is_set(): return account = task.account task.status = "running" task.message = "执行中" worker_db.commit() with self._counter_lock: self._started += 1 current = self._started self._push_log("info", f"[{current}/{total}] 开始: {self._account_name(account)}") cookie = latest_success_cookie(worker_db, account.id) if not cookie: self._mark_task(worker_db, task, "failed", "账号没有成功登录 Cookie") self._push_log("warning", f"[{current}] {self._account_name(account)} 无 Cookie") return update_account_profile_from_cookie(account, cookie) handler = { "refresh_goods": self._execute_refresh_goods, "get_bind_qr": self._execute_get_bind_qr, "confirm_bind": self._execute_confirm_bind, "create_elite_qr": self._execute_create_elite_qr, "create_gold_qr": self._execute_create_gold_qr, "donate_elite_gift": self._execute_donate_elite_gift, "query_points": self._execute_query_points, "exchange_goods": self._execute_exchange_goods, "query_game_name": self._execute_query_game_name, "query_change_bind_time": self._execute_query_change_bind_time, "query_limited_goods": self._execute_query_limited_goods, "query_gold_balance": self._execute_query_gold_balance, "query_exchange_records": self._execute_query_exchange_records, "prefetch_csrf_token": self._execute_prefetch_csrf_token, }.get(task.task_type) if handler is None: self._mark_task(worker_db, task, "failed", "不支持的任务类型") return handler(worker_db, task, account, cookie, config) self._push_log("success", f"[{current}] {self._account_name(account)} {task.message}") except DouyuActivityError as exc: if "task" in locals() and task: self._mark_task(worker_db, task, "failed", str(exc)) self._push_log("warning", f"斗鱼任务失败: {exc}") except Exception as exc: if "task" in locals() and task: self._mark_task(worker_db, task, "error", str(exc)) self._push_log("error", f"斗鱼任务异常: {exc}") finally: worker_db.close() def run(self): """执行批次任务。""" self._push_log("info", f"斗鱼任务批次 {self.batch_id} 开始") try: config = self._config_info(self.db) tasks = ( self.db.query(DouyuTask) .filter(DouyuTask.batch_id == self.batch_id, DouyuTask.status == "planned") .order_by(DouyuTask.id.asc()) .all() ) if not tasks: self._push_log("warning", "没有可执行的斗鱼任务") self._push_log("result", "") return for task in tasks: task.status = "pending" task.message = "等待执行" self.db.commit() total = len(tasks) with ThreadPoolExecutor(max_workers=self.concurrency) as executor: futures = [] for task in tasks: if self._stop.is_set(): break futures.append(executor.submit(self._execute_one, task.id, config, total)) for future in as_completed(futures): try: future.result() except Exception as exc: self._push_log("error", f"Worker 异常: {exc}") if self._stop.is_set(): self._push_log("warning", f"斗鱼任务批次 {self.batch_id} 已停止") else: self._push_log("info", f"斗鱼任务批次 {self.batch_id} 完成") self._push_log("result", "") finally: self.db.close() class DouyuBatchRegistry: """管理运行中的斗鱼任务批次。""" def __init__(self): self._batches: dict[str, dict] = {} def register(self, batch_id: str, log_queue: asyncio.Queue, loop: asyncio.AbstractEventLoop, runner: DouyuBatchRunner): self._batches[batch_id] = { "log_queue": log_queue, "loop": loop, "runner": runner, "finished": False, "updated_at": time.time(), } def get(self, batch_id: str): return self._batches.get(batch_id) def pop(self, batch_id: str): return self._batches.pop(batch_id, None) def mark_finished(self, batch_id: str): if batch_id in self._batches: self._batches[batch_id]["finished"] = True self._batches[batch_id]["updated_at"] = time.time() def active_ids(self) -> set[str]: return { batch_id for batch_id, info in self._batches.items() if not info.get("finished") } douyu_batch_registry = DouyuBatchRegistry()