Files
live-hub-py/web/backend/services/douyu_runner_overview.py
T
yml2213 62d9053e8a feat(account): 官方兑换记录补成账号事件(三作用域)+ 一键查询结果落到账号
## 官方兑换记录 → 真正的账号事件

关键发现:斗鱼官方兑换记录**带真实兑换时间**(`time` 字段),所以可以直接
补成正常的历史事件,时间线因此变完整(这就是「账号全生命周期」该有的样子),
列表的兑换聚合也随之准确 —— 上线前的兑换不再是一片空白。

- **三个作用域都拉**:精英宝典 / 电竞手册(共用 `exchange_records`,只有 manual_id
  不同)/ 和平小店(`xpd_embed_query` → `xpd_purchase_records`)
- 幂等键含作用域:`official_exchange:{scope}:{account}:{commodity}:{time}`
- `detail.scope` 标作用域;`source=official_sync` 与"当场记的"分得开
- 任一作用域失败只记进 `errors`,不影响其余(小店未开通会报
  `Livelink登录态校验失败`,属正常)

## 一键查询结果落到账号(不再只留在任务里)

- 新增 `accounts.high_value_owned`(迁移 0048):**已兑换过的高价值套装名**,
  供列表展示。单开一列而不是让列表读 `overview_snapshot` —— 那份 JSON 有 1~2KB,
  15000 个账号各带一份会把响应放大到几十 MB。
- 迁移**可安全重跑**(启动时会自动跑):加列前查 `information_schema`。
  ⚠️ 回填里**不要** `op.get_bind().commit()` —— MySQL 的 DDL 自动提交而 commit
  会打乱版本记账,表现为"列加上了、版本号没更新",下次启动重复 ADD COLUMN
  直接把服务起不来(生产上就是启动即挂)。
- 迁移里从旧快照**回填** `high_value_owned`,已查过的账号不必重查。

## 两个 bug

1. **「最近一次兑换」按插入 id 排**(应为事件时间):补录的历史事件 id 最大、
   时间却最老,会把几个月前的老兑换显示成"最近一次"。改为 `occurred_at DESC, id DESC`。
2. **手写的 `@model_serializer` 没列新字段** → 属性有值、JSON 里没有。
   这是本项目第三次踩 Pydantic 静默丢字段;已补,并把断言改为看 `model_dump()`。

## 列表「兑换记录」口径

只聚合**官方**记录(`source=official_sync`):用户要求列表看官方的权威口径,
我们当场记的那些留给详情页时间线。另加 `GET /accounts/{id}/exchange-records`
供列表点击预览(按需拉取,不进列表响应)。
2026-10-04 20:03:42 +08:00

