yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

补齐网页Cookie缺失字段并接入会话自动预取guid

Browse Source 
- 新增 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
This commit is contained in:
yml2213
2026-09-01 22:12:58 +08:00
parent 61ce0c9672
commit 5dc736516f
3 changed files with 290 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/elite_session.py
+63
View File
@@ -12,6 +12,7 @@ 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
@@ -28,6 +29,7 @@ class HuyaEliteWssSession:
):
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)
@@ -49,6 +51,39 @@ class HuyaEliteWssSession:
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)
)
@@ -68,6 +103,17 @@ class HuyaEliteWssSession:
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
@@ -89,39 +135,56 @@ class HuyaEliteWssSession:
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):
core/huya/web_cookie_fields.py
+144
View File
@@ -0,0 +1,144 @@
"""虎牙网页 Cookie 缺失字段补齐生成器。
账号 App 登录态只有 16 个 udb_* 认证字段，而活动/商城 WSS 需要完整的网页设备态。
浏览器实测（evidence/browser_cookie_harvest/report.md）确认：
- ``udb_deviceid/udb_guiddata/udb_anobiztoken/udb_anouid`` 由匿名设备态协议
（core/huya/anon_device.py，middle + anonymousLogin）签发；
- ``guid/_qimei_uuid42/__yamid_new/game_did`` 由页面 SDK 脚本写 document.cookie，
无服务端签发接口。其中 ``guid`` 与会话强绑定：**wsLaunch 响应返回的新 guid 必须
回写进 cookie**，否则 getConfig 无响应（2026-09-01 实测）；
``_qimei_uuid42/__yamid_new/game_did`` 为设备标识，随机值格式对齐即可；
- 其余统计/会话字段（SoundValue/Hm_*/rep_cnt 等）服务端仅记录，可随机生成。
2026-09-01 全链路实测：账号 id=58 补齐后 checkUserBindGameAccount 返回
status=200（和平精英已绑定）、getUserScore 返回 status=200（积分 4880）。
"""
from __future__ import annotations
import secrets
import string
import time
# 统计/会话字段生成规则；9.1 抓包值为样例，格式保持对齐
_FILL_MAP: tuple[tuple[str, str], ...] = (
("SoundValue", "0.50"),
("alphaValue", "0.80"),
("isInLiveRoom", "true"),
("udb_passdata", "3"),
("udb_accdata", "undefined"),
)
def _rand_float() -> str:
import random
return f"{random.random():.17f}"
def fill_web_cookie_fields(
cookie: str,
guid: str,
uid: int | str = 0,
*,
keep_guid: bool = True,
) -> str:
"""补齐网页 Cookie 中缺失的字段。
Args:
cookie: 当前 Cookie（可只有 udb_* 认证字段）。
guid: wsLaunch 响应返回的会话 guid（32hex）；缺核心字段时必填。
uid: 虎牙 uid，用于 __yaoldyyuid 等字段。
keep_guid: 已存在 guid 且非空时是否保留（默认 True）。
Returns:
补齐后的 Cookie 字符串（原始字段值不变，仅补缺失项）。
"""
pairs: dict[str, str] = {}
for raw in (cookie or "").split(";"):
item = raw.strip()
if not item or "=" not in item:
continue
key, value = item.split("=", 1)
key = key.strip()
if key:
pairs[key] = value.strip()
def add(key: str, value: str):
if value and key not in pairs:
pairs[key] = value
uid = int(uid or 0)
# 4 个核心设备字段
if guid:
if keep_guid and pairs.get("guid"):
pass # 已有真实 guid 则保留
else:
add("guid", str(guid))
elif "guid" not in pairs:
add("guid", secrets.token_hex(16))
add("_qimei_uuid42", secrets.token_hex(21)) # 42 hex
add("__yamid_new", secrets.token_hex(16).upper()) # 32 hex 大写
alnum = string.ascii_letters + string.digits
import random
add("game_did", "".join(random.choices(alnum, k=30))) # 30 位字母数字
# qimei 皮肤字段
add("_qimei_fingerprint", secrets.token_hex(16))
add("_qimei_h38", secrets.token_hex(19))
# 会话/统计字段
yasmid = _rand_float()
add("__yasmid", yasmid)
add("__yamid_tt1", yasmid)
add("_rep_cnt", str(random.randint(0, 9)))
add("rep_cnt", str(random.randint(0, 9)))
add("h_unt", str(int(time.time())))
if uid:
add("__yaoldyyuid", str(uid))
add("_yasids", "__rootsid%3D" + secrets.token_hex(16).upper())
add("Hm_lvt_51700b6c722f5bb4cf39906a596ea41f", str(int(time.time())))
add("Hm_lpvt_51700b6c722f5bb4cf39906a596ea41f", str(int(time.time())))
add("HMACCOUNT", secrets.token_hex(8).upper())
add("huya_flash_rep_cnt", str(random.randint(1, 400)))
add("huyasp_rep_cnt", str(random.randint(1, 40)))
add("huya_hd_rep_cnt", str(random.randint(1, 40)))
add("huya_web_rep_cnt", str(random.randint(300, 900)))
for key, value in _FILL_MAP:
add(key, value)
return "; ".join(f"{key}={pairs[key]}" for key in pairs)
def missing_web_cookie_fields(cookie: str) -> tuple[str, ...]:
"""返回缺失的网页核心字段（不含统计字段）。"""
from .cookie_utils import cookie_value
core = (
"guid", "_qimei_uuid42", "__yamid_new", "game_did",
"_qimei_fingerprint", "_qimei_h38", "__yasmid", "__yamid_tt1",
"udb_passdata", "_yasids", "HMACCOUNT",
)
return tuple(key for key in core if not cookie_value(cookie, key))
def cookie_fields(cookie: str) -> dict[str, str]:
"""把 Cookie 字符串转 dict（用于测试/调试）。"""
out = {}
for raw in (cookie or "").split(";"):
item = raw.strip()
if not item or "=" not in item:
continue
key, value = item.split("=", 1)
out[key.strip()] = value.strip()
return out
__all__ = [
"cookie_fields",
"fill_web_cookie_fields",
"missing_web_cookie_fields",
]