yml2213
5dc736516f
- 新增 core/huya/web_cookie_fields.py：4 核心设备字段（guid 取自 wsLaunch 会话、 _qimei_uuid42/__yamid_new/game_did 随机对齐格式）+ qimei 皮肤 + 22 个统计/会话字段 - elite_session 活动通道初始化时若缺网页字段，先 wsLaunch 预取服务端认可 guid， 补齐 cookie 后重建会话；业务 RPC 统一走补齐后的 cookie - 实测：账号 id=58（原始 16 字段登录态）checkUserBindGameAccount status=200 （和平精英已绑定/收麦季）、getUserScore status=200（积分 4880） - 附补齐方案与浏览器实测来源记录 docs/HUYA_CHAT_缺失字段补齐方案.md
225 lines
9.1 KiB
Python
225 lines
9.1 KiB
Python
"""虎牙精英宝典的同步 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"]
|