Files
live-hub-py/web/backend/services/douyu_runner_manual.py
T
yml2213 a64380fbcc feat(account): S6 bind_status 三列拆解 —— 四值枚举、唯一写入口、前端改读账号
`accounts.bind_status` 是个**语义垃圾场**:一个自由文本列,6 个文件、**29 处写入**、
20 多种取值,混了三种东西 —— 真状态(`bind_confirmed`)、动作痕迹(`goods_locked`)、
纯查询痕迹(`points_queried`)。它是**覆盖式**的,所以 `bind_confirmed` 会被之后
任意一次查询盖掉 —— 「这个账号到底绑没绑」根本查不出来。
`esports_bind_status` / `xpd_bind_status` 是同一个病。

## 收敛成四值 + 三列一义

`unknown` / `unbound` / `pending` / `bound`,每个作用域一列:

| 列 | 作用域 |
| --- | --- |
| `accounts.role_bind_status`(**新增**) | 精英宝典角色 |
| `accounts.esports_bind_status`(收敛) | 电竞手册角色 |
| `accounts.xpd_bind_status`(收敛) | 和平小店角色 |

- 新增 `services/account_binding.py`:唯一写入口 `set_bind_status()`,
  传旧自由文本直接 `ValueError`(否则垃圾会被写回去);`map_legacy_bind_status()` 负责旧值映射。
- **29 处自由文本写入全部删除或改走枚举**:真状态走枚举,动作/查询痕迹**不再写状态列**
  (动作已由 S5 的直发事件记录,查询痕迹本来就不该是状态)。
- `accounts.bind_status` 保留原值便于比对,**已无任何写入方**,待 S8 drop。
- 一条**源码扫描用例**禁止任何地方再出现 `account.bind_status = "..."` 这类写入。

## 回填不猜(设计文档 §5.5)

只映射**明确陈述了绑定关系**的取值(`bind_confirmed`/`esports_bound`/`xpd_bound` → bound,
`*_qr_*`/`*_ready` → pending,`game_not_bound`/`xpd_not_bound` → unbound),
**其余一律 unknown**。因为列是覆盖式的,「最后一件事是查询」并不代表没绑定。
迁移 SQL 与 Python 映射有一致性用例:**解析迁移源码里的 CASE 逐条比对**
(已反向验证:故意改错一条映射,用例立刻红)。

## ⚠️ 顺带救回一个差点被我自己丢掉的状态

「**电竞手册是否已开通**」以前**只寄生在 `esports_bind_status` 的 `esports_opened`
取值里**(`elite_opened` 有独立布尔列,电竞没有)。归一化会把那个取值抹成 `unknown`
→ **静默丢状态**。所以:
- 新增 `accounts.esports_opened` 布尔列(与 `elite_opened` 对称),开通任务成功时置位;
- 迁移里**必须先 `save_esports_opened()` 再归一化**,顺序用例钉住;
- 详情页「电竞手册」开通状态改读 `esports_opened`。

## 前端不再从任务日志反推账号状态

工作台原来从**当前加载的那一页任务**里挑每个账号最新的一条来推绑定状态
(`latestConfirmBindTaskByAccount` 等)。两个问题:① 支撑任务一旦翻出窗口,
**显示的状态就是错的**;② 这让任务日志变成「账号状态库」,于是它永远不敢清理。

现在状态读账号字段,任务日志只补「扫码流程中的瞬时信息」(如确认前的
`pending_role_name`)。删掉了两个不再需要的派生与对应的 `clearPending` 绕行逻辑。

## 踩到并修掉的东西

- **模型与迁移不一致**:我把模型收窄成 `String(16)`,迁移却留着 `String(32)` ——
  测试库按模型建、生产按迁移建,两边列宽不同。已让迁移也收窄,且**收窄必须在
  归一化之后**(旧取值最长 19 字符,先收窄会截断成废值)。downgrade 也补齐了反向操作,
  已实测 `downgrade -1 && upgrade head` 干净。
- **parity 用例第一版是抄了一份 SQL**(迁移改了用例也不会红,等于没测)——
  改成解析迁移源码;后来又因为列已收窄、19 字符的旧值塞不进测试库,
  干脆改成源码级解析比对(不造数据)。

## 验证

- `TZ=Asia/Shanghai`:**490 passed / 6 skipped**(S5 后 450 → 净增 40)
- `TZ=UTC`(相关文件)106 passed;迁移链 3 passed;ruff 全绿
- 前端 `tsc -b && vite build` 通过、`npm run lint` **0 errors**
- 真实开发库:迁移重放后三列均为 `varchar(16) NOT NULL DEFAULT 'unknown'`;
  12 个账号的旧值 `elite_status_queried`(纯查询痕迹)→ **unknown**(没有猜成 bound)

