yml2213
64c399617a
线上事故：分配给客服的账号（如「远行者103」）在斗鱼工作台与 CK 列表里 查不到，只有账号管理里还能搜到。原因是可见性判定写成了 LoginTask.status == "success"，而按设计保留旧 CK 的三种状态都不再是 success： - 普通登录批次启动时会把复用的成功记录置为 pending/running（旧 CK 保留）， 批次被停止或失败后一直停在 pending，账号永久消失； - CK 重登失败置为 relogin_failed（旧 CK 保留），账号从工作台消失； - 普通登录失败置为 failed（复用记录的旧 CK 未清空），同样消失。 线上现状：12545 个账号持有 CK，但只有 11703 个能出现在工作台，842 个 （644 重登失败 + 176 排队中 + …）对客服和管理员都不可见。 改动： - douyu_service 新增 cookie_holding_filter()/latest_cookie_login_task()， 以「记录里还留着 Cookie」作为唯一判据，并抽出 LOGIN_IN_FLIGHT_STATUSES 与 cookie_operable_filter() 供检测/重登入口复用； - 斗鱼工作台账号列表、任务账号校验、账号管理 has_cookie 筛选、分配校验与 概览、CK 管理/操作列表、首页 CK 统计统一改用该判据； - 工作台取 CK 从 latest_success_login_task 换成 latest_cookie_login_task， 保证重登失败/排队中的账号仍能被任务正常执行； - 新增 cleanup_orphan_login_tasks()：进程重启后把遗留的 pending/running 收敛为 failed 并保留旧 CK，避免页面永久显示排队中。 新增回归测试覆盖上述状态下的可见性、可执行性与启动清理。
473 lines
16 KiB
Python
473 lines
16 KiB
Python
"""斗鱼活动任务服务。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from datetime import UTC, datetime
|
||
|
||
from sqlalchemy import and_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from core.douyu.activity_client import DouyuActivityClient
|
||
from core.douyu.cookie_utils import cookie_value
|
||
|
||
from ..models import Account, DouyuConfig, DouyuTask, LoginTask
|
||
|
||
SUPPORTED_DOUYU_TASK_TYPES = {
|
||
"get_bind_qr": "获取绑定二维码",
|
||
"confirm_bind": "确认绑定",
|
||
"create_elite_qr": "开通精英宝典30",
|
||
"prepare_esports_bind": "绑定电竞手册角色",
|
||
"get_esports_bind_qr": "切换电竞角色",
|
||
"query_esports_game_name": "查询最新游戏角色",
|
||
"confirm_esports_bind": "确认电竞手册绑定",
|
||
"create_esports_qr": "开通电竞手册30",
|
||
"query_esports_points": "查询电竞积分",
|
||
"donate_esports_chicken_gift": "赠送冠军鸡腿",
|
||
"donate_esports_firework_gift": "赠送冠军烟花",
|
||
"refresh_esports_goods": "刷新电竞皮肤商城",
|
||
"exchange_esports_goods": "兑换电竞皮肤",
|
||
"create_gold_qr": "充值鱼翅",
|
||
"donate_elite_gift": "赠送精英令",
|
||
"query_points": "一键查询积分",
|
||
"lock_goods": "锁定商品",
|
||
"pay_locked_order": "支付锁单",
|
||
"exchange_goods": "兑换商品",
|
||
"query_game_name": "一键获取游戏名",
|
||
"query_change_bind_time": "一键查询换绑时间",
|
||
"query_limited_goods": "一键查询限兑商品",
|
||
"query_gold_balance": "一键查询鱼翅余额",
|
||
"refresh_goods": "刷新商品列表",
|
||
"query_exchange_records": "一键查询兑换记录",
|
||
"prefetch_csrf_token": "一键获取兑换 CSRF Token",
|
||
"get_xpd_bind_qr": "生成小店绑定二维码",
|
||
"query_xpd_bind_info": "查询小店绑定信息",
|
||
"confirm_xpd_bind": "确认小店绑定",
|
||
"query_xpd_role": "查询小店绑定角色",
|
||
"refresh_xpd_goods": "刷新小店商品列表",
|
||
"query_xpd_balance": "查询小店点券余额",
|
||
"query_xpd_fragments": "查询小店扭蛋碎片",
|
||
"query_xpd_purchase_records": "查询小店兑换记录",
|
||
"exchange_xpd_goods": "兑换小店商品",
|
||
}
|
||
|
||
DOUYU_HANDBOOK_SCOPES = {"elite", "esports", "peace"}
|
||
DOUYU_HANDBOOK_TASK_TYPES = {
|
||
"elite": {
|
||
"get_bind_qr",
|
||
"confirm_bind",
|
||
"create_elite_qr",
|
||
"create_gold_qr",
|
||
"donate_elite_gift",
|
||
"query_points",
|
||
"lock_goods",
|
||
"pay_locked_order",
|
||
"exchange_goods",
|
||
"query_game_name",
|
||
"query_change_bind_time",
|
||
"query_limited_goods",
|
||
"query_gold_balance",
|
||
"refresh_goods",
|
||
"query_exchange_records",
|
||
"prefetch_csrf_token",
|
||
},
|
||
"esports": {
|
||
"prepare_esports_bind",
|
||
"get_esports_bind_qr",
|
||
"query_esports_game_name",
|
||
"confirm_esports_bind",
|
||
"create_esports_qr",
|
||
"query_esports_points",
|
||
"query_gold_balance",
|
||
"query_change_bind_time",
|
||
"query_limited_goods",
|
||
"refresh_esports_goods",
|
||
"exchange_esports_goods",
|
||
"create_gold_qr",
|
||
"donate_esports_chicken_gift",
|
||
"donate_esports_firework_gift",
|
||
},
|
||
"peace": {
|
||
"get_xpd_bind_qr",
|
||
"query_xpd_bind_info",
|
||
"confirm_xpd_bind",
|
||
"query_xpd_role",
|
||
"refresh_xpd_goods",
|
||
"query_xpd_balance",
|
||
"query_xpd_fragments",
|
||
"query_xpd_purchase_records",
|
||
"exchange_xpd_goods",
|
||
},
|
||
}
|
||
|
||
|
||
DOUYU_CONFIG_DEFAULTS = {
|
||
"manual_id": "G4KA4Qnz4LDp7",
|
||
"rid": "9263298",
|
||
"bind_act_alias": "20260120QYOOB",
|
||
"confirm_act_alias": "20260120QYOOB",
|
||
"legacy_act_alias": "cjm",
|
||
"room_id": "9263298",
|
||
"elite_amount": 3000,
|
||
"esports_manual_id": "1W0yGkOdDbAXb",
|
||
"esports_act_alias": "20260226ZMGDG",
|
||
"esports_amount": 3000,
|
||
"esports_chicken_gift_id": "24766",
|
||
"esports_chicken_skin_id": "0",
|
||
"esports_firework_gift_id": "24767",
|
||
"esports_firework_skin_id": "3850",
|
||
"gold_pay_type": 1,
|
||
"gold_recharge_channel": "wechat_qr",
|
||
"gold_api_product_id": "111570",
|
||
"gold_api_account_template_name": "斗鱼昵称",
|
||
"gift_id": "23643",
|
||
"skin_id": "2942",
|
||
"xpd_act_alias": "20260623KDQFH",
|
||
"xpd_act_id": "46195",
|
||
"xpd_rid": "9263298",
|
||
}
|
||
|
||
DOUYU_CONFIG_FIELDS = tuple(DOUYU_CONFIG_DEFAULTS.keys())
|
||
DOUYU_ACTIVE_TASK_STATUSES = ("planned", "pending", "running")
|
||
|
||
|
||
def douyu_config_value(field: str, value):
|
||
"""读取配置值；空值自动回退到默认值。"""
|
||
default = DOUYU_CONFIG_DEFAULTS[field]
|
||
if isinstance(default, int):
|
||
try:
|
||
return int(value if value is not None else default)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
text = str(value or "").strip()
|
||
return text or str(default)
|
||
|
||
|
||
def apply_douyu_config_defaults(config: DouyuConfig) -> bool:
|
||
"""补齐斗鱼配置默认值，返回是否发生变更。"""
|
||
changed = False
|
||
for field in DOUYU_CONFIG_FIELDS:
|
||
if (
|
||
field == "bind_act_alias"
|
||
and str(getattr(config, field, "") or "").strip() == "20250213NQCYX"
|
||
):
|
||
setattr(config, field, DOUYU_CONFIG_DEFAULTS[field])
|
||
changed = True
|
||
continue
|
||
normalized = douyu_config_value(field, getattr(config, field, None))
|
||
if field == "gold_recharge_channel" and normalized not in {
|
||
"wechat_qr",
|
||
"supplier_api",
|
||
}:
|
||
normalized = DOUYU_CONFIG_DEFAULTS[field]
|
||
if getattr(config, field, None) != normalized:
|
||
setattr(config, field, normalized)
|
||
changed = True
|
||
return changed
|
||
|
||
|
||
def ensure_douyu_config(db: Session) -> DouyuConfig:
|
||
"""获取单条斗鱼配置，不存在则创建。"""
|
||
config = db.query(DouyuConfig).first()
|
||
if config:
|
||
if apply_douyu_config_defaults(config):
|
||
db.commit()
|
||
db.refresh(config)
|
||
return config
|
||
config = DouyuConfig(**DOUYU_CONFIG_DEFAULTS)
|
||
db.add(config)
|
||
db.commit()
|
||
db.refresh(config)
|
||
return config
|
||
|
||
|
||
# 登录/重登批次正在处理的状态：这些记录已被批次接管，不允许并发操作。
|
||
LOGIN_IN_FLIGHT_STATUSES = ("pending", "running", "relogin_pending", "relogin_running")
|
||
|
||
|
||
def cookie_holding_filter():
|
||
"""判定登录记录当前是否仍持有 Cookie。
|
||
|
||
账号是否有 CK 只取决于记录里是否还留着 Cookie，而不是取决于批次状态：
|
||
普通登录在排队时会把复用的成功记录临时置为 ``pending``/``running``，
|
||
CK 重登失败会置为 ``relogin_failed``，这些状态都按设计保留旧 Cookie
|
||
（见 ``login_service`` 的「旧 Cookie 已保留」分支），账号必须继续可见。
|
||
只有 Cookie 真正被清空（账号检测成功）时才不再算持有 CK。
|
||
"""
|
||
return and_(
|
||
LoginTask.cookie.isnot(None),
|
||
LoginTask.cookie != "",
|
||
)
|
||
|
||
|
||
def latest_cookie_login_task(db: Session, account_id: int) -> LoginTask | None:
|
||
"""读取账号最近一条仍持有 Cookie 的登录记录。"""
|
||
return (
|
||
db.query(LoginTask)
|
||
.filter(
|
||
LoginTask.account_id == account_id,
|
||
cookie_holding_filter(),
|
||
)
|
||
.order_by(LoginTask.id.desc())
|
||
.first()
|
||
)
|
||
|
||
|
||
def cookie_operable_filter():
|
||
"""判定登录记录能否基于现有 Cookie 再做检测或重登。
|
||
|
||
正在被登录/重登批次处理的行（pending/running/relogin_*）已由批次接管，
|
||
并发再点检测或重登只会互相覆盖状态，因此排除在外。
|
||
"""
|
||
return and_(
|
||
cookie_holding_filter(),
|
||
LoginTask.status.notin_(LOGIN_IN_FLIGHT_STATUSES),
|
||
)
|
||
|
||
|
||
def latest_cookie(db: Session, account_id: int) -> str:
|
||
"""读取账号当前 Cookie。"""
|
||
task = latest_cookie_login_task(db, account_id)
|
||
return task.cookie if task else ""
|
||
|
||
|
||
def cookie_holding_account_ids_query(db: Session):
|
||
"""返回当前仍持有 Cookie 的斗鱼账号 ID 查询。"""
|
||
return db.query(LoginTask.account_id).filter(cookie_holding_filter()).distinct()
|
||
|
||
|
||
def visible_douyu_task_accounts(db: Session, account_ids: list[int]) -> list[Account]:
|
||
"""只保留当前仍持有 Cookie 的斗鱼账号。"""
|
||
if not account_ids:
|
||
return []
|
||
cookie_ids = select(LoginTask.account_id).where(cookie_holding_filter())
|
||
return (
|
||
db.query(Account)
|
||
.filter(Account.id.in_(account_ids), Account.id.in_(cookie_ids))
|
||
.all()
|
||
)
|
||
|
||
|
||
def update_account_profile_from_cookie(account: Account, cookie: str) -> None:
|
||
"""从 Cookie 回填 uid/nickname。"""
|
||
profile = DouyuActivityClient.profile_from_cookie(cookie)
|
||
if profile["uid"]:
|
||
account.uid = profile["uid"]
|
||
if profile["nickname"]:
|
||
account.nickname = profile["nickname"]
|
||
|
||
|
||
def account_uid(account: Account, cookie: str) -> str:
|
||
"""优先从账号字段读取 uid，缺失时从 Cookie 中取。"""
|
||
return account.uid or cookie_value(cookie, "acf_uid")
|
||
|
||
|
||
def create_douyu_planned_tasks(
|
||
db: Session,
|
||
account_ids: list[int],
|
||
task_type: str,
|
||
handbook_scope: str,
|
||
created_by: int,
|
||
payload: dict | None = None,
|
||
) -> tuple[str, int]:
|
||
"""创建斗鱼任务记录，等待后台执行器消费。"""
|
||
if task_type not in SUPPORTED_DOUYU_TASK_TYPES:
|
||
raise ValueError("不支持的任务类型")
|
||
if handbook_scope not in DOUYU_HANDBOOK_SCOPES:
|
||
raise ValueError("无效的工作台")
|
||
if task_type not in DOUYU_HANDBOOK_TASK_TYPES[handbook_scope]:
|
||
raise ValueError("该任务不属于当前工作台")
|
||
|
||
accounts = visible_douyu_task_accounts(db, account_ids)
|
||
if (
|
||
task_type in {"refresh_goods", "refresh_esports_goods", "refresh_xpd_goods"}
|
||
and accounts
|
||
):
|
||
# 商品快照是全局数据，一个可用 CK 足够；没有 CK 时前端无法选账号创建任务。
|
||
accounts = accounts[:1]
|
||
|
||
batch_id = uuid.uuid4().hex[:12]
|
||
payload = payload or {}
|
||
for account in accounts:
|
||
db.add(
|
||
DouyuTask(
|
||
batch_id=batch_id,
|
||
account_id=account.id,
|
||
task_type=task_type,
|
||
handbook_scope=handbook_scope,
|
||
status="planned",
|
||
message="任务已创建，等待执行",
|
||
result={"payload": payload} if payload else None,
|
||
created_by=created_by,
|
||
)
|
||
)
|
||
db.commit()
|
||
return batch_id, len(accounts)
|
||
|
||
|
||
def cleanup_orphan_douyu_tasks(
|
||
db: Session,
|
||
*,
|
||
active_batch_ids: set[str] | None = None,
|
||
batch_id: str | None = None,
|
||
statuses: tuple[str, ...] = DOUYU_ACTIVE_TASK_STATUSES,
|
||
message: str = "任务已中断（服务重启或批次丢失）",
|
||
) -> int:
|
||
"""清理没有执行器接管的斗鱼任务。"""
|
||
query = db.query(DouyuTask).filter(DouyuTask.status.in_(statuses))
|
||
if batch_id:
|
||
query = query.filter(DouyuTask.batch_id == batch_id)
|
||
elif active_batch_ids is not None and active_batch_ids:
|
||
query = query.filter(~DouyuTask.batch_id.in_(list(active_batch_ids)))
|
||
|
||
tasks = query.all()
|
||
if not tasks:
|
||
return 0
|
||
now = datetime.now(UTC)
|
||
for task in tasks:
|
||
task.status = "stopped"
|
||
task.message = message
|
||
task.finished_at = now
|
||
db.commit()
|
||
return len(tasks)
|
||
|
||
|
||
def slim_douyu_goods(goods: object) -> object:
|
||
"""保留兑换图片和列表展示需要的商品小字段。"""
|
||
if not isinstance(goods, dict):
|
||
return None
|
||
return {
|
||
key: value
|
||
for key, value in goods.items()
|
||
if key
|
||
in {
|
||
"commodityId",
|
||
"commodity_id",
|
||
"commodityName",
|
||
"name",
|
||
"webPic",
|
||
"pic",
|
||
"score",
|
||
"status",
|
||
}
|
||
}
|
||
|
||
|
||
def slim_douyu_limited_goods(goods: object) -> list[dict[str, object]]:
|
||
"""限兑列表只返回前几个商品名，避免任务列表携带完整原始数组。"""
|
||
if not isinstance(goods, list):
|
||
return []
|
||
result: list[dict[str, object]] = []
|
||
for item in goods[:5]:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
result.append(
|
||
{
|
||
key: value
|
||
for key, value in item.items()
|
||
if key in {"commodityId", "commodity_id", "commodityName", "name"}
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def strip_douyu_raw_snapshots(value: object) -> object:
|
||
"""递归移除任务结果中的原始接口快照。"""
|
||
if isinstance(value, dict):
|
||
return {
|
||
key: strip_douyu_raw_snapshots(item)
|
||
for key, item in value.items()
|
||
if key != "raw"
|
||
}
|
||
if isinstance(value, list):
|
||
return [strip_douyu_raw_snapshots(item) for item in value]
|
||
return value
|
||
|
||
|
||
def sanitize_douyu_task_result(
|
||
result: dict | None, task_type: str, *, include_detail: bool = False
|
||
) -> dict | None:
|
||
"""列表/实时推送接口剥离原始快照与大数组；详情接口保留完整 result。"""
|
||
if not isinstance(result, dict):
|
||
return result
|
||
if include_detail:
|
||
return result
|
||
|
||
data = strip_douyu_raw_snapshots(result)
|
||
if not isinstance(data, dict):
|
||
return None
|
||
removable_keys = {
|
||
"bind_info",
|
||
"before_bind_info",
|
||
"bind_candidates",
|
||
"cooldown_bind_info",
|
||
"after_bind_info",
|
||
"activity_bind_snapshot",
|
||
"activity_bind_info",
|
||
"points_query",
|
||
"exchange_balance_query",
|
||
"query_act_aliases",
|
||
}
|
||
if task_type != "query_xpd_purchase_records":
|
||
removable_keys.add("records")
|
||
for key in removable_keys:
|
||
data.pop(key, None)
|
||
|
||
# 和平小店任务 result 内嵌 role（含 gameOpenId 等），列表接口剥离其原始快照
|
||
role = data.get("role")
|
||
if isinstance(role, dict):
|
||
data["role"] = strip_douyu_raw_snapshots(role)
|
||
|
||
goods = data.get("goods")
|
||
if isinstance(goods, list):
|
||
data.pop("goods", None)
|
||
elif isinstance(goods, dict):
|
||
data["goods"] = slim_douyu_goods(goods)
|
||
|
||
if "limited_goods" in data:
|
||
data["limited_goods"] = slim_douyu_limited_goods(data.get("limited_goods"))
|
||
|
||
if task_type not in {
|
||
"get_bind_qr",
|
||
"prepare_esports_bind",
|
||
"get_esports_bind_qr",
|
||
"get_xpd_bind_qr",
|
||
}:
|
||
data.pop("url", None)
|
||
if task_type not in {"create_elite_qr", "create_esports_qr", "create_gold_qr"}:
|
||
data.pop("pay_url", None)
|
||
return data
|
||
|
||
|
||
def douyu_task_payload(task: DouyuTask, *, include_detail: bool = False) -> dict:
|
||
"""生成 REST 与 WebSocket 共用的斗鱼任务载荷。"""
|
||
account = task.account
|
||
|
||
def isoformat(value: datetime | None) -> str | None:
|
||
if value is None:
|
||
return None
|
||
if value.tzinfo is None:
|
||
value = value.replace(tzinfo=UTC)
|
||
return value.isoformat()
|
||
|
||
return {
|
||
"id": task.id,
|
||
"batch_id": task.batch_id,
|
||
"account_id": task.account_id,
|
||
"account_username": (account.username if account else "") or "",
|
||
"account_uid": (account.uid if account else "") or "",
|
||
"account_nickname": (account.nickname if account else "") or "",
|
||
"task_type": task.task_type or "",
|
||
"handbook_scope": task.handbook_scope or "legacy",
|
||
"status": task.status or "",
|
||
"message": task.message or "",
|
||
"result": sanitize_douyu_task_result(
|
||
task.result if isinstance(task.result, dict) else None,
|
||
task.task_type or "",
|
||
include_detail=include_detail,
|
||
),
|
||
"created_by": task.created_by,
|
||
"created_at": isoformat(task.created_at),
|
||
"finished_at": isoformat(task.finished_at),
|
||
}
|