"""一键查询：把账号的关键信息一次性从斗鱼官方拉全并汇总。 ## 定位 一个**只读**聚合任务：鱼翅 / 积分（含到期时间）/ 精英宝典 / 电竞手册 / 小店 / 官方兑换记录 / **限兑商品（重点看高价值套装是否已拥有）**。 **不改任何账号状态、不发业务事件** —— 纯查询不是「事实」， 按既有设计例行查询不进时间线，否则查几次就把时间线刷满了。 ## 为什么「是否已拥有」能直接查官方 `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 .douyu_runner_donate import _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 _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: ... 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[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 里取 —— 复用已有具名函数，不自己扒结构。 overview["assets"]["points_expire_at"] = _ts(_points_expire_at(points)) 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 # ---- 官方兑换记录 ---- if not manual_id: errors["exchange_records"] = "未配置精英宝典 manualID" else: 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 = [] # ---- 🎯 高价值套装判定（依赖兑换记录，必须放在它之后）---- if not manual_id: errors["high_value_goods"] = "未配置精英宝典 manualID" else: try: overview["high_value_goods"] = extract_high_value_goods( pending_goods, _goods_config(config), 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) logger.info(f"[douyu] 一键查询账号 {account.id}: {message}") self._mark_task(db, task, "success", message, overview)