1981 lines
82 KiB
Python
1981 lines
82 KiB
Python
"""斗鱼活动任务批次执行器。"""
|
||
|
||
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
|
||
DOUYU_PAYMENT_POLL_SECONDS = 600
|
||
DOUYU_PAYMENT_POLL_INTERVAL = 5
|
||
DOUYU_GIFT_POINTS_REFRESH_TIMES = 3
|
||
DOUYU_GIFT_POINTS_REFRESH_INTERVAL = 2
|
||
|
||
|
||
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 _action_act_alias(config: dict, key: str) -> str:
|
||
"""动作类接口用的活动 alias，排除只用于查询最新角色的 legacy alias。"""
|
||
alias = str(config.get(key) or "").strip()
|
||
query_only_alias = str(config.get("legacy_act_alias") or "").strip()
|
||
if not alias:
|
||
return ""
|
||
if alias in {query_only_alias, DOUYU_LEGACY_BIND_ACT_ALIAS}:
|
||
return ""
|
||
return alias
|
||
|
||
@classmethod
|
||
def _bind_qr_act_alias(cls, config: dict) -> str:
|
||
"""生成绑定二维码用的活动 alias。"""
|
||
return cls._action_act_alias(config, "bind_act_alias") or cls._action_act_alias(config, "confirm_act_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
|
||
|
||
@classmethod
|
||
def _confirm_act_alias(cls, config: dict) -> str:
|
||
"""确认绑定接口用的活动 alias。"""
|
||
return cls._action_act_alias(config, "confirm_act_alias") or cls._action_act_alias(config, "bind_act_alias")
|
||
|
||
# 兼容旧调用名
|
||
@classmethod
|
||
def _current_bind_act_alias(cls, config: dict) -> str:
|
||
return cls._confirm_act_alias(config) or 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 _apply_esports_bind_info_to_account(self, account: Account, bind_info: dict, status: str) -> None:
|
||
"""将电竞手册角色状态写入专属字段，避免覆盖精英宝典数据。"""
|
||
role_name = str(bind_info.get("role_name") or "")
|
||
account.esports_game_name = role_name or account.esports_game_name
|
||
account.esports_game_channel = self._role_channel(bind_info) or account.esports_game_channel
|
||
account.esports_change_role_wait_time = self._to_int(bind_info.get("change_role_wait_time"))
|
||
account.esports_bind_status = status
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
|
||
def _esports_bind_state(
|
||
self,
|
||
client: DouyuActivityClient,
|
||
act_alias: str,
|
||
) -> dict:
|
||
"""查询电竞手册活动状态，接口同时返回当前角色和换绑冷却信息。"""
|
||
activity_info = client.esports_bind_info(act_alias)
|
||
activity_snapshot = self._bind_snapshot(activity_info)
|
||
esports_bound = activity_snapshot["is_bound_act"]
|
||
has_selected_role = bool(activity_snapshot["role_name"])
|
||
tx_act = activity_info.get("tx_act") or {}
|
||
return {
|
||
**activity_snapshot,
|
||
"is_bound_role": has_selected_role,
|
||
"act_alias": act_alias,
|
||
"game_id": tx_act.get("gameId") or "",
|
||
"esports_bound": esports_bound,
|
||
"has_selected_role": has_selected_role,
|
||
"bind_ready_for_confirm": has_selected_role and not esports_bound,
|
||
"bind_confirmed": esports_bound,
|
||
"bind_phase": "confirmed" if esports_bound else ("role_ready" if has_selected_role else "waiting_role"),
|
||
"activity_bind_info": activity_info,
|
||
"activity_bind_snapshot": activity_snapshot,
|
||
"role_source": "activity",
|
||
}
|
||
|
||
@staticmethod
|
||
def _esports_role_text(state: dict) -> str:
|
||
role_name = str(state.get("role_name") or "")
|
||
channel = " / ".join(
|
||
str(part)
|
||
for part in [state.get("plat_name"), state.get("area_name")]
|
||
if part
|
||
)
|
||
if not role_name:
|
||
return ""
|
||
return f"{role_name}（{channel}）" if channel else role_name
|
||
|
||
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 _refresh_account_points(
|
||
self,
|
||
client: DouyuActivityClient,
|
||
account: Account,
|
||
cookie: str,
|
||
*,
|
||
ctn: str | None = None,
|
||
) -> dict:
|
||
"""刷新账号积分并写回账号表。"""
|
||
uid = account_uid(account, cookie)
|
||
if not uid:
|
||
raise DouyuActivityError("Cookie 中没有 acf_uid，无法查询积分")
|
||
ctn_value = ctn or client.acf_ccn(refresh_subscribe=False)
|
||
result = client.query_points(uid=uid, ctn=ctn_value)
|
||
points = self._to_int(result.get("points"))
|
||
account.uid = uid
|
||
account.points = points
|
||
update_account_profile_from_cookie(account, cookie)
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
return {"points": points, "points_query": result}
|
||
|
||
def _refresh_account_gold_balance(self, client: DouyuActivityClient, account: Account) -> dict:
|
||
"""刷新鱼翅和钱包兑换余额并写回账号表。"""
|
||
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.updated_at = datetime.now(timezone.utc)
|
||
return {
|
||
"gold_balance": account.gold_balance,
|
||
"exchange_balance": account.exchange_balance,
|
||
"gold": gold,
|
||
"exchange_balance_query": exchange,
|
||
}
|
||
|
||
def _wait_points_after_payment(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
client: DouyuActivityClient,
|
||
cookie: str,
|
||
ctn: str,
|
||
result: dict,
|
||
) -> bool:
|
||
"""等待宝典支付到账；积分达到 300 视为开通成功。"""
|
||
deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
|
||
result["payment_polling"] = True
|
||
result["payment_target_points"] = 300
|
||
poll_count = 0
|
||
last_points = None
|
||
while not self._stop.is_set() and time.monotonic() <= deadline:
|
||
try:
|
||
points_result = self._refresh_account_points(client, account, cookie, ctn=ctn)
|
||
db.commit()
|
||
poll_count += 1
|
||
last_points = points_result["points"]
|
||
result.update(points_result)
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_polling"] = True
|
||
if last_points is not None and last_points >= 300:
|
||
result["payment_polling"] = False
|
||
result["elite_opened"] = True
|
||
return True
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
f"精英宝典支付码已生成，等待开通到账（当前积分 {last_points if last_points is not None else '-'}）",
|
||
result,
|
||
)
|
||
except Exception as exc:
|
||
poll_count += 1
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_poll_error"] = str(exc)
|
||
self._update_task_progress(db, task, "running", f"等待开通到账: {exc}", result)
|
||
if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
|
||
break
|
||
result["payment_polling"] = False
|
||
result["elite_opened"] = False
|
||
result["points"] = last_points
|
||
return False
|
||
|
||
def _refresh_esports_handbook(
|
||
self,
|
||
client: DouyuActivityClient,
|
||
account: Account,
|
||
*,
|
||
manual_id: str,
|
||
) -> dict:
|
||
"""刷新电竞手册开通状态并将积分写回账号。"""
|
||
result = client.esports_user_info(manual_id=manual_id)
|
||
manual_type = self._to_int(result.get("manual_type"))
|
||
manual_score = self._to_int(result.get("manual_score"))
|
||
account.esports_points = manual_score
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
return {
|
||
"esports_manual_type": manual_type,
|
||
"esports_manual_score": manual_score,
|
||
"esports_expire_time": result.get("expire_time"),
|
||
"esports_user_info": result,
|
||
"esports_points": manual_score,
|
||
"points": manual_score,
|
||
}
|
||
|
||
def _wait_esports_open_after_payment(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
client: DouyuActivityClient,
|
||
*,
|
||
manual_id: str,
|
||
result: dict,
|
||
baseline_manual_type: int | None,
|
||
baseline_manual_score: int | None,
|
||
) -> bool:
|
||
"""等待电竞手册支付到账，以 manualType=1 或积分变化作为成功条件。"""
|
||
deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
|
||
result["payment_polling"] = True
|
||
result["esports_manual_type_baseline"] = baseline_manual_type
|
||
result["esports_manual_score_baseline"] = baseline_manual_score
|
||
poll_count = 0
|
||
last_manual_type = baseline_manual_type
|
||
last_manual_score = baseline_manual_score
|
||
|
||
while not self._stop.is_set() and time.monotonic() <= deadline:
|
||
try:
|
||
handbook_result = self._refresh_esports_handbook(
|
||
client,
|
||
account,
|
||
manual_id=manual_id,
|
||
)
|
||
db.commit()
|
||
poll_count += 1
|
||
last_manual_type = handbook_result["esports_manual_type"]
|
||
last_manual_score = handbook_result["esports_manual_score"]
|
||
result.update(handbook_result)
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_polling"] = True
|
||
opened = (
|
||
last_manual_type is not None
|
||
and last_manual_type >= 1
|
||
) or (
|
||
baseline_manual_score is not None
|
||
and last_manual_score is not None
|
||
and last_manual_score > baseline_manual_score
|
||
)
|
||
if opened:
|
||
result["payment_polling"] = False
|
||
result["esports_opened"] = True
|
||
return True
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
"电竞手册支付码已生成，等待开通到账"
|
||
f"（类型 {last_manual_type if last_manual_type is not None else '-'}，"
|
||
f"积分 {last_manual_score if last_manual_score is not None else '-'}）",
|
||
result,
|
||
)
|
||
except Exception as exc:
|
||
poll_count += 1
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_poll_error"] = str(exc)
|
||
self._update_task_progress(db, task, "running", f"等待电竞手册到账: {exc}", result)
|
||
if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
|
||
break
|
||
|
||
result["payment_polling"] = False
|
||
result["esports_opened"] = False
|
||
result["esports_manual_type"] = last_manual_type
|
||
result["esports_manual_score"] = last_manual_score
|
||
result["esports_points"] = last_manual_score
|
||
result["points"] = last_manual_score
|
||
return False
|
||
|
||
def _wait_gold_balance_after_payment(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
client: DouyuActivityClient,
|
||
result: dict,
|
||
baseline_gold: int | None,
|
||
) -> bool:
|
||
"""等待鱼翅充值到账；余额变化后写回账号表。"""
|
||
deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
|
||
result["payment_polling"] = True
|
||
result["baseline_gold_balance"] = baseline_gold
|
||
poll_count = 0
|
||
last_gold = baseline_gold
|
||
baseline_ready = baseline_gold is not None
|
||
while not self._stop.is_set() and time.monotonic() <= deadline:
|
||
try:
|
||
balance_result = self._refresh_account_gold_balance(client, account)
|
||
db.commit()
|
||
poll_count += 1
|
||
last_gold = balance_result["gold_balance"]
|
||
result.update(balance_result)
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_polling"] = True
|
||
if not baseline_ready and last_gold is not None:
|
||
baseline_gold = last_gold
|
||
result["baseline_gold_balance"] = baseline_gold
|
||
baseline_ready = True
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
f"鱼翅支付码已生成，已记录当前余额 {last_gold}，等待到账",
|
||
result,
|
||
)
|
||
if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
|
||
break
|
||
continue
|
||
changed = last_gold is not None and (baseline_gold is None or last_gold != baseline_gold)
|
||
if changed:
|
||
result["payment_polling"] = False
|
||
result["gold_recharged"] = True
|
||
return True
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
f"鱼翅支付码已生成，等待到账（当前鱼翅 {last_gold if last_gold is not None else '-'}）",
|
||
result,
|
||
)
|
||
except Exception as exc:
|
||
poll_count += 1
|
||
result["payment_poll_count"] = poll_count
|
||
result["payment_poll_error"] = str(exc)
|
||
self._update_task_progress(db, task, "running", f"等待鱼翅到账: {exc}", result)
|
||
if self._stop.wait(DOUYU_PAYMENT_POLL_INTERVAL):
|
||
break
|
||
result["payment_polling"] = False
|
||
result["gold_recharged"] = False
|
||
result["gold_balance"] = last_gold
|
||
return False
|
||
|
||
def _refresh_points_after_elite_gift(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
client: DouyuActivityClient,
|
||
cookie: str,
|
||
ctn: str | None,
|
||
result: dict,
|
||
baseline_points: int | None,
|
||
gift_count: int,
|
||
) -> dict:
|
||
"""赠送精英令后短轮询积分；1 个精英令约等于 10 积分。"""
|
||
expected_gain = max(0, gift_count) * 10
|
||
target_points = baseline_points + expected_gain if baseline_points is not None else None
|
||
result["gift_points_baseline"] = baseline_points
|
||
result["gift_points_expected_gain"] = expected_gain
|
||
result["gift_points_target"] = target_points
|
||
|
||
last_points = None
|
||
refresh_result: dict = {}
|
||
for index in range(1, DOUYU_GIFT_POINTS_REFRESH_TIMES + 1):
|
||
refresh_result = self._refresh_account_points(client, account, cookie, ctn=ctn)
|
||
db.commit()
|
||
last_points = refresh_result["points"]
|
||
result.update(refresh_result)
|
||
result["gift_points_refresh_count"] = index
|
||
if target_points is None or (last_points is not None and last_points >= target_points):
|
||
result["gift_points_confirmed"] = target_points is None or last_points is not None
|
||
return refresh_result
|
||
if index < DOUYU_GIFT_POINTS_REFRESH_TIMES:
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
f"赠送精英令成功，等待积分同步（当前 {last_points if last_points is not None else '-'}，预期 {target_points}）",
|
||
result,
|
||
)
|
||
if self._stop.wait(DOUYU_GIFT_POINTS_REFRESH_INTERVAL):
|
||
break
|
||
|
||
result["gift_points_confirmed"] = False
|
||
result["points"] = last_points
|
||
return refresh_result
|
||
|
||
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)
|
||
act_alias = self._confirm_act_alias(config) or self._bind_qr_act_alias(config)
|
||
if not act_alias:
|
||
self._mark_task(db, task, "failed", "请先配置开通宝典活动 actAlias")
|
||
return
|
||
result = client.create_elite_qr(
|
||
ctn=ctn,
|
||
act_alias=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._update_task_progress(db, task, "running", "精英宝典支付码已生成，等待开通到账", result)
|
||
opened = self._wait_points_after_payment(db, task, account, client, cookie, ctn, result)
|
||
if self._stop.is_set():
|
||
self._mark_task(db, task, "stopped", "任务已停止", result)
|
||
return
|
||
if opened:
|
||
account.bind_status = "elite_opened"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
self._mark_task(db, task, "success", f"精英宝典已开通，积分: {account.points}", result)
|
||
return
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"failed",
|
||
f"未检测到精英宝典开通到账，当前积分: {account.points if account.points is not None else '-'}",
|
||
result,
|
||
)
|
||
|
||
def _execute_prepare_esports_bind(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
cookie: str,
|
||
config: dict,
|
||
):
|
||
"""打开电竞手册绑定面板：查活动状态、当前角色和换绑冷却。"""
|
||
client = self._client(cookie)
|
||
act_alias = str(config.get("esports_act_alias") or "").strip()
|
||
if not act_alias:
|
||
self._mark_task(db, task, "failed", "请先配置电竞手册活动 actAlias")
|
||
return
|
||
|
||
state = self._esports_bind_state(client, act_alias)
|
||
esports_bound = state["esports_bound"]
|
||
role_text = self._esports_role_text(state)
|
||
self._apply_esports_bind_info_to_account(
|
||
account,
|
||
state,
|
||
"esports_bound" if esports_bound else ("esports_bind_ready" if role_text else "game_not_bound"),
|
||
)
|
||
# 二维码用于重新选择角色，不应因为已有角色或换绑冷却而隐藏。
|
||
# 冷却是否允许最终由 actBind 返回结果决定。
|
||
qr_result = client.get_esports_bind_qr(act_alias)
|
||
result = {
|
||
**state,
|
||
**qr_result,
|
||
"esports_bind_dialog": True,
|
||
"can_open_role_selector": True,
|
||
}
|
||
if esports_bound:
|
||
message = f"电竞手册已绑定: {role_text or '-'}，可扫码换绑"
|
||
elif role_text:
|
||
message = f"当前角色: {role_text}，可扫码切换角色或直接完成绑定"
|
||
else:
|
||
message = "请扫码选择游戏角色，完成后查询最新角色"
|
||
self._mark_task(db, task, "success", message, result)
|
||
|
||
def _execute_get_esports_bind_qr(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
cookie: str,
|
||
config: dict,
|
||
):
|
||
"""生成电竞手册切换角色用的腾讯入口。"""
|
||
client = self._client(cookie)
|
||
act_alias = str(config.get("esports_act_alias") or "").strip()
|
||
if not act_alias:
|
||
self._mark_task(db, task, "failed", "请先配置电竞手册活动 actAlias")
|
||
return
|
||
|
||
state = self._esports_bind_state(client, act_alias)
|
||
qr_result = client.get_esports_bind_qr(act_alias)
|
||
result = {
|
||
**state,
|
||
**qr_result,
|
||
"esports_bind_dialog": True,
|
||
"can_open_role_selector": True,
|
||
"bind_phase": "switching_role",
|
||
}
|
||
account.esports_bind_status = "esports_role_switching"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"success",
|
||
"请在腾讯页面选择角色，返回后查询最新角色",
|
||
result,
|
||
)
|
||
|
||
def _execute_confirm_esports_bind(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
cookie: str,
|
||
config: dict,
|
||
):
|
||
"""通过 actBind 确认电竞手册绑定，并回读唯一状态接口确认结果。"""
|
||
client = self._client(cookie)
|
||
act_alias = str(config.get("esports_act_alias") or "").strip()
|
||
if not act_alias:
|
||
self._mark_task(db, task, "failed", "请先配置电竞手册活动 actAlias")
|
||
return
|
||
|
||
account.esports_bind_status = "esports_bind_confirming"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
self._update_task_progress(
|
||
db,
|
||
task,
|
||
"running",
|
||
"正在确认电竞手册绑定",
|
||
{
|
||
"bind_phase": "confirming",
|
||
"bind_confirmed": False,
|
||
},
|
||
)
|
||
confirm_result = client.confirm_esports_bind(
|
||
act_alias,
|
||
room_id=str(config.get("room_id") or "9263298"),
|
||
)
|
||
after_state = self._esports_bind_state(client, act_alias)
|
||
result = {
|
||
**after_state,
|
||
"confirm": confirm_result,
|
||
"bind_phase": "confirmed" if after_state["esports_bound"] else "confirm_failed",
|
||
}
|
||
if not after_state["esports_bound"]:
|
||
self._apply_esports_bind_info_to_account(account, after_state, "esports_bind_ready")
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"failed",
|
||
f"电竞手册绑定未生效: {self._esports_role_text(after_state) or '-'}",
|
||
result,
|
||
)
|
||
return
|
||
|
||
self._apply_esports_bind_info_to_account(account, after_state, "esports_bound")
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"success",
|
||
f"电竞手册绑定成功: {self._esports_role_text(after_state) or '-'}",
|
||
result,
|
||
)
|
||
|
||
def _execute_query_esports_game_name(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
account: Account,
|
||
cookie: str,
|
||
config: dict,
|
||
):
|
||
"""查询电竞手册活动返回的最新角色和换绑冷却状态。"""
|
||
client = self._client(cookie)
|
||
act_alias = str(config.get("esports_act_alias") or "").strip()
|
||
if not act_alias:
|
||
self._mark_task(db, task, "failed", "请先配置电竞手册活动 actAlias")
|
||
return
|
||
|
||
state = self._esports_bind_state(client, act_alias)
|
||
role_name = state["role_name"]
|
||
is_bound = state["esports_bound"]
|
||
self._apply_esports_bind_info_to_account(
|
||
account,
|
||
state,
|
||
"esports_bound" if is_bound else ("esports_bind_ready" if role_name else "game_not_bound"),
|
||
)
|
||
result = {
|
||
**state,
|
||
"esports_bind_dialog": True,
|
||
"can_open_role_selector": True,
|
||
}
|
||
if role_name:
|
||
message = f"最新角色: {self._esports_role_text(state)}"
|
||
else:
|
||
message = "未查询到游戏角色，请先切换角色"
|
||
self._mark_task(db, task, "success", message, result)
|
||
|
||
def _execute_create_esports_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)
|
||
act_alias = str(config.get("esports_act_alias") or "").strip()
|
||
manual_id = str(config.get("esports_manual_id") or "").strip()
|
||
if not act_alias or not manual_id:
|
||
self._mark_task(db, task, "failed", "请先配置电竞手册活动 actAlias 和 manualID")
|
||
return
|
||
|
||
baseline_manual_type = None
|
||
baseline_manual_score = None
|
||
baseline_result: dict = {}
|
||
try:
|
||
baseline_result = self._refresh_esports_handbook(client, account, manual_id=manual_id)
|
||
baseline_manual_type = baseline_result["esports_manual_type"]
|
||
baseline_manual_score = baseline_result["esports_manual_score"]
|
||
db.commit()
|
||
except Exception as exc:
|
||
self._push_log("warning", f"生成电竞手册支付码前查询活动状态失败: {exc}")
|
||
|
||
if baseline_manual_type is not None and baseline_manual_type >= 1:
|
||
account.esports_bind_status = "esports_opened"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"success",
|
||
f"电竞手册已开通，积分: {baseline_manual_score if baseline_manual_score is not None else '-'}",
|
||
{**baseline_result, "esports_opened": True, "payment_polling": False},
|
||
)
|
||
return
|
||
|
||
result = client.create_esports_qr(
|
||
ctn=ctn,
|
||
act_alias=act_alias,
|
||
amount=int(config["esports_amount"]),
|
||
room_id=str(config["room_id"]),
|
||
)
|
||
result.update(baseline_result)
|
||
account.esports_bind_status = "esports_qr_created"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
self._update_task_progress(db, task, "running", "电竞手册支付码已生成，等待开通到账", result)
|
||
opened = self._wait_esports_open_after_payment(
|
||
db,
|
||
task,
|
||
account,
|
||
client,
|
||
manual_id=manual_id,
|
||
result=result,
|
||
baseline_manual_type=baseline_manual_type,
|
||
baseline_manual_score=baseline_manual_score,
|
||
)
|
||
if self._stop.is_set():
|
||
self._mark_task(db, task, "stopped", "任务已停止", result)
|
||
return
|
||
if opened:
|
||
account.esports_bind_status = "esports_opened"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"success",
|
||
f"电竞手册已开通，积分: {account.esports_points if account.esports_points is not None else '-'}",
|
||
result,
|
||
)
|
||
return
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"failed",
|
||
"未检测到电竞手册开通到账"
|
||
f"，类型: {result.get('esports_manual_type', '-')}，"
|
||
f"积分: {result.get('esports_manual_score', '-')}",
|
||
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)
|
||
baseline_gold = account.gold_balance
|
||
try:
|
||
baseline = self._refresh_account_gold_balance(client, account)
|
||
baseline_gold = baseline["gold_balance"]
|
||
db.commit()
|
||
except Exception as exc:
|
||
self._push_log("warning", f"生成鱼翅码前刷新余额失败: {exc}")
|
||
result = client.create_gold_qr(amount=amount, pay_type=int(config["gold_pay_type"]))
|
||
account.bind_status = "gold_qr_created"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
self._update_task_progress(db, task, "running", f"鱼翅 {amount} 元支付码已生成，等待到账", result)
|
||
recharged = self._wait_gold_balance_after_payment(db, task, account, client, result, baseline_gold)
|
||
if self._stop.is_set():
|
||
self._mark_task(db, task, "stopped", "任务已停止", result)
|
||
return
|
||
if recharged:
|
||
account.bind_status = "gold_recharged"
|
||
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 '-'}",
|
||
result,
|
||
)
|
||
return
|
||
self._mark_task(
|
||
db,
|
||
task,
|
||
"failed",
|
||
f"未检测到鱼翅到账，当前余额: {account.gold_balance if account.gold_balance is not None else '-'}",
|
||
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)
|
||
ctn = None
|
||
baseline_points = account.points
|
||
try:
|
||
ctn = client.acf_ccn(refresh_subscribe=False)
|
||
baseline_result = self._refresh_account_points(client, account, cookie, ctn=ctn)
|
||
db.commit()
|
||
baseline_points = baseline_result["points"]
|
||
except Exception as exc:
|
||
self._push_log("warning", f"赠送精英令前刷新积分失败: {exc}")
|
||
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"]),
|
||
)
|
||
result["gift_points_baseline"] = baseline_points
|
||
refresh_errors = []
|
||
try:
|
||
result.update(self._refresh_account_gold_balance(client, account))
|
||
except Exception as exc:
|
||
refresh_errors.append(f"鱼翅余额: {exc}")
|
||
try:
|
||
result.update(
|
||
self._refresh_points_after_elite_gift(
|
||
db,
|
||
task,
|
||
account,
|
||
client,
|
||
cookie,
|
||
ctn,
|
||
result,
|
||
baseline_points,
|
||
gift_count,
|
||
)
|
||
)
|
||
except Exception as exc:
|
||
refresh_errors.append(f"积分: {exc}")
|
||
if refresh_errors:
|
||
result["refresh_errors"] = refresh_errors
|
||
account.bind_status = "gift_donated"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
message = f"赠送精英令成功: {gift_count}"
|
||
if account.gold_balance is not None:
|
||
message += f"，鱼翅余额: {account.gold_balance}"
|
||
if account.points is not None:
|
||
message += f"，积分: {account.points}"
|
||
if result.get("gift_points_target") is not None and not result.get("gift_points_confirmed"):
|
||
message += f"（未确认涨到 {result['gift_points_target']}）"
|
||
self._mark_task(db, task, "success", message, 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)
|
||
result = self._refresh_account_points(client, account, cookie, ctn=ctn)
|
||
account.bind_status = "points_queried"
|
||
account.updated_at = datetime.now(timezone.utc)
|
||
points = result["points"]
|
||
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)
|
||
result = self._refresh_account_gold_balance(client, account)
|
||
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 '-'}",
|
||
result,
|
||
)
|
||
|
||
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,
|
||
"prepare_esports_bind": self._execute_prepare_esports_bind,
|
||
"get_esports_bind_qr": self._execute_get_esports_bind_qr,
|
||
"query_esports_game_name": self._execute_query_esports_game_name,
|
||
"confirm_esports_bind": self._execute_confirm_esports_bind,
|
||
"create_esports_qr": self._execute_create_esports_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()
|