125 lines
3.7 KiB
Python
125 lines
3.7 KiB
Python
"""虎牙活动积分与兑换记录的唯一查询入口。
|
||
|
||
当前活动 SID 必须优先从 ``PlayMallUI.queryAllocation`` 的实时响应发现；
|
||
``huya_config.sid`` 只作为接口不可用时的兼容兜底。工作台任务和外部单协作
|
||
都通过本模块查询，避免两个入口各自解析 SID 后出现「接口 200、记录 0 条」。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Mapping
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
from .huya_points_center import (
|
||
fetch_points_center_catalog,
|
||
points_center_options_from_config,
|
||
)
|
||
|
||
|
||
def _int(value: Any) -> int:
|
||
try:
|
||
return int(value or 0)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
|
||
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] + "..."
|
||
|
||
|
||
@dataclass
|
||
class HuyaActivityQuery:
|
||
"""一次活动查询的原始结果；业务判定由调用方完成。"""
|
||
|
||
sid: int = 0
|
||
sid_source: str = ""
|
||
configured_sid: int = 0
|
||
score_response: Any = None
|
||
records_response: Any = None
|
||
errors: list[str] = field(default_factory=list)
|
||
sid_error: str = ""
|
||
|
||
|
||
def query_huya_activity(
|
||
client: Any,
|
||
*,
|
||
uid: int,
|
||
cookie: str,
|
||
config_info: Mapping[str, Any] | None = None,
|
||
configured_sid: int = 0,
|
||
current_sid: int = 0,
|
||
current_sid_source: str = "",
|
||
want_score: bool = False,
|
||
want_records: bool = False,
|
||
timeout: float = 15.0,
|
||
) -> HuyaActivityQuery:
|
||
"""发现当前 SID，并按需查询积分和兑换记录。
|
||
|
||
SID 发现失败但有配置兜底时不算查询错误，只在 ``sid_error`` 中保留诊断；
|
||
两个业务接口分别捕获异常，保证一个失败不会阻断另一个。
|
||
"""
|
||
configured = _int(configured_sid)
|
||
sid = _int(current_sid)
|
||
if sid:
|
||
sid_source = current_sid_source or "provided"
|
||
sid_error = ""
|
||
else:
|
||
options = points_center_options_from_config(config_info or {})
|
||
catalog = fetch_points_center_catalog(
|
||
uid,
|
||
cookie,
|
||
client=client,
|
||
include_all_tabs=False,
|
||
**options,
|
||
)
|
||
if catalog.sid:
|
||
sid = _int(catalog.sid)
|
||
sid_source = "queryAllocation"
|
||
sid_error = ""
|
||
elif configured:
|
||
sid = configured
|
||
sid_source = "config"
|
||
sid_error = catalog.message
|
||
else:
|
||
sid = 0
|
||
sid_source = ""
|
||
sid_error = catalog.message or "积分中心没有返回活动 SID"
|
||
|
||
result = HuyaActivityQuery(
|
||
sid=sid,
|
||
sid_source=sid_source,
|
||
configured_sid=configured,
|
||
sid_error=sid_error,
|
||
)
|
||
if not sid:
|
||
result.errors.append(f"无法确定当前活动 SID：{sid_error}")
|
||
return result
|
||
|
||
if want_score:
|
||
try:
|
||
result.score_response = client.query_user_score(
|
||
uid=uid,
|
||
cookie=cookie,
|
||
sid=sid,
|
||
timeout=timeout,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 活动接口边界需要保留单项失败
|
||
result.errors.append(f"积分接口异常：{type(exc).__name__}: {_clip(exc)}")
|
||
|
||
if want_records:
|
||
try:
|
||
result.records_response = client.get_user_prize_records(
|
||
uid=uid,
|
||
cookie=cookie,
|
||
sid=sid,
|
||
timeout=timeout,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 活动接口边界需要保留单项失败
|
||
result.errors.append(
|
||
f"兑换记录接口异常：{type(exc).__name__}: {_clip(exc)}"
|
||
)
|
||
|
||
return result
|