500 lines
21 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.
"""一键查询:把账号的关键信息一次性从斗鱼官方拉全并汇总。
## 定位
一个**只读**聚合任务:鱼翅 / 积分(含到期时间)/ 精英宝典 / 电竞手册 /
小店 / 官方兑换记录 / **限兑商品(重点看高价值套装是否已拥有)**。
**不改任何账号状态、不发业务事件** —— 纯查询不是「事实」,
按既有设计例行查询不进时间线,否则查几次就把时间线刷满了。
## 为什么「是否已拥有」能直接查官方
`query_limited_goods` 返回的商品项里 `count` / `foreverCount` / `storeStatus`
是**账号维度**的(不是全局库存)—— 已用生产镜像库实测,见
`extract_high_value_goods` 的文档。
## 诚实边界
**单项失败不整体失败**:账号没绑电竞手册是常态,不该让整次查询失败。
失败写进 `errors[项]`,并在 UI 如实展示 ——
避免「没查到」被误读成「没有」。
"""
from __future__ import annotations
import json
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from loguru import logger
from sqlalchemy.orm import Session
from core.douyu import DouyuActivityClient
from ..models import Account, DouyuTask
from .business_events import record_official_exchange
from .douyu_runner_donate import _parse_expire_at, _points_expire_at
#: 默认关注的高价值商品。按**商品名包含**匹配 —— 官方列表里的名字带前缀
#: (实际是「套装-兔兔白日梦」),所以配置里写核心词即可。
DEFAULT_HIGH_VALUE_GOODS: tuple[str, ...] = ("兔兔白日梦", "狂战萌兔")
#: 官方兑换记录默认取多少条(完整记录可能很多,一次性拉回来没必要)
EXCHANGE_RECORD_LIMIT = 20
#: 单项错误的摘要长度(避免把整个 traceback 塞进 task.result)
_ERROR_MAX = 200
def _err(exc: BaseException) -> str:
return f"{type(exc).__name__}: {exc}"[:_ERROR_MAX]
def extract_high_value_goods(
items: list[dict[str, Any]] | None,
wanted: tuple[str, ...] | list[str] | None = DEFAULT_HIGH_VALUE_GOODS,
records: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
"""挑出高价值商品,并判定「该账号是否已获得」。
## 判定依据:**官方兑换记录**,不是限兑商品列表
⚠️ 这里推翻了本功能第一版的实现,记录一下踩的坑:
第一版拿 `query_limited_goods` 商品项里的 `count` / `storeStatus` 判断,
依据是「账号 558(兑换过)count=0,账号 15407(没兑换)count=96982」。
**那是错的** —— 两条记录取自 **2026-08-01** 和 **2026-10-04**,相隔两个月,
我把**时间差异**当成了**账号差异**。
在**同一时刻**对比 558 / 607(都兑换过)和 15507(没有),
商品项的**所有字段完全一致**,`preExchange.isExists` 也都是 0 ——
说明 `count` / `foreverCount` / `storeStatus` 是**全局库存/活动状态**,随时间变化,
**不暴露账号维度的拥有信息**。用它判"已拥有"会给运营错误的结论,
而对高价值资产给错结论是这里最不能接受的失败方式。
**正确依据**:`exchange_records(manual_id)` —— 斗鱼官方的**账号级**兑换记录。
实测同一时刻:
====== ============================== ==========
账号 官方兑换记录 判定
====== ============================== ==========
558 破片手榴弹-兔兔白日梦 已获得
607 破片手榴弹-兔兔白日梦 已获得
15507 (0 条) 未获得
====== ============================== ==========
## 两处信息各司其职
- `records`(官方兑换记录)→ **是否已获得**
- `items`(限兑商品列表)→ 该商品**当前**的积分价 / 库存 / 活动时间(是全局信息,
只用来展示"现在还能不能换、要多少分")
## 名字按子串匹配
同一件东西会有多个商品名(实测有「套装-兔兔白日梦」和「破片手榴弹-兔兔白日梦」),
所以按核心词子串匹配,把整个名字家族都算上。
配置里的名字**一个都不能默默消失**:既不在商品列表、也不在兑换记录里的,
返回 `missing=True`,让「没查到这个商品」和「这个账号没有」区分得开。
"""
names = [str(w).strip() for w in (wanted or []) if str(w).strip()]
rows = [it for it in (items or []) if isinstance(it, dict)]
recs = [r for r in (records or []) if isinstance(r, dict)]
out: list[dict[str, Any]] = []
for want in names:
hit = next(
(it for it in rows if want in str(it.get("commodityName") or "")),
None,
)
# 该账号在这件东西上的**全部**兑换记录(可能是多个变体)
got = [
str(r.get("commodityName") or r.get("name") or "")
for r in recs
if want in str(r.get("commodityName") or r.get("name") or "")
]
out.append(
{
"name": str(hit.get("commodityName") or want) if hit else want,
"commodity_id": str(hit.get("commodityId") or "") if hit else "",
# 🎯 是否已获得:**只看官方兑换记录**
"owned": bool(got),
"obtained_names": got,
# 以下是全局的"现在能不能换"信息,仅用于展示
"score": hit.get("score") if hit else None,
"count": hit.get("count") if hit else None,
"count_limited": hit.get("countLimited") if hit else None,
"start_time": _ts(hit.get("startTime")) if hit else None,
"end_time": _ts(hit.get("endTime")) if hit else None,
"pic": str(hit.get("pic") or "") if hit else "",
"in_goods_list": hit is not None,
"missing": hit is None and not got,
"reason": (
"限兑商品列表和官方兑换记录里都没有这个商品"
if hit is None and not got
else ""
),
}
)
return out
def _group_by_scope(
pairs: list[tuple[str, dict[str, Any]]],
) -> dict[str, list[dict[str, Any]]]:
"""把 (作用域, 记录) 扁平列表按作用域分组。"""
grouped: dict[str, list[dict[str, Any]]] = {}
for scope, rec in pairs:
grouped.setdefault(scope, []).append(rec)
return grouped
def _record_official_exchanges(
db: Session, account: Account, pairs: list[tuple[str, dict[str, Any]]]
) -> int:
"""把官方兑换记录补成账号事件,返回**新写入**的条数。
幂等由 `record_official_exchange` 的 `source_ref` 保证(含作用域),
所以一键查询反复执行不会重复落事件。
"""
added = 0
for scope, rec in pairs:
if not isinstance(rec, dict):
continue
when = _ts(rec.get("time"))
occurred = _parse_expire_at(when) if when is not None else None
score_raw = rec.get("score")
try:
score = int(str(score_raw)) if score_raw not in (None, "") else None
except (TypeError, ValueError):
score = None
if record_official_exchange(
db,
account,
commodity_id=str(rec.get("commodityId") or rec.get("commodity_id") or ""),
commodity_name=str(
rec.get("commodityName") or rec.get("commodity_name") or ""
),
score=score,
occurred_at=occurred,
show_name=str(rec.get("showName") or rec.get("show_name") or ""),
exchange_num=rec.get("exchangeNum") or rec.get("count"),
scope=scope,
):
added += 1
return added
def _goods_config(config: dict) -> list[str]:
"""高价值商品名单:优先用配置,非法/为空则回落默认。"""
raw = config.get("high_value_goods")
if isinstance(raw, str) and raw.strip():
try:
raw = json.loads(raw)
except ValueError:
raw = [p for p in raw.replace(",", ",").split(",") if p.strip()]
if isinstance(raw, (list, tuple)):
cleaned = [str(x).strip() for x in raw if str(x).strip()]
if cleaned:
return cleaned
return list(DEFAULT_HIGH_VALUE_GOODS)
def _ts(value: Any) -> Any:
"""斗鱼用 `0` / `"0"` / 空串表示「没有这个时间」。
直接透传的后果:前端把 `0` 当 Unix 时间戳 → 显示成 **1970/1/1 08:00:00**,
运营看到会以为"到期时间是 1970 年"。所以在这里就归成 `None`,
让「没有」和「有一个很早的时间」区分得开。
"""
if value is None:
return None
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return None if int(value) <= 0 else value
text = str(value).strip()
if not text:
return None
if text.lstrip("-").isdigit() and int(text) <= 0:
return None
return value
class OverviewMixin:
"""一键查询域:只读聚合,不写状态、不发事件。"""
if TYPE_CHECKING:
def _client(self, cookie: str) -> DouyuActivityClient: ...
def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...
def _push_log(self, level: str, message: str) -> None: ...
def _refresh_account_gold_balance(
self, client: DouyuActivityClient, account: Account
) -> dict: ...
def _refresh_account_points(
self,
client: DouyuActivityClient,
account: Account,
cookie: str,
*,
ctn: str = "",
) -> dict: ...
def _refresh_elite_handbook(
self,
client: DouyuActivityClient,
account: Account,
*,
manual_id: str,
rid: str = "",
) -> dict: ...
def _refresh_esports_handbook(
self, client: DouyuActivityClient, account: Account, *, manual_id: str
) -> dict: ...
# ---- 三个作用域的官方兑换记录抓取 ----
#
# 精英宝典 / 电竞手册 共用 `exchange_records`(只有 manual_id 不同);
# 和平小店是另一套(道聚城短时授权 + 购买记录)。
@staticmethod
def _scope_configured(
scope: str, manual_id: str, esports_manual_id: str, config: dict
) -> bool:
if scope == "elite":
return bool(manual_id)
if scope == "esports":
return bool(esports_manual_id)
return all(
str(config.get(k) or "").strip()
for k in ("xpd_act_alias", "xpd_rid", "xpd_act_id")
)
def _fetch_elite_records(
self,
client: DouyuActivityClient,
config: dict,
manual_id: str,
esports_manual_id: str,
) -> list[dict[str, Any]]:
return (
client.exchange_records(
manual_id=manual_id, limit=EXCHANGE_RECORD_LIMIT
).get("records")
or []
)
def _fetch_esports_records(
self,
client: DouyuActivityClient,
config: dict,
manual_id: str,
esports_manual_id: str,
) -> list[dict[str, Any]]:
return (
client.exchange_records(
manual_id=esports_manual_id, limit=EXCHANGE_RECORD_LIMIT
).get("records")
or []
)
def _fetch_xpd_records(
self,
client: DouyuActivityClient,
config: dict,
manual_id: str,
esports_manual_id: str,
) -> list[dict[str, Any]]:
"""和平小店:先要一次性短时授权,再查购买记录。"""
embed = client.xpd_embed_query(
act_alias=str(config["xpd_act_alias"]),
rid=str(config["xpd_rid"]),
)
return (
client.xpd_purchase_records(
embed_query=embed["query"],
act_id=str(config["xpd_act_id"]),
).get("records")
or []
)
def _execute_query_account_overview(
self, db: Session, task: DouyuTask, account: Account, cookie: str, config: dict
):
"""一键查询:把该账号的关键信息从斗鱼拉全。
每一项独立 try/except —— 单项失败只记进 `errors`,
绝不因此让整次查询失败(账号没绑电竞手册就是最常见的例子)。
"""
client = self._client(cookie)
manual_id = str(config.get("manual_id") or "").strip()
esports_manual_id = str(config.get("esports_manual_id") or "").strip()
rid = str(config.get("rid") or "").strip()
overview: dict[str, Any] = {
"queried_at": datetime.now(UTC).isoformat(),
"assets": {},
"elite": {},
"esports": {},
"high_value_goods": [],
"exchange_records": {},
"errors": {},
}
errors: dict[str, str] = overview["errors"]
# (作用域, 官方记录) 扁平列表,便于补事件
all_records: list[tuple[str, dict[str, Any]]] = []
pending_goods: list[dict[str, Any]] = []
# ---- 鱼翅 + 钱包兑换余额 ----
try:
gold = self._refresh_account_gold_balance(client, account)
overview["assets"]["gold_balance"] = gold.get("gold_balance")
overview["assets"]["exchange_balance"] = gold.get("exchange_balance")
except Exception as exc: # noqa: BLE001 外部接口边界需要宽泛兜底
errors["gold"] = _err(exc)
# ---- 积分 + 到期时间 ----
try:
ctn = client.acf_ccn(refresh_subscribe=False)
points = self._refresh_account_points(client, account, cookie, ctn=ctn)
overview["assets"]["points"] = points.get("points")
# `_refresh_account_points` 只回 {"points", "points_query"},
# 到期时间得从 points_query 里取 —— 复用已有具名函数,不自己扒结构。
expire_raw = _ts(_points_expire_at(points))
overview["assets"]["points_expire_at"] = expire_raw
# **也写回账号**:积分到期时间是账号的信息,本来就该由账号表承载并
# 在账号详情里展示 —— 查询只是一次「刷新」,不是另开一份副本。
expire_dt = _parse_expire_at(expire_raw)
account.points_expire_at = expire_dt
except Exception as exc: # noqa: BLE001
errors["points"] = _err(exc)
# ---- 精英宝典(开通状态 / 到期时间 / 积分)----
if not manual_id:
errors["elite"] = "未配置精英宝典 manualID"
else:
try:
elite = self._refresh_elite_handbook(
client, account, manual_id=manual_id, rid=rid
)
overview["elite"] = {
"opened": elite.get("elite_opened"),
"expire_time": _ts(elite.get("elite_expire_time")),
"points": elite.get("elite_manual_score"),
}
except Exception as exc: # noqa: BLE001
errors["elite"] = _err(exc)
# ---- 电竞手册 ----
if not esports_manual_id:
errors["esports"] = "未配置电竞手册 manualID"
else:
try:
esports = self._refresh_esports_handbook(
client, account, manual_id=esports_manual_id
)
manual_type = esports.get("esports_manual_type")
# `_refresh_esports_handbook` 不直接给「是否开通」;
# 语义以 create_esports_qr 里的判定为准:manual_type >= 1 = 已开通。
opened = (
manual_type is not None and int(manual_type) >= 1
if isinstance(manual_type, (int, float))
else None
)
overview["esports"] = {
"opened": opened,
"manual_type": manual_type,
"points": esports.get("esports_points"),
"expire_time": _ts(esports.get("esports_expire_time")),
}
except Exception as exc: # noqa: BLE001
errors["esports"] = _err(exc)
goods_items: list[dict[str, Any]] = []
try:
goods = client.query_limited_goods(manual_id=manual_id, rid=rid)
goods_items = goods.get("limited_goods") or []
overview["limited_goods_total"] = len(goods_items)
except Exception as exc: # noqa: BLE001
errors["limited_goods"] = _err(exc)
# 判定「是否已获得」必须用**官方兑换记录**(账号级),
# 不能用限兑商品列表的 count/storeStatus(那是全局库存,会给出错误结论)。
# 所以这里放在兑换记录**之后**才能判定 —— 见下面的顺序。
pending_goods = goods_items
# ---- 官方兑换记录(**三个作用域都拉**)----
#
# 精英宝典 / 电竞手册 用的是同一个接口,只是 manual_id 不同;
# 和平小店是另一套(先拿道聚城短时授权 query,再查购买记录)。
# 任一作用域失败只记进 errors,不影响其余两个。
for scope, label, fetcher in (
("elite", "精英宝典", self._fetch_elite_records),
("esports", "电竞手册", self._fetch_esports_records),
("peace", "和平小店", self._fetch_xpd_records),
):
key = f"exchange_records_{scope}"
if not self._scope_configured(scope, manual_id, esports_manual_id, config):
errors[key] = f"未配置{label}活动参数,跳过"
continue
try:
rows = fetcher(client, config, manual_id, esports_manual_id)
except Exception as exc: # noqa: BLE001 外部接口边界需要宽泛兜底
errors[key] = _err(exc)
continue
overview["exchange_records"][scope] = rows[:EXCHANGE_RECORD_LIMIT]
all_records.extend((scope, r) for r in rows)
overview[f"exchange_record_count_{scope}"] = len(rows)
overview["exchange_record_count"] = len(all_records)
# 把官方记录**补成账号事件**(带真实兑换时间、按作用域去重)。
overview["exchange_events_added"] = _record_official_exchanges(
db, account, all_records
)
overview["exchange_records"] = {
scope: rows[:EXCHANGE_RECORD_LIMIT]
for scope, rows in _group_by_scope(all_records).items()
}
# ---- 🎯 高价值套装判定(依赖兑换记录,必须放在它之后)----
if not manual_id:
errors["high_value_goods"] = "未配置精英宝典 manualID"
else:
try:
# ⚠️ 必须传**摊平后的记录**:`all_records` 是 (作用域, 记录) 元组列表,
# 直接传进去会被 `isinstance(r, dict)` 全过滤掉 —— 表现为
# "所有高价值套装都判定为未获得"(踩过)。
overview["high_value_goods"] = extract_high_value_goods(
pending_goods,
_goods_config(config),
[rec for _scope, rec in all_records],
)
except Exception as exc: # noqa: BLE001
errors["high_value_goods"] = _err(exc)
# 查询会回写账号上的余额/积分缓存(各 _refresh_* 内部已写),
# 但**不改生命周期状态**,也不发事件。
account.updated_at = datetime.now(UTC)
owned = [
g["name"]
for g in overview["high_value_goods"]
if not g.get("missing") and g.get("owned")
]
message = f"一键查询完成;高价值套装已拥有 {len(owned)} 个"
if owned:
message += f": {', '.join(owned)}"
if errors:
message += f"({len(errors)} 项未取到)"
# 落到账号上:任务会被保留策略清掉,而「这个账号现在什么情况」是账号的信息,
# 应该在账号管理里看,不该依赖一条日志的寿命。
account.overview_snapshot = overview
account.overview_queried_at = datetime.now(UTC)
# 再落「已兑换过的高价值套装名」给列表用(列表不能带整份快照)
account.high_value_owned = [
g["name"] for g in overview.get("high_value_goods", []) if g.get("owned")
]
logger.info(f"[douyu] 一键查询账号 {account.id}: {message}")
self._mark_task(db, task, "success", message, overview)