"""一键查询：把账号的关键信息一次性从斗鱼官方拉全并汇总。 ## 定位 一个**只读**聚合任务：鱼翅 / 积分（含到期时间）/ 精英宝典 / 电竞手册 / 小店 / 官方兑换记录 / **限兑商品（重点看高价值套装是否已拥有）**。 **不改任何账号状态、不发业务事件** —— 纯查询不是「事实」， 按既有设计例行查询不进时间线，否则查几次就把时间线刷满了。 ## 为什么「是否已拥有」能直接查官方 `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 .account_binding import derive_from_bind_info, set_bind_status from .business_events import record_official_exchange, task_actor 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 #: 配件散件排除关键词列表（避免把头盔、背包误识别为高价值套装） SUIT_EXCLUDE_KEYWORDS = ("头盔", "背包", "手榴弹", "挂件", "降落伞", "枪") def _err(exc: BaseException) -> str: return f"{type(exc).__name__}: {exc}"[:_ERROR_MAX] def _match_suit_item(want: str, rows: list[dict[str, Any]]) -> dict[str, Any] | None: """在商品列表中精确匹配套装商品，严格排除头盔、背包等配件干扰。""" # 1. 优先完全一致匹配 for it in rows: cname = str(it.get("commodityName") or "").strip() if cname == want: return it # 2. 核心词（如 兔兔白日梦 / 狂战萌兔）精准匹配，排除配件关键词 core = want.replace("套装-", "").strip() # 优先找带「套装」前缀的 for it in rows: cname = str(it.get("commodityName") or "").strip() if ( core in cname and not any(k in cname for k in SUIT_EXCLUDE_KEYWORDS) and "套装" in cname ): return it # 3. 兜底匹配不带排除词的项 for it in rows: cname = str(it.get("commodityName") or "").strip() if core in cname and not any(k in cname for k in SUIT_EXCLUDE_KEYWORDS): return it return None def _match_suit_records(want: str, recs: list[dict[str, Any]]) -> list[str]: """在官方兑换记录中匹配对应套装，严格排除配件干扰。""" core = want.replace("套装-", "").strip() got: list[str] = [] for r in recs: cname = str(r.get("commodityName") or r.get("name") or "").strip() if core in cname and not any(k in cname for k in SUIT_EXCLUDE_KEYWORDS): got.append(cname) return got 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]]: """挑出高价值套装，并判定「该账号是否已获得 / 已达兑换上限」。 ## 判定依据：**官方商城商品 status** + 官方兑换记录双重保障 在斗鱼官方精英手册商城中： - `item.status == 1`：斗鱼官方页面按钮展示为【已达兑换上限】（该账号限量 1 件已满，已兑换，不可再兑）； - `item.status == 0`：斗鱼官方页面按钮展示为【立即兑换】（该账号未兑换，可兑换）； - `records`（官方兑换记录）：斗鱼官方常会清理历史兑换记录导致 records 返回空列表（0 条）， 但若 records 中仍存在记录，同样确认为已兑换；若商品已在商城下架不在列表里，records 作为历史兜底。 """ 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 = _match_suit_item(want, rows) got = _match_suit_records(want, recs) # 🎯 核心判定： # 1. 官方商城明确声明 status == 1 即为「已达兑换上限」（已兑换） # 2. 官方历史兑换记录中存在记录即为「已兑换」 status_blocked = hit is not None and str(hit.get("status") or "") == "1" records_owned = bool(got) is_owned = status_blocked or records_owned reason = "" if hit is None and not got: reason = "商品列表和官方兑换记录里都没有这个商品" elif status_blocked and not records_owned: reason = "官方商城显示已达兑换上限" elif records_owned and not status_blocked: reason = "官方兑换记录显示已兑换" elif status_blocked and records_owned: reason = "已达兑换上限且有官方兑换记录" out.append( { "name": str(hit.get("commodityName") or want) if hit else want, "commodity_id": str(hit.get("commodityId") or "") if hit else "", # 🎯 是否已获得 / 已达上限 "owned": is_owned, "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": reason, } ) 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]]], task: Any = None, ) -> int: """把官方兑换记录补成账号事件，返回**新写入**的条数。 幂等由 `record_official_exchange` 的 `source_ref` 保证（含作用域）， 所以一键查询反复执行不会重复落事件。 `task` 只用来取**操作人**（`task.created_by`）—— 谁点的这次一键查询， 这条兑换历史就记在谁名下。不传则退化为「无操作人」，与老行为一致。 """ 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, **task_actor(task), ): 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_bind_info( self, client: DouyuActivityClient, scope: str, alias: str ) -> dict[str, Any]: """按作用域查绑定信息（三个接口不同，收敛在一处）。""" if scope == "elite": return client.bind_info(alias) if scope == "esports": return client.esports_bind_info(alias) return client.xpd_bind_info(act_alias=alias) 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: # 🎯 拉取完整商城商品列表（含每个商品准确的 status 状态：1=已达兑换上限，0=可兑换） goods = client.list_goods(manual_id=manual_id, rid=rid) goods_items = goods.get("goods") or [] overview["limited_goods_total"] = len(goods_items) except Exception as exc: # noqa: BLE001 errors["limited_goods"] = _err(exc) # 商品列表带有 status 准入状态，结合兑换记录在后文判定高价值资产 pending_goods = goods_items # ---- 角色绑定状态（三个作用域都查）---- # # 这三个值以前**从来没人查过** —— 只是 S6 迁移时从旧的垃圾场列 # `bind_status` 能推的推、推不出的留 `unknown`，于是界面上 97% 显示"未知"。 # 实测校准：开发库 4 个账号里 3 个其实**已绑定**（`is_bound_role=True` 且有角色名）， # 全被 unknown 盖住了。 # 和平小店总开关：关着时**完全跳过**它的所有请求。 # 小店目前是关闭状态，不跳过的话每个账号都会产生一条 # "Livelink登录态校验失败"（实测 200 个里失败 195 次，100%）， # 既淹掉真错误、又让人以为出故障。跳过时**不记 errors** —— # 这是配置成关的，不是失败。 peace_on = bool(config.get("peace_enabled")) if not peace_on: overview["peace_skipped"] = ( "和平小店已关闭（douyu_config.peace_enabled=False）" ) overview["bind"] = {} for scope, label, alias_key in ( ("elite", "精英宝典", "bind_act_alias"), ("esports", "电竞手册", "esports_act_alias"), ("peace", "和平小店", "xpd_act_alias"), ): if scope == "peace" and not peace_on: continue alias = str(config.get(alias_key) or "").strip() if not alias: errors[f"bind_{scope}"] = f"未配置{label}活动别名" continue try: info = self._fetch_bind_info(client, scope, alias) derived = derive_from_bind_info(info) # 把**明细**也带上：运营要看的是"绑的哪个角色、哪个区服、 # 什么时候能换绑"，不是一个"已绑定"就完事。 overview["bind"][scope] = { "derived": derived, "role_name": str(info.get("role_name") or info.get("role") or ""), "area_name": str(info.get("area_name") or ""), "plat_name": str(info.get("plat_name") or ""), "nickname": str(info.get("nickname") or ""), "change_role_wait_time": _ts(info.get("change_role_wait_time")), "can_change_time": _ts(info.get("can_change_time")), } if derived is not None: # 唯一写入者；推不出来（None）时**不动原值** —— 宁可留 unknown， # 也不要把"没查到"写成"没绑定" set_bind_status(account, scope, derived) except Exception as exc: # noqa: BLE001 外部接口边界需要宽泛兜底 errors[f"bind_{scope}"] = _err(exc) # ---- 官方兑换记录（**三个作用域都拉**）---- # # 精英宝典 / 电竞手册 用的是同一个接口，只是 manual_id 不同； # 和平小店是另一套（先拿道聚城短时授权 query，再查购买记录）。 # 任一作用域失败只记进 errors，不影响其余两个。 for scope, label, fetcher in ( ("elite", "精英宝典", self._fetch_elite_records), ("esports", "电竞手册", self._fetch_esports_records), ("peace", "和平小店", self._fetch_xpd_records), ): if scope == "peace" and not peace_on: continue 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, task ) 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)