yml2213
578e259eea
修复迁移中断重跑、旧快照回填覆盖、降级将 unknown 误写成可兑换，以及迁移前 CK 已失效账号未作废旧结论的问题。统一 CK error 的展示与筛选口径，补齐旧快照迁移和兑换兜底反向用例，并同步高价值结论时间语义。同时纳入 .gitignore 更新。
279 lines
11 KiB
Python
279 lines
11 KiB
Python
"""斗鱼高价值套装的官方准入状态。
|
||
|
||
唯一裁判是精英宝典 ``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)
|
||
# 只有明确判定 `invalid` 才降级展示。`error` 表示这次检测没查成，
|
||
# 不等于账号状态变了；若也降级，列表会显示已过期，但筛选/统计仍把
|
||
# 同一账号算作可兑换，造成同一屏两套结论。
|
||
ck_broken = (account.ck_check_status or "") == "invalid"
|
||
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)
|