diff --git a/web/backend/migrations/versions/20261004_0048_overview_summary.py b/web/backend/migrations/versions/20261004_0048_overview_summary.py new file mode 100644 index 0000000..33488e2 --- /dev/null +++ b/web/backend/migrations/versions/20261004_0048_overview_summary.py @@ -0,0 +1,113 @@ +"""账号列表要显示一键查询的结果：加一个**极小**的字段 + +账号列表的「兑换记录」列一直来自我们自己的 `account_events`， +而事件是**上线后才开始记**的 —— 更早的兑换在列表里就是空的， +看起来像"从没兑换过"（用户报的正是这个：账号 6966677706 明明有官方兑换记录， +列表却显示未兑换）。 + +一键查询能查到官方兑换记录，但那份 JSON 有 1~2KB。 +账号列表是全站最热的接口之一，15000 个账号各带一份会把响应放大到几十 MB， +所以**不能**直接把 `overview_snapshot` 返回给列表。 + +这里加一个只含列表所需两项的小字段： +`{"owned": ["兔兔白日梦"], "exchanges": 2}` +""" + +from __future__ import annotations + +import json + +import sqlalchemy as sa +from alembic import op + +revision: str = "20261004_0048" +down_revision: str | None = "20261004_0047" +branch_labels: str | None = None +depends_on: str | None = None + + +def _backfill_from_snapshot() -> None: + """从已有的 `overview_snapshot` 回填。 + + 必须回填：加这个字段之前已经查过的账号，快照里**早就有**这两项信息， + 不回填的话列表会对它们显示「未查询」—— 明明查过，界面却在骗人。 + """ + conn = op.get_bind() + rows = conn.execute( + sa.text( + "SELECT id, overview_snapshot FROM accounts " + "WHERE overview_snapshot IS NOT NULL" + ) + ).fetchall() + updated = 0 + for account_id, snapshot in rows: + data = snapshot + if isinstance(data, (str, bytes)): + try: + data = json.loads(data) + except ValueError: + continue + if not isinstance(data, dict): + continue + goods = data.get("high_value_goods") + owned = [ + g.get("name") + for g in (goods if isinstance(goods, list) else []) + if isinstance(g, dict) and g.get("owned") and g.get("name") + ] + # 只留「已兑换过哪些高价值套装」。 + # 兑换条数不再存这里 —— 列表改成从**官方事件**聚合（口径更准， + # 而且不用维护两套计数）。 + owned_value = owned + conn.execute( + sa.text("UPDATE accounts SET high_value_owned = :owned WHERE id = :id"), + {"owned": json.dumps(owned_value, ensure_ascii=False), "id": account_id}, + ) + updated += 1 + # ⚠️ **不要**在这里 `conn.commit()`。 + # 曾经加过，后果很隐蔽：DDL（ALTER TABLE）在 MySQL 上是自动提交的， + # 但 commit 会打乱 alembic 自己的版本记账 —— 表现为 + # **列加上了、版本号却没更新**，下次启动升级时又去 ADD COLUMN， + # 直接 `Duplicate column name` 把服务起不来（生产上就是启动即挂）。 + # 数据改动用 `op.get_bind().execute(...)` 写即可，提交由 alembic 负责 + # （仓库里 0042 的 S1 回填就是这么写的，一直正常）。 + if updated: + print(f"[0048] 已从快照回填 {updated} 个账号的 high_value_owned") + + +def _column_exists(name: str) -> bool: + """列是否已存在。 + + 应用启动时会自动跑迁移，所以迁移必须**可安全重跑** —— + 万一出现「列已在、版本号没更新」的状态（本迁移就踩过）， + 无脑 ADD COLUMN 会直接 `Duplicate column name` 让服务起不来。 + """ + conn = op.get_bind() + count = conn.execute( + sa.text( + "SELECT COUNT(*) FROM information_schema.columns " + "WHERE table_schema = DATABASE() " + "AND table_name = 'accounts' AND column_name = :name" + ), + {"name": name}, + ).scalar() + return bool(count) + + +def upgrade() -> None: + if not _column_exists("high_value_owned"): + op.add_column( + "accounts", + sa.Column( + "high_value_owned", + sa.JSON(), + nullable=True, + comment="一键查询得到的「已兑换过的高价值套装名」列表（列表用，极小）", + ), + ) + # 无论列是不是这一步加的，都跑回填（幂等：覆盖式写入） + _backfill_from_snapshot() + + +def downgrade() -> None: + op.drop_column("accounts", "high_value_owned") diff --git a/web/backend/models.py b/web/backend/models.py index fab5a8b..65ce0b6 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -104,6 +104,14 @@ class Account(Base): role_bind_status: Mapped[str] = mapped_column( String(16), default="unknown", server_default="unknown" ) + # 一键查询得到的「该账号已兑换过的高价值套装名」列表（如 ["套装-兔兔白日梦"]）。 + # + # 为什么单开一列、而不是让列表读 `overview_snapshot`：那份 JSON 有 1~2KB， + # 账号列表是全站最热的接口之一，15000 个账号各带一份会把响应放大到几十 MB。 + # 这里只有几个短字符串。 + # + # 兑换**条数**不存这里 —— 列表直接从官方事件聚合，不维护第二套计数。 + high_value_owned: Mapped[list | None] = mapped_column(JSON, nullable=True) # 最近一次「一键查询」的快照（结构化结果）。存在账号上而不是只留在任务里： # 任务会被保留策略清掉，而运营看的是"这个账号现在什么情况"。 overview_snapshot: Mapped[dict | None] = mapped_column(JSON, nullable=True) diff --git a/web/backend/routers/accounts.py b/web/backend/routers/accounts.py index 6074c74..ab6b2c4 100644 --- a/web/backend/routers/accounts.py +++ b/web/backend/routers/accounts.py @@ -41,6 +41,7 @@ from ..services.account_events_service import ( EVENT_RECHARGE, EVENT_REMARK_CHANGED, EVENT_TAG_CHANGED, + OFFICIAL_SYNC_SOURCE, emit_bulk, ) from ..services.account_lifecycle import ( @@ -251,12 +252,20 @@ def list_accounts( .filter( AccountEvent.account_id.in_(ids), AccountEvent.event_type == EVENT_EXCHANGE, + # 列表的「兑换记录」只反映**斗鱼官方记录**（口径权威且完整）。 + # 我们当场记的那些留给详情页时间线 —— 用户明确要求这么分。 + AccountEvent.source == OFFICIAL_SYNC_SOURCE, ) .group_by(AccountEvent.account_id) .all() ): exchange_stats[int(account_id)] = (int(count or 0), last_at, "") # ② 最近一次兑换的商品名（窗口函数取每个账号的最新一条） + # + # ⚠️ 按 **`occurred_at`** 排，不能按 `id`： + # 一键查询会把斗鱼官方的**历史**兑换记录补成事件，它们的 `id` 是刚插进去的 + # （最新），但 `occurred_at` 是几年前的。按 id 排会把一条老兑换显示成"最近一次" + # —— 实测就把 08-30 的兔兔白日梦错报成了 10-01（那是另一次兑换）。 ranked = ( db.query( AccountEvent.account_id.label("account_id"), @@ -264,13 +273,17 @@ def list_accounts( func.row_number() .over( partition_by=AccountEvent.account_id, - order_by=AccountEvent.id.desc(), + order_by=( + AccountEvent.occurred_at.desc(), + AccountEvent.id.desc(), + ), ) .label("rn"), ) .filter( AccountEvent.account_id.in_(ids), AccountEvent.event_type == EVENT_EXCHANGE, + AccountEvent.source == OFFICIAL_SYNC_SOURCE, ) .subquery() ) @@ -306,6 +319,12 @@ def list_accounts( last_exchange_at=exchange_stats.get(acc.id, (0, None, ""))[1], last_exchange_name=exchange_stats.get(acc.id, (0, None, ""))[2], **bind_payload(acc), + # 一键查询得到的「已兑换过的高价值套装」（不是整份快照 —— + # 列表带不起 1~2KB/账号 的 JSON）。兑换条数不在这里： + # 上面的 exchange_count 已经是**官方事件**聚合的结果。 + high_value_owned=acc.high_value_owned or [], + # 「有没有查过」靠它判断 —— 否则"查过但没兑换"和"没查过"都是空，分不开 + overview_queried_at=acc.overview_queried_at, ) # 只有管理员可看完整字段（密码、邮箱等） if can_include_sensitive: @@ -909,6 +928,63 @@ def batch_delete_accounts_selection( return {"message": f"已删除 {deleted} 个账号", "deleted": deleted, "success": True} +@router.get("/{account_id}/exchange-records") +def account_exchange_records( + account_id: int, + db: Session = Depends(get_db), + current: User = Depends(get_current_user), +): + """该账号在三个作用域下的**斗鱼官方兑换记录**（按作用域分组）。 + + 列表的「兑换记录」格点开就用这个 —— **按需拉取**，不把记录塞进列表响应 + （15000 个账号各带一堆记录会把响应撑爆）。 + + 数据源是 `source=official_sync` 的事件，也就是一键查询从官方补录的那些， + 它们带**真实兑换时间**。 + """ + if ( + not _visible_accounts_query(db, current) + .filter(Account.id == account_id) + .first() + ): + raise HTTPException(status_code=404, detail="账号不存在或无权查看") + + rows = ( + db.query(AccountEvent) + .filter( + AccountEvent.account_id == account_id, + AccountEvent.event_type == EVENT_EXCHANGE, + AccountEvent.source == OFFICIAL_SYNC_SOURCE, + ) + .order_by(AccountEvent.occurred_at.desc(), AccountEvent.id.desc()) + .all() + ) + grouped: dict[str, list[dict[str, object]]] = { + "elite": [], + "esports": [], + "peace": [], + } + for row in rows: + detail = row.detail or {} + scope = str(detail.get("scope") or "elite") + grouped.setdefault(scope, []).append( + { + "title": row.title or "", + "commodity_id": detail.get("commodity_id") or "", + "commodity_name": detail.get("commodity_name") or "", + "score": detail.get("score"), + "show_name": detail.get("show_name") or "", + "exchange_num": detail.get("exchange_num"), + "occurred_at": row.occurred_at.isoformat() if row.occurred_at else None, + } + ) + return { + "groups": grouped, + "total": sum(len(v) for v in grouped.values()), + "queried_at": None, + } + + @router.get("/{account_id}/overview") def account_overview( account_id: int, @@ -1003,6 +1079,7 @@ def account_overview( "points_expire_at": acc.points_expire_at, "overview_snapshot": acc.overview_snapshot, "overview_queried_at": acc.overview_queried_at, + "high_value_owned": acc.high_value_owned or [], "elite_opened": acc.elite_opened, "elite_expire_time": acc.elite_expire_time, "esports_points": acc.esports_points, diff --git a/web/backend/schemas.py b/web/backend/schemas.py index 48ca1bd..9f031f5 100644 --- a/web/backend/schemas.py +++ b/web/backend/schemas.py @@ -202,6 +202,12 @@ class AccountOut(BaseModel): exchange_count: int = 0 last_exchange_at: datetime | None = None last_exchange_name: str = "" + # 一键查询得到的（账号表里本来没有的）信息，供列表展示。 + # ⚠️ 必须在这里声明：Pydantic 默认 `extra='ignore'`， + # 没声明的字段会被**静默丢掉** —— 路由明明返回了，接口却看不到。 + high_value_owned: list[str] = Field(default_factory=list) + # 「有没有做过一键查询」靠它判断（列表不返回整份快照） + overview_queried_at: datetime | None = None model_config = ConfigDict(from_attributes=True) @@ -236,6 +242,12 @@ class AccountOut(BaseModel): "exchange_count": self.exchange_count, "last_exchange_at": _isoformat(self.last_exchange_at), "last_exchange_name": self.last_exchange_name, + # ⚠️ 这个 `_serialize` 是**手写**的：字段只声明在模型上**不够**， + # 不在这里列出来就会被丢掉。踩过 —— 路由返回的对象属性明明有值， + # 前端拿到的 JSON 里却没有（属性访问能过、序列化后没有， + # 只断言属性的测试因此假绿）。 + "high_value_owned": self.high_value_owned, + "overview_queried_at": _isoformat(self.overview_queried_at), } diff --git a/web/backend/services/account_events_service.py b/web/backend/services/account_events_service.py index b2c7d08..5da64b7 100644 --- a/web/backend/services/account_events_service.py +++ b/web/backend/services/account_events_service.py @@ -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" # 积分到期（资产减少） diff --git a/web/backend/services/business_events.py b/web/backend/services/business_events.py index 25b151a..5ab6b18 100644 --- a/web/backend/services/business_events.py +++ b/web/backend/services/business_events.py @@ -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" diff --git a/web/backend/services/douyu_runner_overview.py b/web/backend/services/douyu_runner_overview.py index 7f06216..3a89635 100644 --- a/web/backend/services/douyu_runner_overview.py +++ b/web/backend/services/douyu_runner_overview.py @@ -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)