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`
供列表点击预览(按需拉取,不进列表响应)。
This commit is contained in:
@@ -48,6 +48,9 @@ EVENT_RELOGIN_OK = "relogin_ok" # 自动重登成功
|
||||
EVENT_RELOGIN_FAILED = "relogin_failed" # 自动重登失败(转人工/退避)
|
||||
EVENT_BIND_OK = "bind_ok" # 绑定/换绑
|
||||
EVENT_RECHARGE = "recharge" # 充值到账
|
||||
#: 事件来源:从斗鱼**官方兑换记录**补录(区别于任务执行时当场记的)
|
||||
OFFICIAL_SYNC_SOURCE = "official_sync"
|
||||
|
||||
EVENT_EXCHANGE = "exchange" # 兑换(含消耗积分)
|
||||
EVENT_POINTS_CREDIT = "points_credit" # 积分到账(含到期时间)
|
||||
EVENT_POINTS_EXPIRE = "points_expire" # 积分到期(资产减少)
|
||||
|
||||
@@ -44,7 +44,9 @@ from .account_events_service import (
|
||||
EVENT_POINTS_CREDIT,
|
||||
EVENT_POINTS_EXPIRE,
|
||||
EVENT_RECHARGE,
|
||||
OFFICIAL_SYNC_SOURCE,
|
||||
emit,
|
||||
to_naive_utc,
|
||||
)
|
||||
|
||||
|
||||
@@ -331,3 +333,66 @@ def record_bind(
|
||||
subject_key=role_name or "",
|
||||
source_ref=task_ref(task_id),
|
||||
)
|
||||
|
||||
|
||||
def record_official_exchange(
|
||||
db: Session,
|
||||
account: Account,
|
||||
*,
|
||||
commodity_id: str,
|
||||
commodity_name: str,
|
||||
score: int | None,
|
||||
occurred_at: datetime | None,
|
||||
show_name: str = "",
|
||||
exchange_num: int | None = None,
|
||||
scope: str = "elite",
|
||||
) -> bool:
|
||||
"""把斗鱼官方兑换记录写成账号事件。
|
||||
|
||||
## 为什么要写进事件流
|
||||
|
||||
列表的「兑换记录」列一直来自我们自己的 `account_events`,而事件是**上线后**
|
||||
才开始记的 —— 上线前的兑换在界面上就是空白,看着像"从没兑换过"
|
||||
(用户报过:账号明明有官方兑换记录,列表却显示「未兑换」)。
|
||||
|
||||
官方记录**带真实兑换时间**(`time` 字段),所以可以直接补成正常的历史事件,
|
||||
时间线因此变完整(这就是"账号全生命周期"该有的样子),
|
||||
列表的聚合也随之自动准确 —— 不需要再单独维护一套"官方条数"的旁路。
|
||||
|
||||
## 幂等
|
||||
|
||||
`source_ref` 用 `官方:账号:商品:时刻`,同一条记录查多少次都只落一条。
|
||||
这也是必须的:一键查询会被反复执行。
|
||||
"""
|
||||
if not commodity_id:
|
||||
return False
|
||||
return emit(
|
||||
db,
|
||||
account_id=account.id,
|
||||
event_type=EVENT_EXCHANGE,
|
||||
title=f"兑换 {commodity_name}" if commodity_name else "兑换",
|
||||
category=CATEGORY_BUSINESS,
|
||||
occurred_at=occurred_at,
|
||||
amount=-int(score) if score else None,
|
||||
detail={
|
||||
"commodity_id": commodity_id,
|
||||
"commodity_name": commodity_name,
|
||||
"score": score,
|
||||
"exchange_num": exchange_num,
|
||||
"show_name": show_name,
|
||||
# 标明来源:这条是**从官方记录补录**的,不是我们亲眼看着兑换的
|
||||
"source": OFFICIAL_SYNC_SOURCE,
|
||||
# 哪个作用域的官方记录(精英宝典 / 电竞手册 / 和平小店)
|
||||
"scope": scope,
|
||||
},
|
||||
source=OFFICIAL_SYNC_SOURCE,
|
||||
subject_type="goods",
|
||||
source_ref=(
|
||||
f"official_exchange:{scope}:{account.id}:{commodity_id}:{_stamp(occurred_at)}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _stamp(value: datetime | None) -> str:
|
||||
"""幂等键里的时间部分;没有时间就用固定标记(同一商品仍只落一条)。"""
|
||||
return to_naive_utc(value).strftime("%Y%m%d%H%M%S") if value else "unknown"
|
||||
|
||||
@@ -33,6 +33,7 @@ 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
|
||||
|
||||
#: 默认关注的高价值商品。按**商品名包含**匹配 —— 官方列表里的名字带前缀
|
||||
@@ -138,6 +139,52 @@ def extract_high_value_goods(
|
||||
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")
|
||||
@@ -205,6 +252,72 @@ class OverviewMixin:
|
||||
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
|
||||
):
|
||||
@@ -224,11 +337,12 @@ class OverviewMixin:
|
||||
"elite": {},
|
||||
"esports": {},
|
||||
"high_value_goods": [],
|
||||
"exchange_records": [],
|
||||
"exchange_records": {},
|
||||
"errors": {},
|
||||
}
|
||||
errors: dict[str, str] = overview["errors"]
|
||||
all_records: list[dict[str, Any]] = []
|
||||
# (作用域, 官方记录) 扁平列表,便于补事件
|
||||
all_records: list[tuple[str, dict[str, Any]]] = []
|
||||
pending_goods: list[dict[str, Any]] = []
|
||||
|
||||
# ---- 鱼翅 + 钱包兑换余额 ----
|
||||
@@ -308,29 +422,51 @@ class OverviewMixin:
|
||||
# 所以这里放在兑换记录**之后**才能判定 —— 见下面的顺序。
|
||||
pending_goods = goods_items
|
||||
|
||||
# ---- 官方兑换记录 ----
|
||||
if not manual_id:
|
||||
errors["exchange_records"] = "未配置精英宝典 manualID"
|
||||
else:
|
||||
# ---- 官方兑换记录(**三个作用域都拉**)----
|
||||
#
|
||||
# 精英宝典 / 电竞手册 用的是同一个接口,只是 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:
|
||||
records = client.exchange_records(
|
||||
manual_id=manual_id, limit=EXCHANGE_RECORD_LIMIT
|
||||
)
|
||||
rows = records.get("records") or []
|
||||
all_records = rows
|
||||
overview["exchange_records"] = rows[:EXCHANGE_RECORD_LIMIT]
|
||||
overview["exchange_record_count"] = len(rows)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
errors["exchange_records"] = _err(exc)
|
||||
all_records = []
|
||||
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), all_records
|
||||
pending_goods,
|
||||
_goods_config(config),
|
||||
[rec for _scope, rec in all_records],
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
errors["high_value_goods"] = _err(exc)
|
||||
@@ -354,6 +490,10 @@ class OverviewMixin:
|
||||
# 应该在账号管理里看,不该依赖一条日志的寿命。
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user