未做(后续):S6 的**账号页重构**(列表关键列 / 生命周期主页)保留;
`bind_status` 旧列待 S8 drop;保留策略里「状态派生类查询」现在技术上可下探到 7 天档,
保守起见留待确认(影响的是在用页面显示,不是历史)。
2026-10-04 16:28:06 +08:00

551 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""斗鱼任务执行器:手册开通与积分(由 douyu_runner.py 按功能域拆分)。"""
from __future__ import annotations
import time
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from sqlalchemy.orm import Session
from core.douyu import DouyuActivityClient, DouyuActivityError
from ..models import Account, DouyuTask
from .account_events_service import EVENT_ELITE_OPENED, EVENT_ESPORTS_OPENED, emit
from .douyu_runner_core import DOUYU_PAYMENT_POLL_INTERVAL, DOUYU_PAYMENT_POLL_SECONDS
from .douyu_service import account_uid, update_account_profile_from_cookie
class ManualMixin:
"""手册域:精英/电竞手册开通支付、积分查询与到账轮询。"""
if TYPE_CHECKING:
_stop: Any
@staticmethod
def _to_int(value: Any) -> int | None: ...
def _client(self, cookie: str) -> DouyuActivityClient: ...
def _task_payload(self, task: DouyuTask) -> dict: ...
def _push_log(self, level: str, message: str) -> None: ...
def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...
def _update_task_progress(self, *args: Any, **kwargs: Any) -> None: ...
@classmethod
def _confirm_act_alias(cls, config: dict) -> str: ...
@classmethod
def _bind_qr_act_alias(cls, config: dict) -> str: ...
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(UTC)
return {"points": points, "points_query": result}
@staticmethod
def _format_expire_text(expire_time: int | None) -> str:
"""把到期时间戳格式化成中文描述,缺失或为 0 时返回空串。"""
if not expire_time or expire_time <= 0:
return ""
expire_at = datetime.fromtimestamp(expire_time, tz=UTC).astimezone()
return f"(到期 {expire_at:%Y-%m-%d %H:%M})"
def _refresh_elite_handbook(
self,
client: DouyuActivityClient,
account: Account,
*,
manual_id: str,
rid: str = "",
) -> dict:
"""查询精英宝典开通状态(manualType/expireTime)并写回账号。
积分区分不了「开通后积分花完」和「从未开通」,manualType 才是权威口径:
0=未开通,>=1=已开通。
"""
result = client.elite_user_info(manual_id=manual_id, rid=rid)
manual_type = self._to_int(result.get("manual_type")) or 0
manual_score = self._to_int(result.get("manual_score"))
expire_time = self._to_int(result.get("expire_time")) or 0
opened = manual_type >= 1
account.elite_opened = opened
account.elite_expire_time = expire_time or None
if manual_score is not None:
account.points = manual_score
account.updated_at = datetime.now(UTC)
return {
"elite_manual_type": manual_type,
"elite_manual_score": manual_score,
"elite_expire_time": expire_time,
"elite_opened": opened,
"elite_user_info": result,
"points": manual_score,
}
def _execute_query_elite_status(
self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
):
"""查询精英宝典是否已开通、到期时间与积分。"""
manual_id = str(config.get("manual_id") or "").strip()
if not manual_id:
self._mark_task(db, task, "failed", "请先配置精英宝典 manualID")
return
client = self._client(cookie)
result = self._refresh_elite_handbook(
client,
account,
manual_id=manual_id,
rid=str(config.get("rid") or ""),
)
account.updated_at = datetime.now(UTC)
state = "已开通" if result["elite_opened"] else "未开通"
expire_text = self._format_expire_text(result["elite_expire_time"])
points = result["elite_manual_score"]
self._mark_task(
db,
task,
"success",
f"精英宝典{state}{expire_text},积分: {points if points is not None else '-'}",
result,
)
def _wait_points_after_payment(
self,
db: Session,
task: DouyuTask,
account: Account,
client: DouyuActivityClient,
cookie: str,
ctn: str,
result: dict,
*,
baseline_points: int | None = None,
) -> bool:
"""等待宝典支付到账。
判定口径是「积分比开通前有增长」:旧口径 `积分 >= 300` 会让账上本来就
有 300+ 积分的账号在出码后第一轮轮询就被判成功,二维码刚弹出来就消失
(线上 2~4 秒 success 的案例都是这种)。拿不到基线时才退回旧口径。
"""
deadline = time.monotonic() + DOUYU_PAYMENT_POLL_SECONDS
result["payment_polling"] = True
result["payment_target_points"] = 300
result["elite_points_baseline"] = baseline_points
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:
arrived = (
last_points > baseline_points
if baseline_points is not None
else last_points >= 300
)
if arrived:
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: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
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(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: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
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 _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
# 开通前先查状态(对齐电竞手册):已开通的账号直接给出结论,不再生成
# 支付码 —— 既避免二维码刚弹出来就因「积分本来就够」被判成功而消失,
# 也避免给已开通账号生成一张无意义的真实支付单。
manual_id = str(config.get("manual_id") or "").strip()
baseline_points: int | None = None
if manual_id:
try:
baseline = self._refresh_elite_handbook(
client,
account,
manual_id=manual_id,
rid=str(config.get("rid") or ""),
)
db.commit()
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self._push_log(
"warning", f"开通前查询精英宝典状态失败,按未开通流程继续: {exc}"
)
else:
if baseline["elite_opened"]:
expire_text = self._format_expire_text(
baseline["elite_expire_time"]
)
points = baseline["elite_manual_score"]
account.updated_at = datetime.now(UTC)
self._mark_task(
db,
task,
"success",
f"精英宝典已开通{expire_text},积分: "
f"{points if points is not None else '-'}",
{
**baseline,
"elite_already_opened": True,
"payment_polling": False,
},
)
return
baseline_points = baseline["elite_manual_score"]
# 到账轮询读的是 credit/balance,基线也用同一个接口取,
# 避免两个口径不一致时把「余额本来就高」误判成刚到账。
try:
baseline_points = self._refresh_account_points(
client, account, cookie, ctn=ctn
)["points"]
db.commit()
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self._push_log(
"warning", f"开通前查询积分基线失败,改用手册积分: {exc}"
)
result = client.create_elite_qr(
ctn=ctn,
act_alias=act_alias,
amount=int(config["elite_amount"]),
room_id=str(config["room_id"]),
)
account.updated_at = datetime.now(UTC)
self._update_task_progress(
db, task, "running", "精英宝典支付码已生成,等待开通到账", result
)
opened = self._wait_points_after_payment(
db,
task,
account,
client,
cookie,
ctn,
result,
baseline_points=baseline_points,
)
if self._stop.is_set():
self._mark_task(db, task, "stopped", "任务已停止", result)
return
if opened:
account.elite_opened = True
account.updated_at = datetime.now(UTC)
# elite_opened 是**覆盖式**布尔值:开通之后再操作也看不出「什么时候开的」。
# 在这里记一条事件,补上开通时间。
emit(
db,
account_id=account.id,
event_type=EVENT_ELITE_OPENED,
title="精英宝典开通",
occurred_at=account.updated_at,
detail={
"points": account.points,
"expire_time": account.elite_expire_time,
},
source="manual",
source_ref=(
f"elite_opened:{account.id}:{int(account.updated_at.timestamp())}"
),
)
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_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: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self._push_log("warning", f"生成电竞手册支付码前查询活动状态失败: {exc}")
if baseline_manual_type is not None and baseline_manual_type >= 1:
# 这里已确认开通:写独立状态列(以前只把它塞进 esports_bind_status)
account.esports_opened = True
account.updated_at = datetime.now(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.updated_at = datetime.now(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.updated_at = datetime.now(UTC)
# 与精英宝典同理:状态字段是覆盖式的,这里补上「什么时候开的」。
# 注意只在**真正开通**这一支发事件;上面「基线查到已开通」那一支是检测,
# 每次都发会刷屏。
emit(
db,
account_id=account.id,
event_type=EVENT_ESPORTS_OPENED,
title="电竞手册开通",
occurred_at=account.updated_at,
detail={"points": account.esports_points},
source="manual",
source_ref=(
f"esports_opened:{account.id}:{int(account.updated_at.timestamp())}"
),
)
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_query_esports_points(
self,
db: Session,
task: DouyuTask,
account: Account,
cookie: str,
config: dict,
):
"""查询电竞手册积分。"""
manual_id = str(config.get("esports_manual_id") or "").strip()
if not manual_id:
self._mark_task(db, task, "failed", "请先配置电竞手册 manualID")
return
client = self._client(cookie)
result = self._refresh_esports_handbook(client, account, manual_id=manual_id)
account.updated_at = datetime.now(UTC)
points = result["esports_points"]
self._mark_task(
db,
task,
"success",
f"电竞积分: {points if points is not None else '-'}",
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.updated_at = datetime.now(UTC)
points = result["points"]
self._mark_task(
db,
task,
"success",
f"积分: {points if points is not None else '-'}",
result,
)