yml2213
44b762ea8c
- 高价值套装精确匹配「套装-兔兔白日梦」与「套装-狂战萌兔」，排除头盔/背包/手榴弹散件干扰 - 增加 is_suit_queried_expr，未查询或 NULL 账号严禁误判为双套均可兑换 - 修复前后端序列化空值保持，完善列表与详情页准入状态和判定依据展示 - 新增头盔排除与 NULL 统计防错回归单测，全量通过 689 个用例
574 lines
24 KiB
Python
574 lines
24 KiB
Python
"""一键查询：把账号的关键信息一次性从斗鱼官方拉全并汇总。
|
||
|
||
## 定位
|
||
|
||
一个**只读**聚合任务：鱼翅 / 积分（含到期时间）/ 精英宝典 / 电竞手册 /
|
||
小店 / 官方兑换记录 / **限兑商品（重点看高价值套装是否已拥有）**。
|
||
|
||
**不改任何账号状态、不发业务事件** —— 纯查询不是「事实」，
|
||
按既有设计例行查询不进时间线，否则查几次就把时间线刷满了。
|
||
|
||
## 为什么「是否已拥有」能直接查官方
|
||
|
||
`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
|
||
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):
|
||
if "套装" 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]]]
|
||
) -> 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_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
|
||
)
|
||
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)
|