"""斗鱼高价值套装的官方准入状态。 唯一裁判是精英宝典 ``storehome`` 商品列表中的账号级 ``status``: - ``0`` = 当前账号可兑换； - ``1`` = 当前账号已达兑换上限； - 商品缺失、列表为空、请求失败 = 未知。 兑换记录和积分记录会长期保留，但可能被官方清理或暂时查不到，因此只作历史展示， 不参与当前是否拥有的判定。 """ from __future__ import annotations from datetime import UTC, datetime, timedelta from typing import Any from ..models import Account HIGH_VALUE_STATUS_OWNED = "owned" HIGH_VALUE_STATUS_AVAILABLE = "available" HIGH_VALUE_STATUS_UNKNOWN = "unknown" HIGH_VALUE_KEYS: tuple[str, ...] = ("bairimeng", "mengtu") HIGH_VALUE_NAMES: dict[str, str] = { "bairimeng": "套装-兔兔白日梦", "mengtu": "套装-狂战萌兔", } DEFAULT_HIGH_VALUE_GOODS: tuple[str, ...] = tuple(HIGH_VALUE_NAMES.values()) # 列表展示的“当前准入”最多接受 7 天前的官方结果；过期只能显示待确认。 HIGH_VALUE_FRESH_SECONDS = 7 * 24 * 3600 #: 结论被作废（CK 失效期间）后，多久再拉一次官方状态。 #: #: 取 6 小时是**故意**的：CK 恢复之后，账号的高价值结论必须尽快重新确认 —— #: 失效期间账号可能被别人兑换过，运营正要拿这条结论去兑换。 #: 但也不能更密：商品接口对某些账号稳定失败（没配 manualID），那类账号的 #: 结论会一直是 ``unknown``，每轮都重选就是反复撞墙。 HIGH_VALUE_UNKNOWN_RETRY_SECONDS = 6 * 3600 # 防止头盔、背包等同一套装家族的散件被误当成整套。 SUIT_EXCLUDE_KEYWORDS = ("头盔", "背包", "手榴弹", "挂件", "降落伞", "枪") def high_value_key(name: str) -> str: """把配置里的商品名映射为稳定的短键。""" text = str(name or "").strip() for key, suit_name in HIGH_VALUE_NAMES.items(): if suit_name.replace("套装-", "") in text: return key return "" def _match_suit_item(want: str, rows: list[dict[str, Any]]) -> dict[str, Any] | None: """精确匹配整套，严格排除头盔、背包等散件。""" for item in rows: if str(item.get("commodityName") or "").strip() == want: return item core = want.replace("套装-", "").strip() for item in rows: name = str(item.get("commodityName") or "").strip() if ( core in name and "套装" in name and not any(word in name for word in SUIT_EXCLUDE_KEYWORDS) ): return item for item in rows: name = str(item.get("commodityName") or "").strip() if core in name and not any(word in name for word in SUIT_EXCLUDE_KEYWORDS): return item return None def _match_suit_records(want: str, records: list[dict[str, Any]]) -> list[str]: """匹配历史记录，仅用于界面展示，不参与当前准入判定。""" core = want.replace("套装-", "").strip() matched: list[str] = [] for record in records: name = str(record.get("commodityName") or record.get("name") or "").strip() if core in name and not any(word in name for word in SUIT_EXCLUDE_KEYWORDS): matched.append(name) return matched def _official_state(item: dict[str, Any] | None) -> tuple[str, Any]: if item is None: return HIGH_VALUE_STATUS_UNKNOWN, None status = item.get("status") normalized = str(status).strip() if status is not None else "" if normalized == "0": return HIGH_VALUE_STATUS_AVAILABLE, status if normalized == "1": return HIGH_VALUE_STATUS_OWNED, status return HIGH_VALUE_STATUS_UNKNOWN, status def _ts(value: Any) -> Any: """斗鱼用 0/空串表示没有时间，归一成 None。""" if value is None or 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 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]]: """按官方 ``storehome.status`` 生成高价值套装的当前准入状态。""" names = [str(name).strip() for name in (wanted or []) if str(name).strip()] rows = [item for item in (items or []) if isinstance(item, dict)] historical = [item for item in (records or []) if isinstance(item, dict)] result: list[dict[str, Any]] = [] for wanted_name in names: item = _match_suit_item(wanted_name, rows) state, raw_status = _official_state(item) obtained_names = _match_suit_records(wanted_name, historical) if state == HIGH_VALUE_STATUS_OWNED: reason = "官方商城显示已达兑换上限" elif state == HIGH_VALUE_STATUS_AVAILABLE: reason = "官方商城显示当前可兑换" elif item is None: reason = "官方商城未返回该套装，当前状态未知" else: reason = f"官方商城状态异常（status={raw_status!r}）" if obtained_names and state != HIGH_VALUE_STATUS_UNKNOWN: reason += "；历史兑换记录仅供展示" result.append( { "key": high_value_key(wanted_name), "name": str(item.get("commodityName") or wanted_name) if item else wanted_name, "commodity_id": str(item.get("commodityId") or "") if item else "", "state": state, # 兼容详情页旧字段；是否拥有的唯一来源仍是官方 status。 "owned": state == HIGH_VALUE_STATUS_OWNED, "available": state == HIGH_VALUE_STATUS_AVAILABLE, "status": raw_status, "obtained_names": obtained_names, "score": item.get("score") if item else None, "count": item.get("count") if item else None, "count_limited": item.get("countLimited") if item else None, "start_time": _ts(item.get("startTime")) if item else None, "end_time": _ts(item.get("endTime")) if item else None, "pic": str(item.get("pic") or "") if item else "", "in_goods_list": item is not None, "missing": item is None, "reason": reason, } ) return result def apply_high_value_goods( account: Account, goods: list[dict[str, Any]], *, checked_at: datetime | None = None, ) -> None: """把一次成功读取的官方商城结果写到账号当前状态。""" by_key = { str(item.get("key") or high_value_key(str(item.get("name") or ""))): item for item in goods if isinstance(item, dict) } for key in HIGH_VALUE_KEYS: item = by_key.get(key) or {} state = str(item.get("state") or HIGH_VALUE_STATUS_UNKNOWN) if state not in { HIGH_VALUE_STATUS_OWNED, HIGH_VALUE_STATUS_AVAILABLE, HIGH_VALUE_STATUS_UNKNOWN, }: state = HIGH_VALUE_STATUS_UNKNOWN setattr(account, f"high_value_{key}_status", state) account.high_value_checked_at = checked_at or datetime.now(UTC) def mark_high_value_owned( account: Account, commodity_name: str, *, checked_at: datetime | None = None, ) -> bool: """官方兑换成功后立即收口单个套装状态。 正常路径会再查一次 ``storehome``；这个函数是查询失败时的保守兜底， 依据是官方兑换接口已经成功，不再依赖可能被清理的兑换记录。 """ key = high_value_key(commodity_name) if not key: return False setattr(account, f"high_value_{key}_status", HIGH_VALUE_STATUS_OWNED) account.high_value_checked_at = checked_at or datetime.now(UTC) return True def invalidate_high_value(account: Account) -> None: """CK 失效时把高价值结论**作废**（降为「待确认」）。 为什么 CK 一失效就要作废：官方 ``storehome.status`` 是**账号维度**的， 正常只随这个账号自己的兑换而变。但 CK 失效那段时间，账号**可能被别人 手工登录并兑换**（这些号是共享给客服的）—— 这时缓存里的「可兑换」就 不成立了。 更要命的是 CK 恢复之后：如果不作废，``high_value_payload`` 不再降级， 那条陈旧的「可换」会**重新显示成绿色**，运营就会拿着一份失效结论去兑换。 所以作废是**单向**的：写 ``unknown``，等下一次真的读到官方 ``storehome`` 再写回真实的 ``available``/``owned``。``checked_at`` 保留（那确实是 「上次成功读到官方状态」的时间），但结论本身已经不算数。 """ for key in HIGH_VALUE_KEYS: setattr(account, f"high_value_{key}_status", HIGH_VALUE_STATUS_UNKNOWN) def high_value_payload( account: Account, *, now: datetime | None = None ) -> dict[str, Any]: """给列表/详情使用的紧凑三态结果。""" current = now or datetime.now(UTC) checked_at = account.high_value_checked_at if checked_at is not None and checked_at.tzinfo is None: checked_at = checked_at.replace(tzinfo=UTC) # CK 失效 = 查不动了。此时缓存里的「可兑换」已经无法确认当前是否还成立， # 只能当作待确认 —— 否则运营会拿一个过期结论去兑换。 ck_broken = (account.ck_check_status or "") in {"invalid", "error"} stale = bool( ck_broken or ( checked_at and current - checked_at > timedelta(seconds=HIGH_VALUE_FRESH_SECONDS) ) ) values = { key: str(getattr(account, f"high_value_{key}_status", "") or "") for key in HIGH_VALUE_KEYS } has_unknown = any( value not in {HIGH_VALUE_STATUS_OWNED, HIGH_VALUE_STATUS_AVAILABLE} for value in values.values() ) if checked_at is None: state = "never" elif stale: state = "stale" elif has_unknown: state = "partial" else: state = "complete" return { "state": state, "checked_at": checked_at, "stale": stale, "fresh": state in {"complete", "partial"}, **values, "names": HIGH_VALUE_NAMES, } def high_value_fresh_cutoff(*, now: datetime | None = None) -> datetime: current = now or datetime.now(UTC) return current - timedelta(seconds=HIGH_VALUE_FRESH_SECONDS)