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

修复虎牙绑定角色识别

Browse Source
This commit is contained in:
yml2213
2026-07-22 16:59:12 +08:00
parent 4654612d81
commit 69b6120ede
4 changed files with 452 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/activity_structs.py
+7
View File
@@ -14,6 +14,7 @@ class ActivityUserId(TafStruct):
self.sCookie: str = ""
self.iTokenType: int = 0
self.sDeviceInfo: str = ""
self.sQIMEI: str = ""
def write_to(self, os: TafOutputStream):
os.write_int64(0, self.lUid)
@@ -23,6 +24,7 @@ class ActivityUserId(TafStruct):
os.write_string(4, self.sCookie)
os.write_int32(5, self.iTokenType)
os.write_string(6, self.sDeviceInfo)
os.write_string(7, self.sQIMEI)
def read_from(self, ins: TafInputStream):
self.lUid = ins.read_int64(0, default=self.lUid)
@@ -32,6 +34,7 @@ class ActivityUserId(TafStruct):
self.sCookie = ins.read_string(4, default=self.sCookie)
self.iTokenType = ins.read_int32(5, default=self.iTokenType)
self.sDeviceInfo = ins.read_string(6, default=self.sDeviceInfo)
self.sQIMEI = ins.read_string(7, default=self.sQIMEI)
class GetUserScoreReq(TafStruct):
@@ -778,6 +781,7 @@ class GameRole(TafStruct):
self.platName: str = ""
self.partitionName: str = ""
self.gameSerial: str = ""
self.extendInfo: str = ""
def read_from(self, ins: TafInputStream):
self.roleId = ins.read_string(0, default=self.roleId)
@@ -786,6 +790,7 @@ class GameRole(TafStruct):
self.platName = ins.read_string(3, default=self.platName)
self.partitionName = ins.read_string(4, default=self.partitionName)
self.gameSerial = ins.read_string(5, default=self.gameSerial)
self.extendInfo = ins.read_string(6, default=self.extendInfo)
def write_to(self, os: TafOutputStream):
os.write_string(0, self.roleId)
@@ -794,6 +799,7 @@ class GameRole(TafStruct):
os.write_string(3, self.platName)
os.write_string(4, self.partitionName)
os.write_string(5, self.gameSerial)
os.write_string(6, self.extendInfo)
def to_dict(self) -> dict:
return {
@@ -803,6 +809,7 @@ class GameRole(TafStruct):
"plat_name": self.platName,
"partition_name": self.partitionName,
"game_serial": self.gameSerial,
"extend_info": self.extendInfo,
}
core/huya/http_client.py
+118 -11
View File
@@ -7,6 +7,7 @@
Response: Wup 包 (tRsp)
"""
import base64
import hashlib
import json
import random
import struct
@@ -121,18 +122,66 @@ class HuyaHttpClient:
self.logger = logger or print
@staticmethod
def _normalize_cookie(cookie: str) -> str:
def _cookie_first_value(cookie: str, key: str) -> str:
"""读取 Cookie 中第一次出现的 key；浏览器 Cookie 里 guid 可能重复。"""
for item_key, item_value in cookie_pairs(cookie):
if item_key == key and item_value:
return item_value
return ""
@classmethod
def _resolve_cookie_guid(cls, cookie: str) -> str:
"""给 cdnws baseinfo 准备稳定 guid，缺失时用账号 Cookie 派生。"""
guid = cls._cookie_first_value(cookie, "guid")
if guid:
return guid
udb_guid = cls._cookie_first_value(cookie, "udb_guiddata")
if udb_guid:
return udb_guid[:32]
parts = [
cls._cookie_first_value(cookie, "yyuid"),
cls._cookie_first_value(cookie, "udb_uid"),
cls._cookie_first_value(cookie, "udb_passport"),
cls._cookie_first_value(cookie, "username"),
cls._cookie_first_value(cookie, "udb_biztoken"),
]
seed = "|".join(part for part in parts if part)
return hashlib.sha256(seed.encode("utf-8")).hexdigest()[:32] if seed else ""
@classmethod
def _normalize_cookie(cls, cookie: str) -> str:
"""HTTP 业务 UserId.sCookie 需要带 huya_ua 前缀，避免重复写入。"""
if not cookie:
return ""
pairs = [(key, value) for key, value in cookie_pairs(normalize_huya_cookie(cookie)) if key != "huya_ua"]
if not any(key == "guid" for key, _ in pairs):
guid = cls._resolve_cookie_guid(cookie)
if guid:
pairs.append(("guid", guid))
return normalize_cookie_pairs([("huya_ua", HTTP_HUYA_UA), *pairs])
@classmethod
def _generate_rpc_baseinfo(
cls,
uid: int,
guid: str,
cookie: str,
trace_id: str = None,
) -> str:
"""生成与浏览器 cdnws 调用一致的 baseinfo。"""
rpc_cookie = cls._normalize_cookie(cookie)
rpc_guid = guid or cls._resolve_cookie_guid(cookie)
return generate_http_baseinfo(uid, rpc_guid, rpc_cookie, trace_id=trace_id)
@staticmethod
def _build_user(uid: int, guid: str, cookie: str):
"""构造 HTTP 业务 UserId。"""
from .shop_structs import UserId
user = UserId()
user.lUid = uid
user.sGuid = guid
user.sGuid = guid or HuyaHttpClient._resolve_cookie_guid(cookie)
user.sToken = ""
user.sHuYaUA = HTTP_HUYA_UA
user.sCookie = HuyaHttpClient._normalize_cookie(cookie)
@@ -169,7 +218,7 @@ class HuyaHttpClient:
user = ActivityUserId()
user.lUid = uid
user.sHuYaUA = HTTP_HUYA_UA
user.sCookie = normalize_huya_cookie(cookie)
user.sCookie = HuyaHttpClient._normalize_cookie(cookie)
return user
def call_rpc(self, service: str, method: str,
@@ -195,7 +244,7 @@ class HuyaHttpClient:
wup_data = wup.encode() # [4字节长度][wup body]
# 2. 构造 baseinfo
baseinfo = generate_http_baseinfo(uid, guid, cookie)
baseinfo = self._generate_rpc_baseinfo(uid, guid, cookie)
url = f"https://{CDNWS_HOST}/?baseinfo={baseinfo}"
self.logger(f"[HTTP] POST {service}.{method}")
@@ -476,7 +525,7 @@ class HuyaHttpClient:
req = urllib.request.Request(endpoint, data=payload, method="POST", headers={
"User-Agent": PC_UA,
"Origin": "https://livelink.qq.com",
"Referer": bind_page_url,
"Referer": "https://livelink.qq.com/",
"Content-Type": "application/json;charset=UTF-8",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
@@ -519,6 +568,56 @@ class HuyaHttpClient:
"qrcode_token": jdata.get("qrcodeToken") or "",
}
def get_livelink_qrcode_status(self, qrcode_token: str, timeout: float = 15.0) -> dict | None:
"""轮询 livelink 小程序码扫码状态。"""
import gzip
token = (qrcode_token or "").strip()
if not token:
return None
payload = json.dumps(
{"qrcodeToken": token},
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
endpoint = f"{LIVELINK_BIND_API_BASE}/api/qrcode/poolingQrcodeStatus"
req = urllib.request.Request(endpoint, data=payload, method="POST", headers={
"User-Agent": PC_UA,
"Origin": "https://livelink.qq.com",
"Referer": "https://livelink.qq.com/",
"Content-Type": "application/json;charset=UTF-8",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
})
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
resp_data = resp.read()
if resp_data[:2] == b"\x1f\x8b" or resp.headers.get("Content-Encoding") == "gzip":
resp_data = gzip.decompress(resp_data)
except Exception as e:
self.logger(f"[LIVELINK] ❌ 二维码状态请求失败: {type(e).__name__}: {e}")
return None
try:
data = json.loads(resp_data.decode("utf-8", "replace"))
except Exception as e:
self.logger(f"[LIVELINK] ❌ 二维码状态响应解析失败: {type(e).__name__}: {e}")
return None
if data.get("iRet") != 0:
self.logger(f"[LIVELINK] ❌ 二维码状态接口失败: iRet={data.get('iRet')} msg={data.get('sMsg') or '-'}")
return None
jdata = data.get("jData") if isinstance(data.get("jData"), dict) else {}
return {
"is_scan": bool(jdata.get("isScan")),
"is_expired": bool(jdata.get("isExpired")),
"is_completed": bool(jdata.get("isCompleted")),
"is_failure": bool(jdata.get("isFailure")),
"raw": jdata,
}
@staticmethod
def build_bind_urls(
livelink_param: dict,
@@ -526,6 +625,7 @@ class HuyaHttpClient:
game_auth_scene: str = "",
nick_name: str = "",
face_url: str = "",
redirect_url: str = "",
) -> dict:
"""按活动页 JS 逻辑拼绑定二维码 URL。"""
from urllib.parse import quote, urlencode
@@ -551,7 +651,7 @@ class HuyaHttpClient:
f"&path={mini_path}&username=gh_49af166706ae"
)
pc_query = dict(bind_query)
pc_query["redirectUrl"] = ""
pc_query["redirectUrl"] = quote(redirect_url or "", safe="")
pc_url = f"https://livelink.qq.com/act/a20200527midgroupage/pc/?{join_query(pc_query)}"
huya_params = {
@@ -712,7 +812,7 @@ class HuyaHttpClient:
wup.newdata["tReq"] = os.get_bytes()
wup_data = wup.encode()
baseinfo = generate_http_baseinfo(uid, guid, cookie)
baseinfo = self._generate_rpc_baseinfo(uid, guid, cookie)
url = f"https://{CDNWS_HOST}/?baseinfo={baseinfo}"
self.logger(f"[HTTP] POST shopMiddleUI.payOrderSubmitV5")
self.logger(f"[HTTP] 发送 hex前60: {wup_data[:60].hex()}")
@@ -747,23 +847,30 @@ class HuyaHttpClient:
def _self_check():
uid = 1199647239697
cookie = "yyuid=1199647239697; udb_passport=test"
guid = "0a8973013ad3476a53029ddb9edb071f"
cookie = f"yyuid=1199647239697; udb_passport=test; guid={guid}"
trace_id = "0123456789abcdef:0123456789abcdef:0:0"
encoded = generate_http_baseinfo(uid, "", cookie, trace_id=trace_id)
encoded = HuyaHttpClient._generate_rpc_baseinfo(uid, "", cookie, trace_id=trace_id)
raw = base64.b64decode(urllib.parse.unquote(encoded))
parsed = WSConnectParaInfo()
parsed.read_from(TafInputStream(raw))
assert parsed.lUid == uid
assert parsed.sGuid == guid
assert parsed.sUA == HTTP_HUYA_UA
assert parsed.sCookie == cookie
assert parsed.sCookie.startswith(f"huya_ua={HTTP_HUYA_UA}; ")
assert f"guid={guid}" in parsed.sCookie
assert parsed.sTraceId == trace_id
user = HuyaHttpClient._build_activity_user(uid, cookie)
assert user.sHuYaUA == HTTP_HUYA_UA
assert user.sCookie == cookie
assert user.sGuid == ""
assert user.sCookie == parsed.sCookie
empty_user = HuyaHttpClient._build_activity_user(0, "")
assert empty_user.sCookie == ""
if __name__ == "__main__":