"""虎牙精英宝典的同步 WSS 会话封装。 任务执行器运行在线程中，使用本类把异步 WSS 生命周期限制在一个账号任务内， 避免在每次 RPC 时重复建连，也不把 asyncio 对象跨线程传递。 """ from __future__ import annotations import asyncio import urllib.parse from collections.abc import Callable from typing import Literal from .http_client import HuyaHttpClient, generate_http_baseinfo from .web_cookie_fields import missing_web_cookie_fields from .wss_client import ACTIVITY_WS_HOST, SHOP_BASEINFO, SHOP_WS_HOST, HuyaWssClient class HuyaEliteWssSession: """一个账号、一个通道、一个可复用的同步 WSS 会话。""" def __init__( self, uid: int, cookie: str, kind: Literal["activity", "shop"], logger: Callable[[str], None] | None = None, host: str | None = None, ): self.uid = int(uid or 0) self.cookie = cookie or "" self._original_cookie = self.cookie self.guid = HuyaHttpClient._resolve_cookie_guid(self.cookie) self.kind = kind self.logger = logger or (lambda _message: None) self.http = HuyaHttpClient(logger=self.logger) self.loop = asyncio.new_event_loop() self.client: HuyaWssClient | None = None self._closed = False if kind == "activity": baseinfo = urllib.parse.unquote( generate_http_baseinfo(self.uid, self.guid, "") ) connect_host = host or ACTIVITY_WS_HOST origin = "https://zt.huya.com" else: baseinfo = SHOP_BASEINFO connect_host = host or SHOP_WS_HOST origin = "https://m-shop.yaoguo.com" self.client = HuyaWssClient(baseinfo=baseinfo, logger=self.logger) try: if kind == "activity" and missing_web_cookie_fields(self.cookie): # 网页设备态缺核心字段（guid/_qimei_uuid42/__yamid_new/game_did 等）： # 先跑一次 wsLaunch 拿到服务端认可的会话 guid，写回 cookie 补齐后再建会话。 # 否则 getConfig / 业务 RPC 无响应（2026-09-01 实测）。 probe = HuyaWssClient(baseinfo=baseinfo, logger=self.logger) try: self.loop.run_until_complete( probe.connect(host=connect_host, origin=origin, cookie=self.cookie) ) self.loop.run_until_complete( probe.call_ws_launch_activity(self.uid, self.guid) ) launch_guid = probe._launch_guid or "" self.logger( f"[WSS] 预取会话 guid={launch_guid or '(空)'}，补齐网页 Cookie 字段" ) finally: if probe.ws is not None: try: self.loop.run_until_complete(probe.disconnect()) except Exception: # noqa: BLE001 清理探针连接失败不影响主流程 self.logger("[WSS] 预取探针清理失败") from .web_cookie_fields import fill_web_cookie_fields self.cookie = fill_web_cookie_fields( self.cookie, launch_guid, self.uid ) self.guid = HuyaHttpClient._resolve_cookie_guid(self.cookie) baseinfo = urllib.parse.unquote( generate_http_baseinfo(self.uid, self.guid, "") ) self.client = HuyaWssClient(baseinfo=baseinfo, logger=self.logger) self.loop.run_until_complete( self.client.connect(host=connect_host, origin=origin, cookie=self.cookie) ) initialized = self.loop.run_until_complete( self.client.initialize_activity(self.uid, self.guid, self.cookie) if kind == "activity" else self.client.initialize(self.uid, "", self.cookie) ) if not initialized: raise RuntimeError(f"虎牙 {kind} WSS 初始化失败") except Exception: self.close() raise def _run(self, coroutine): if self._closed or self.client is None: raise RuntimeError("虎牙 WSS 会话已关闭") return self.loop.run_until_complete(coroutine) def _resolve_cookie(self, cookie: str) -> str: """业务 RPC 统一使用补齐后的网页 Cookie。 任务侧往往传原始登录态 Cookie（缺网页设备字段）；初始化阶段已经 wsLaunch 取回会话 guid 并补齐字段（见 __init__），这里把缺字段的 输入替换为补齐后的 self.cookie。 """ if not cookie or cookie == self._original_cookie or missing_web_cookie_fields(cookie): return self.cookie return cookie def close(self): if self._closed: return self._closed = True if self.client is not None and self.client.ws is not None: try: self.loop.run_until_complete(self.client.disconnect()) except Exception: # noqa: BLE001 - cleanup must not mask task result self.logger("[WSS] 会话清理异常") self.loop.close() def __enter__(self): return self def __exit__(self, _exc_type, _exc, _tb): self.close() def get_act_info(self, act_id: int): return self._run(self.client.get_act_info(act_id)) def get_act_task_detail(self, uid: int, cookie: str, act_id: int): cookie = self._resolve_cookie(cookie) return self._run(self.client.get_act_task_detail(uid, cookie, act_id)) def get_act_user_task_detail(self, uid: int, cookie: str, act_id: int): cookie = self._resolve_cookie(cookie) return self._run(self.client.get_act_user_task_detail(uid, cookie, act_id)) def get_user_score(self, uid: int, cookie: str, sid: int): cookie = self._resolve_cookie(cookie) return self._run(self.client.get_user_score(uid, cookie, sid)) def query_user_score(self, uid: int, cookie: str, sid: int): return self.get_user_score(uid, cookie, sid) def get_act_prize_list(self, uid: int, cookie: str, sid: int): cookie = self._resolve_cookie(cookie) return self._run(self.client.get_act_prize_list(uid, cookie, sid)) def get_act_prize_detail(self, uid: int, cookie: str, sid: int, pid: int): cookie = self._resolve_cookie(cookie) return self._run(self.client.get_act_prize_detail(uid, cookie, sid, pid)) def score_exchange_prize(self, uid: int, cookie: str, sid: int, pid: int): cookie = self._resolve_cookie(cookie) return self._run(self.client.score_exchange_prize(uid, cookie, sid, pid)) def get_user_prize_records(self, uid: int, cookie: str, sid: int): cookie = self._resolve_cookie(cookie) return self._run(self.client.get_user_prize_records(uid, cookie, sid)) def _resolve_kwargs(self, kwargs: dict) -> dict: if kwargs.get("cookie") is not None: kwargs = dict(kwargs) kwargs["cookie"] = self._resolve_cookie(kwargs["cookie"]) return kwargs def check_user_bind_game_account(self, **kwargs): kwargs = self._resolve_kwargs(kwargs) return self._run(self.client.check_user_bind_game_account(**kwargs)) def confirm_bind_act_account(self, **kwargs): kwargs = self._resolve_kwargs(kwargs) return self._run(self.client.confirm_bind_act_account(**kwargs)) def get_live_link_param(self, **kwargs): kwargs = self._resolve_kwargs(kwargs) return self._run(self.client.get_live_link_param(**kwargs)) def get_user_profile_batch(self, **kwargs): kwargs = self._resolve_kwargs(kwargs) return self._run(self.client.get_user_profile_batch(**kwargs)) def build_bind_urls(self, *args, **kwargs): return self.http.build_bind_urls(*args, **kwargs) def get_livelink_mini_qrcode(self, *args, **kwargs): return self.http.get_livelink_mini_qrcode(*args, **kwargs) def get_livelink_qrcode_status(self, *args, **kwargs): return self.http.get_livelink_qrcode_status(*args, **kwargs) def get_goods_info(self, **kwargs): return self._run(self.client.get_goods_info(**kwargs)) def list_pay_channels(self, **kwargs): return self._run(self.client.list_pay_channels(**kwargs)) def check_hy_protocol(self, **kwargs): return self._run(self.client.check_hy_protocol(**kwargs)) def check_user_buy_auth(self, **kwargs): return self._run(self.client.check_user_buy_auth(**kwargs)) def create_order(self, **kwargs): return self._run(self.client.create_order(**kwargs)) def pay_order_submit(self, **kwargs): return self._run(self.client.pay_order_submit(**kwargs)) def order_detail(self, **kwargs): return self._run(self.client.order_detail(**kwargs)) def query_user_order_list(self, **kwargs): return self._run(self.client.query_user_order_list(**kwargs)) __all__ = ["HuyaEliteWssSession"]