333 lines
11 KiB
Python
333 lines
11 KiB
Python
"""虎牙积分中心唯一目录读取实现。
|
||
|
||
2026-11-13 两份 HAR（``11.13-兑换记录.har``、
|
||
``11.13-虎牙充值2元-兑换幸运币.har``）确认：
|
||
|
||
* ``PlayMallUI.queryAllocation`` 是积分中心商品、页签、活动 SID 和赛季配置的
|
||
唯一权威来源；
|
||
* 基础响应只返回第一个页签，官方页面随后按 ``pageId + waterfallTabId``
|
||
补拉其它页签；
|
||
* ``webActUI.getActPrizeList`` 不在最新兑换链路中，旧实现已删除。
|
||
|
||
商品刷新、实时下拉、兑换前校验、外部单积分解析、CK 预览、兑换记录 SID 和
|
||
宝典赛季都只能调用本模块，避免各入口重复实现后再次出现旧 prize_id。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from collections.abc import Mapping
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from loguru import logger
|
||
from sqlalchemy.orm import Session
|
||
|
||
from core.huya.shop_structs import (
|
||
extract_points_center_sid,
|
||
extract_points_center_tabs,
|
||
parse_points_center_gifts,
|
||
)
|
||
|
||
from ..huya_defaults import (
|
||
HUYA_DEFAULT_POINTS_PAGE_CODE,
|
||
HUYA_DEFAULT_POINTS_SCENE,
|
||
HUYA_DEFAULT_POINTS_SOURCE_ID,
|
||
)
|
||
from ..models import HuyaConfig
|
||
from .huya_service import huya_config_value
|
||
|
||
GIFTS_LIMIT = 200
|
||
POINTS_SEASON_MS_FLOOR = 1_000_000_000_000
|
||
|
||
|
||
def _clip(value: Any, limit: int = 200) -> str:
|
||
text = str(value if value is not None else "").strip()
|
||
return text if len(text) <= limit else text[: limit - 3] + "..."
|
||
|
||
|
||
def _walk_season_nodes(node: Any, found: list[dict]) -> None:
|
||
if isinstance(node, dict):
|
||
config = node.get("tag0")
|
||
if isinstance(config, dict):
|
||
try:
|
||
start_ms = int(config.get("tag10") or 0)
|
||
end_ms = int(config.get("tag11") or 0)
|
||
config_id = int(config.get("tag0") or 0)
|
||
except (TypeError, ValueError):
|
||
start_ms = end_ms = config_id = 0
|
||
if start_ms >= POINTS_SEASON_MS_FLOOR and end_ms > start_ms:
|
||
found.append(
|
||
{
|
||
"name": _clip(config.get("tag2"), 128),
|
||
"start_ms": start_ms,
|
||
"end_ms": end_ms,
|
||
"config_id": config_id,
|
||
}
|
||
)
|
||
for value in node.values():
|
||
_walk_season_nodes(value, found)
|
||
elif isinstance(node, list):
|
||
for value in node:
|
||
_walk_season_nodes(value, found)
|
||
|
||
|
||
def extract_points_center_season(payload: Any) -> dict | None:
|
||
"""取积分中心当前宝典赛季；认不出结构时返回 ``None``。"""
|
||
found: list[dict] = []
|
||
_walk_season_nodes(payload, found)
|
||
return max(found, key=lambda item: item["end_ms"]) if found else None
|
||
|
||
|
||
def resolve_points_center_config(db: Session) -> dict[str, Any]:
|
||
"""读取积分中心请求参数；空配置自动回落到当前页面默认值。"""
|
||
row = db.query(HuyaConfig).first()
|
||
|
||
def value(field: str, default: str) -> str:
|
||
return (
|
||
huya_config_value(
|
||
field, getattr(row, field, None) if row is not None else None
|
||
)
|
||
or default
|
||
)
|
||
|
||
try:
|
||
scene = int(value("points_scene", HUYA_DEFAULT_POINTS_SCENE))
|
||
except (TypeError, ValueError):
|
||
scene = int(HUYA_DEFAULT_POINTS_SCENE)
|
||
|
||
page_id_text = value("points_page_id", "")
|
||
try:
|
||
page_id = int(page_id_text) if page_id_text else None
|
||
except ValueError:
|
||
page_id = None
|
||
|
||
tab_ids: list[int] = []
|
||
for part in re.split(r"[,，\s]+", value("points_tab_ids", "")):
|
||
if part.isdigit():
|
||
tab_id = int(part)
|
||
if tab_id not in tab_ids:
|
||
tab_ids.append(tab_id)
|
||
|
||
return points_center_options_from_config(
|
||
{
|
||
"points_page_code": value(
|
||
"points_page_code", HUYA_DEFAULT_POINTS_PAGE_CODE
|
||
),
|
||
"points_source_id": value(
|
||
"points_source_id", HUYA_DEFAULT_POINTS_SOURCE_ID
|
||
),
|
||
"points_scene": scene,
|
||
"points_page_id": page_id or "",
|
||
"points_tab_ids": ",".join(str(item) for item in tab_ids),
|
||
}
|
||
)
|
||
|
||
|
||
def points_center_options_from_config(
|
||
config_info: Mapping[str, Any],
|
||
) -> dict[str, Any]:
|
||
"""把配置字段规范的公共入口，任务和路由不再各自解析。"""
|
||
tab_ids = [
|
||
int(part)
|
||
for part in str(config_info.get("points_tab_ids") or "")
|
||
.replace("，", ",")
|
||
.replace(" ", ",")
|
||
.split(",")
|
||
if part.isdigit()
|
||
]
|
||
return {
|
||
"page_code": (
|
||
str(config_info.get("points_page_code") or "").strip()
|
||
or HUYA_DEFAULT_POINTS_PAGE_CODE
|
||
),
|
||
"source_id": (
|
||
str(config_info.get("points_source_id") or "").strip()
|
||
or HUYA_DEFAULT_POINTS_SOURCE_ID
|
||
),
|
||
"scene": int(config_info.get("points_scene") or 0)
|
||
or int(HUYA_DEFAULT_POINTS_SCENE),
|
||
"page_id": int(config_info.get("points_page_id") or 0) or None,
|
||
"tab_ids": tab_ids or None,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class PointsCenterCatalog:
|
||
goods: list[dict] = field(default_factory=list)
|
||
tabs: list[dict] = field(default_factory=list)
|
||
sid: int | None = None
|
||
season: dict | None = None
|
||
payload: dict | None = None
|
||
message: str = ""
|
||
|
||
|
||
def fetch_points_center_catalog(
|
||
uid: int,
|
||
cookie: str,
|
||
*,
|
||
client: Any | None = None,
|
||
page_code: str = HUYA_DEFAULT_POINTS_PAGE_CODE,
|
||
source_id: str = HUYA_DEFAULT_POINTS_SOURCE_ID,
|
||
scene: int = int(HUYA_DEFAULT_POINTS_SCENE),
|
||
page_id: int | None = None,
|
||
tab_ids: list[int] | None = None,
|
||
include_all_tabs: bool = True,
|
||
) -> PointsCenterCatalog:
|
||
"""一次读取积分中心的商品、页签、SID 和赛季配置。
|
||
|
||
``include_all_tabs=False`` 仅供只取 SID/赛季的调用方使用，避免为了元数据
|
||
额外请求所有商品页签。商品列表调用必须保持默认 ``True``。
|
||
"""
|
||
if client is None:
|
||
from core.huya.http_client import HuyaHttpClient
|
||
|
||
client = HuyaHttpClient(
|
||
logger=lambda message: logger.debug(f"[huya-points-center] {message}")
|
||
)
|
||
|
||
def call(
|
||
tab_id: int | None = None, tab_page_id: int | None = None
|
||
) -> tuple[dict | None, str]:
|
||
kwargs: dict[str, Any] = {
|
||
"uid": int(uid),
|
||
"guid": "",
|
||
"cookie": cookie,
|
||
"page_code": page_code,
|
||
"source_id": source_id,
|
||
"scene": scene,
|
||
"deep": True,
|
||
}
|
||
if tab_id is not None:
|
||
kwargs["page_id"] = int(page_id or tab_page_id or 0)
|
||
kwargs["waterfall_tab_id"] = int(tab_id)
|
||
try:
|
||
response = client.play_mall_query_allocation(**kwargs)
|
||
except Exception as exc: # noqa: BLE001 外部接口异常收敛为目录消息
|
||
logger.debug(
|
||
f"[huya-points-center] queryAllocation 异常："
|
||
f"{type(exc).__name__}: {exc}"
|
||
)
|
||
return None, f"{type(exc).__name__}: {_clip(exc)}"
|
||
|
||
if response is None:
|
||
return None, "积分中心接口没有返回数据"
|
||
code = getattr(response, "code", None)
|
||
if code is None:
|
||
code = getattr(response, "status", 0)
|
||
code = int(code or 0)
|
||
if code != 200:
|
||
message = _clip(
|
||
getattr(response, "message", "") or getattr(response, "msg", "")
|
||
)
|
||
return None, f"code={code}{('：' + message) if message else ''}"
|
||
|
||
payload = getattr(response, "payload", None)
|
||
if payload is None and hasattr(response, "to_dict"):
|
||
try:
|
||
payload = (response.to_dict() or {}).get("payload")
|
||
except Exception: # noqa: BLE001 第三方响应结构异常时按空处理
|
||
payload = None
|
||
return payload if isinstance(payload, dict) else None, ""
|
||
|
||
base_payload, error = call()
|
||
if error:
|
||
return PointsCenterCatalog(message=f"拉取商品失败：{error}")
|
||
if base_payload is None:
|
||
return PointsCenterCatalog(message="积分中心接口没有返回商品数据")
|
||
|
||
goods = parse_points_center_gifts(base_payload)
|
||
tabs = extract_points_center_tabs(base_payload)
|
||
sid = extract_points_center_sid(base_payload)
|
||
season = extract_points_center_season(base_payload)
|
||
if not include_all_tabs or (not goods and not tabs):
|
||
message = (
|
||
f"已拉取 {len(goods)} 个礼包（积分中心）"
|
||
if goods
|
||
else "积分中心没返回商品：接口通了但列表是空的"
|
||
)
|
||
return PointsCenterCatalog(
|
||
goods=goods[:GIFTS_LIMIT],
|
||
tabs=tabs,
|
||
sid=sid,
|
||
season=season,
|
||
payload=base_payload,
|
||
message=message,
|
||
)
|
||
|
||
discovered = {(int(tab["page_id"]), int(tab["tab_id"])): tab for tab in tabs}
|
||
loaded_categories = {str(gift.get("category_name") or "").strip() for gift in goods}
|
||
tab_errors: list[str] = []
|
||
if tab_ids:
|
||
targets: list[dict] = []
|
||
for tab_id in tab_ids:
|
||
matched = next(
|
||
(
|
||
tab
|
||
for (tab_page_id, tab_id_value), tab in discovered.items()
|
||
if tab_id_value == int(tab_id)
|
||
and (page_id is None or tab_page_id == int(page_id))
|
||
),
|
||
None,
|
||
)
|
||
if matched is not None:
|
||
targets.append(matched)
|
||
elif page_id is not None:
|
||
targets.append(
|
||
{
|
||
"page_id": int(page_id),
|
||
"tab_id": int(tab_id),
|
||
"name": f"页签 {tab_id}",
|
||
}
|
||
)
|
||
else:
|
||
tab_errors.append(f"页签 {tab_id}：响应里没发现，且未配置 pageId")
|
||
else:
|
||
targets = [
|
||
tab
|
||
for tab in tabs
|
||
if str(tab.get("name") or "").strip() not in loaded_categories
|
||
]
|
||
|
||
seen_products = {
|
||
str(gift.get("product_id") or "") for gift in goods if gift.get("product_id")
|
||
}
|
||
loaded_tabs = len(loaded_categories) if goods else 0
|
||
for tab in targets:
|
||
target_page_id = int(page_id or tab["page_id"])
|
||
tab_payload, tab_error = call(int(tab["tab_id"]), target_page_id)
|
||
if tab_error:
|
||
tab_errors.append(f"{tab.get('name') or tab['tab_id']}：{tab_error}")
|
||
continue
|
||
tab_goods = parse_points_center_gifts(tab_payload)
|
||
added = 0
|
||
for gift in tab_goods:
|
||
product_id = str(gift.get("product_id") or "")
|
||
if product_id and product_id not in seen_products:
|
||
seen_products.add(product_id)
|
||
goods.append(gift)
|
||
added += 1
|
||
if added or tab_goods:
|
||
loaded_tabs += 1
|
||
|
||
goods = goods[:GIFTS_LIMIT]
|
||
if not goods:
|
||
return PointsCenterCatalog(
|
||
tabs=tabs,
|
||
sid=sid,
|
||
season=season,
|
||
payload=base_payload,
|
||
message="积分中心没返回商品：接口通了但列表是空的",
|
||
)
|
||
suffix = f"，{loaded_tabs} 个页签" if loaded_tabs > 1 else ""
|
||
if tab_errors:
|
||
suffix += f"；部分页签未加载：{'；'.join(tab_errors[:3])}"
|
||
return PointsCenterCatalog(
|
||
goods=goods,
|
||
tabs=tabs,
|
||
sid=sid,
|
||
season=season,
|
||
payload=base_payload,
|
||
message=f"已拉取 {len(goods)} 个礼包（积分中心{suffix}）",
|
||
)
|