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

type: 收窄虎牙 HTTP 与 WSS 客户端类型

Browse Source
This commit is contained in:
yml2213
2026-08-30 20:04:08 +08:00
parent ec3c14d0a3
commit 6439dc1942
2 changed files with 445 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/http_client.py
+235 -89
View File
@@ -6,6 +6,7 @@
Body: Wup 包 (含4字节长度前缀)
Response: Wup 包 (tRsp)
"""
import base64
import hashlib
import json
@@ -13,7 +14,7 @@ import random
import struct
import urllib.parse
import urllib.request
from typing import Optional, Callable
from typing import Any, Optional, Callable
from .cookie_utils import cookie_pairs, normalize_cookie_pairs, normalize_huya_cookie
from .taf_protocol import TafOutputStream, TafInputStream, TafType, TafStruct
@@ -21,9 +22,11 @@ from .wup_protocol import WupRequest, WupResponse
CDNWS_HOST = "cdnws.api.huya.com"
LIVELINK_BIND_API_BASE = "https://livelinkbind.game.qq.com"
PC_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/149.0.0.0 Safari/537.36")
PC_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/149.0.0.0 Safari/537.36"
)
HTTP_HUYA_UA = "webh5&0.0.1&websocket&&diypc_52775"
@@ -42,6 +45,7 @@ class WSConnectParaInfo(TafStruct):
tag9 sTraceId string "hex8:hex8:0:0"
tag10 mCustomHeaders Map<string,string>
"""
def __init__(self):
self.lUid: int = 0
self.sGuid: str = ""
@@ -88,12 +92,13 @@ class WSConnectParaInfo(TafStruct):
def _gen_trace_id() -> str:
"""生成 sTraceId (格式 hex8:hex8:0:0，HAR 实证)"""
h = '%016x' % random.getrandbits(64)
h = "%016x" % random.getrandbits(64)
return f"{h}:{h}:0:0"
def generate_http_baseinfo(uid: int, guid: str, cookie: str,
trace_id: str = None) -> str:
def generate_http_baseinfo(
uid: int, guid: str, cookie: str, trace_id: str | None = None
) -> str:
"""
构造 HTTP POST 的 baseinfo URL 参数
@@ -111,14 +116,14 @@ def generate_http_baseinfo(uid: int, guid: str, cookie: str,
info.write_to(os)
raw = os.get_bytes()
# base64 (与 JS window.btoa 一致)
b64 = base64.b64encode(raw).decode('ascii')
return urllib.parse.quote(b64, safe='')
b64 = base64.b64encode(raw).decode("ascii")
return urllib.parse.quote(b64, safe="")
class HuyaHttpClient:
"""虎牙 HTTP POST RPC 客户端"""
def __init__(self, logger: Callable[[str], None] = None):
def __init__(self, logger: Callable[[str], None] | None = None):
self.logger = logger or print
@staticmethod
@@ -155,7 +160,11 @@ class HuyaHttpClient:
"""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"]
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:
@@ -168,7 +177,7 @@ class HuyaHttpClient:
uid: int,
guid: str,
cookie: str,
trace_id: str = None,
trace_id: str | None = None,
) -> str:
"""生成与浏览器 cdnws 调用一致的 baseinfo。"""
rpc_cookie = cls._normalize_cookie(cookie)
@@ -179,6 +188,7 @@ class HuyaHttpClient:
def _build_user(uid: int, guid: str, cookie: str):
"""构造 HTTP 业务 UserId。"""
from .shop_structs import UserId
user = UserId()
user.lUid = uid
user.sGuid = guid or HuyaHttpClient._resolve_cookie_guid(cookie)
@@ -194,6 +204,7 @@ class HuyaHttpClient:
def _build_shop_app(source_id: str = "yellowcarlist", scene: int = 7):
"""构造 HTTP 业务 ShopAppInfo。"""
from .shop_structs import ShopAppInfo
info = ShopAppInfo()
info.sAppId = "huya"
info.sBizType = ""
@@ -205,6 +216,7 @@ class HuyaHttpClient:
def _build_order_env(item_count: int = 1):
"""构造下单/支付环境参数。"""
import time
click_time = int(time.time() * 1000)
return {
"user_click_seq": f'[{{"x":258,"y":487,"time":{click_time},"id":""}}]',
@@ -215,16 +227,24 @@ class HuyaHttpClient:
def _build_activity_user(uid: int, cookie: str):
"""构造活动组件 UserId，与当前积分兑换组件保持一致。"""
from .activity_structs import ActivityUserId
user = ActivityUserId()
user.lUid = uid
user.sHuYaUA = HTTP_HUYA_UA
user.sCookie = HuyaHttpClient._normalize_cookie(cookie)
return user
def call_rpc(self, service: str, method: str,
req_struct: TafStruct, rsp_class=None,
uid: int = 0, guid: str = "", cookie: str = "",
timeout: float = 15.0):
def call_rpc(
self,
service: str,
method: str,
req_struct: TafStruct,
rsp_class=None,
uid: int = 0,
guid: str = "",
cookie: str = "",
timeout: float = 15.0,
):
"""
HTTP POST RPC 调用
@@ -249,28 +269,38 @@ class HuyaHttpClient:
self.logger(f"[HTTP] POST {service}.{method}")
self.logger(f"[HTTP] URL: https://{CDNWS_HOST}/?baseinfo=<redacted>")
self.logger(f"[HTTP] 发送 body: {len(wup_data)} 字节, hex前60: {wup_data[:60].hex()}")
self.logger(
f"[HTTP] 发送 body: {len(wup_data)} 字节, hex前60: {wup_data[:60].hex()}"
)
# 3. POST
req = urllib.request.Request(url, data=wup_data, method='POST', headers={
'User-Agent': PC_UA,
'Origin': 'https://zt.huya.com',
'Referer': 'https://zt.huya.com/',
'Content-Type': 'application/octet-stream',
'Accept-Language': 'zh-CN,zh;q=0.9',
})
req = urllib.request.Request(
url,
data=wup_data,
method="POST",
headers={
"User-Agent": PC_UA,
"Origin": "https://zt.huya.com",
"Referer": "https://zt.huya.com/",
"Content-Type": "application/octet-stream",
"Accept-Language": "zh-CN,zh;q=0.9",
},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
resp_data = resp.read()
# 检测 gzip 压缩 (响应 hex 1f8b 开头)
if resp_data[:2] == b'\x1f\x8b':
if resp_data[:2] == b"\x1f\x8b":
import gzip
resp_data = gzip.decompress(resp_data)
except Exception as e:
self.logger(f"[HTTP] ❌ 请求失败: {type(e).__name__}: {e}")
return None
self.logger(f"[HTTP] 收到响应: {len(resp_data)} 字节, hex前60: {resp_data[:60].hex()}")
self.logger(
f"[HTTP] 收到响应: {len(resp_data)} 字节, hex前60: {resp_data[:60].hex()}"
)
# 4. 解析响应 Wup
wup_resp = WupResponse()
@@ -289,7 +319,12 @@ class HuyaHttpClient:
if result is None:
result = wup_resp.readStruct("tResp", rsp_class)
if result is not None and hasattr(result, "to_dict"):
log_data = result.to_log_dict() if hasattr(result, "to_log_dict") else result.to_dict()
decoded_result: Any = result
log_data = (
decoded_result.to_log_dict()
if hasattr(decoded_result, "to_log_dict")
else decoded_result.to_dict()
)
decoded = json.dumps(log_data, ensure_ascii=False, separators=(",", ":"))
self.logger(f"[HTTP] 解码 tRsp: {decoded}")
return result
@@ -299,6 +334,7 @@ class HuyaHttpClient:
def query_user_score(self, uid: int, cookie: str, sid: int, timeout: float = 15.0):
"""查询活动可用积分。"""
from .activity_structs import GetUserScoreReq, GetUserScoreResp
req = GetUserScoreReq()
req.userId = self._build_activity_user(uid, cookie)
req.sid = sid
@@ -312,9 +348,12 @@ class HuyaHttpClient:
timeout=timeout,
)
def get_act_prize_list(self, uid: int, cookie: str, sid: int, timeout: float = 15.0):
def get_act_prize_list(
self, uid: int, cookie: str, sid: int, timeout: float = 15.0
):
"""查询活动可兑换商品列表。"""
from .activity_structs import GetActPrizeListReq, GetActPrizeListResp
req = GetActPrizeListReq()
req.userId = self._build_activity_user(uid, cookie)
req.sid = sid
@@ -328,9 +367,12 @@ class HuyaHttpClient:
timeout=timeout,
)
def get_user_prize_records(self, uid: int, cookie: str, sid: int, timeout: float = 15.0):
def get_user_prize_records(
self, uid: int, cookie: str, sid: int, timeout: float = 15.0
):
"""查询用户兑换记录。"""
from .activity_structs import GetUserPrizeRecordsReq, GetUserPrizeRecordsResp
req = GetUserPrizeRecordsReq()
req.userId = self._build_activity_user(uid, cookie)
req.sid = sid
@@ -344,9 +386,12 @@ class HuyaHttpClient:
timeout=timeout,
)
def score_exchange_prize(self, uid: int, cookie: str, sid: int, pid: int, timeout: float = 15.0):
def score_exchange_prize(
self, uid: int, cookie: str, sid: int, pid: int, timeout: float = 15.0
):
"""兑换活动积分商品。"""
from .activity_structs import ScoreExchangePrizeReq, ScoreExchangePrizeResp
req = ScoreExchangePrizeReq()
req.userId = self._build_activity_user(uid, cookie)
req.sid = int(sid or 0)
@@ -361,9 +406,12 @@ class HuyaHttpClient:
timeout=timeout,
)
def get_act_task_detail(self, uid: int, cookie: str, act_id: int, timeout: float = 15.0):
def get_act_task_detail(
self, uid: int, cookie: str, act_id: int, timeout: float = 15.0
):
"""查询活动任务详情，用于发现充值商品 SPU。"""
from .activity_structs import GetActTaskDetailReq, GetActTaskDetailResp
req = GetActTaskDetailReq()
req.userId = self._build_activity_user(uid, cookie)
req.actId = int(act_id or 0)
@@ -389,7 +437,11 @@ class HuyaHttpClient:
timeout: float = 15.0,
):
"""查询用户游戏账号绑定状态。"""
from .activity_structs import CheckUserBindGameAccountReq, CheckUserBindGameAccountResp
from .activity_structs import (
CheckUserBindGameAccountReq,
CheckUserBindGameAccountResp,
)
req = CheckUserBindGameAccountReq()
req.userId = self._build_activity_user(uid, cookie)
req.gid = int(gid or 0)
@@ -418,7 +470,11 @@ class HuyaHttpClient:
timeout: float = 15.0,
):
"""确认虎牙号与当前活动的游戏账号绑定关系。"""
from .activity_structs import ConfirmBindActAccountReq, ConfirmBindActAccountResp
from .activity_structs import (
ConfirmBindActAccountReq,
ConfirmBindActAccountResp,
)
req = ConfirmBindActAccountReq()
req.userId = self._build_activity_user(uid, cookie)
req.gid = int(gid or 0)
@@ -446,6 +502,7 @@ class HuyaHttpClient:
):
"""获取绑定二维码拉起参数。"""
from .activity_structs import GetLiveLinkParamReq, GetLiveLinkParamResp
req = GetLiveLinkParamReq()
req.userId = self._build_activity_user(uid, cookie)
req.gid = int(gid or 0)
@@ -463,9 +520,12 @@ class HuyaHttpClient:
timeout=timeout,
)
def get_user_profile_batch(self, uid: int, cookie: str, target_uids: list[int], timeout: float = 15.0):
def get_user_profile_batch(
self, uid: int, cookie: str, target_uids: list[int], timeout: float = 15.0
):
"""批量查询虎牙用户资料，用于补齐绑定页头像和昵称。"""
from .activity_structs import GetUserProfileBatchReq, GetUserProfileBatchResp
req = GetUserProfileBatchReq()
req.userId = self._build_activity_user(0, "")
req.uidList = [int(item) for item in target_uids if int(item or 0)]
@@ -493,7 +553,9 @@ class HuyaHttpClient:
mime = "image/png" if text.startswith("iVBOR") else "image/jpeg"
return f"data:{mime};base64,{text}"
def get_livelink_mini_qrcode(self, bind_page_url: str, timeout: float = 15.0) -> dict | None:
def get_livelink_mini_qrcode(
self, bind_page_url: str, timeout: float = 15.0
) -> dict | None:
"""请求绑定页内部接口，获取真正可扫的小程序码图片。"""
import gzip
import time
@@ -522,24 +584,32 @@ class HuyaHttpClient:
self.logger(
f"[LIVELINK] POST getMiniAppQrcode plat={live_plat_id or '-'} actId={act_id or '-'}"
)
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",
"Gap-Plat-Id": live_plat_id,
"Gap-Act-Id": act_id,
"Gap-User-From": "pc",
"Gap-From-Id": from_id,
"Vulcan-AccType": "auth_in_h5",
"Vulcan-PlatUser-Session": "",
})
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",
"Gap-Plat-Id": live_plat_id,
"Gap-Act-Id": act_id,
"Gap-User-From": "pc",
"Gap-From-Id": from_id,
"Vulcan-AccType": "auth_in_h5",
"Vulcan-PlatUser-Session": "",
},
)
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":
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}")
@@ -552,7 +622,9 @@ class HuyaHttpClient:
return None
if data.get("iRet") != 0:
self.logger(f"[LIVELINK] ❌ 小程序码接口失败: iRet={data.get('iRet')} msg={data.get('sMsg') or '-'}")
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 {}
@@ -562,13 +634,17 @@ class HuyaHttpClient:
self.logger("[LIVELINK] ❌ 小程序码接口未返回 qrcode")
return None
self.logger(f"[LIVELINK] 小程序码获取成功: image={len(image_src)} 字符 token={'有' if jdata.get('qrcodeToken') else '无'}")
self.logger(
f"[LIVELINK] 小程序码获取成功: image={len(image_src)} 字符 token={'有' if jdata.get('qrcodeToken') else '无'}"
)
return {
"mini_qrcode_image": image_src,
"qrcode_token": jdata.get("qrcodeToken") or "",
}
def get_livelink_qrcode_status(self, qrcode_token: str, timeout: float = 15.0) -> dict | None:
def get_livelink_qrcode_status(
self, qrcode_token: str, timeout: float = 15.0
) -> dict | None:
"""轮询 livelink 小程序码扫码状态。"""
import gzip
@@ -582,18 +658,26 @@ class HuyaHttpClient:
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",
})
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":
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}")
@@ -602,11 +686,15 @@ class HuyaHttpClient:
try:
data = json.loads(resp_data.decode("utf-8", "replace"))
except Exception as e:
self.logger(f"[LIVELINK] ❌ 二维码状态响应解析失败: {type(e).__name__}: {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 '-'}")
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 {}
@@ -675,9 +763,20 @@ class HuyaHttpClient:
"exchange_qq_coin_url": f"{huya_base}?{urlencode(exchange_params)}",
}
def get_goods_info(self, uid, guid, cookie, pid, spu_id, sku_id=0, game_id="",
source_id="yellowcarlist", scene=7):
def get_goods_info(
self,
uid,
guid,
cookie,
pid,
spu_id,
sku_id=0,
game_id="",
source_id="yellowcarlist",
scene=7,
):
from .shop_structs import GetGoodsInfoReqV5, GoodsInfoRsp
req = GetGoodsInfoReqV5()
req.userId = self._build_user(uid, guid, cookie)
req.shopAppInfo = self._build_shop_app(source_id, scene)
@@ -685,11 +784,21 @@ class HuyaHttpClient:
req.spuId = spu_id
req.skuId = sku_id
req.gameId = game_id
return self.call_rpc("shopMiddleUI", "getGoodsInfoV5", req,
GoodsInfoRsp, uid, guid, cookie)
return self.call_rpc(
"shopMiddleUI", "getGoodsInfoV5", req, GoodsInfoRsp, uid, guid, cookie
)
def query_user_order_list(self, uid, guid, cookie, offset=0, page_size=10,
order_type=1, status=0, timeout=15.0):
def query_user_order_list(
self,
uid,
guid,
cookie,
offset=0,
page_size=10,
order_type=1,
status=0,
timeout=15.0,
):
"""查询用户订单列表，用于支付二维码后的付款状态轮询。"""
from .shop_structs import QueryUserOrderListReq, QueryUserOrderListRsp
@@ -724,13 +833,28 @@ class HuyaHttpClient:
)
return fallback if fallback is not None else result
def create_order(self, uid, guid, cookie, pid, spu_id, sku_id,
item_count=1, source_id="yellowcarlist", game_id="", scene=7,
order_type=None):
def create_order(
self,
uid,
guid,
cookie,
pid,
spu_id,
sku_id,
item_count=1,
source_id="yellowcarlist",
game_id="",
scene=7,
order_type=None,
):
"""order_type=None 时自动尝试从 1 到 10 找到有效值"""
from .shop_structs import (CreateOrderReqV5, CreateOrderRsp,
CreateOrderExtraParam, CreateOrderPromotionParam,
CreateOrderAccountParam)
from .shop_structs import (
CreateOrderReqV5,
CreateOrderRsp,
CreateOrderExtraParam,
CreateOrderPromotionParam,
CreateOrderAccountParam,
)
# 如果指定了具体值，直接试
types_to_try = [order_type] if order_type is not None else list(range(1, 11))
@@ -759,10 +883,13 @@ class HuyaHttpClient:
req.gameCategoryId = 507
req.ext = ""
result = self.call_rpc("shopMiddleUI", "createOrderV5", req,
CreateOrderRsp, uid, guid, cookie)
result: Any = self.call_rpc(
"shopMiddleUI", "createOrderV5", req, CreateOrderRsp, uid, guid, cookie
)
if result is not None:
self.logger(f" orderType={ot}: code={result.code} orderId={result.orderId} msg={result.message}")
self.logger(
f" orderType={ot}: code={result.code} orderId={result.orderId} msg={result.message}"
)
if result.code == 200 and result.orderId:
self.logger(f"[✓] 找到有效 orderType={ot}")
return result
@@ -781,11 +908,21 @@ class HuyaHttpClient:
return "Weixin"
return text
def pay_order_submit(self, uid, guid, cookie, order_id, pay_type=1,
pid=0, source_id="yellowcarlist", scene=7,
item_count=1):
def pay_order_submit(
self,
uid,
guid,
cookie,
order_id,
pay_type=1,
pid=0,
source_id="yellowcarlist",
scene=7,
item_count=1,
):
"""发起支付，结构与 WSS HAR 中 payOrderSubmitV5 对齐。"""
from .shop_structs import PayOrderRes
os = TafOutputStream()
os.write_struct_begin(0)
@@ -817,22 +954,31 @@ class HuyaHttpClient:
self.logger(f"[HTTP] POST shopMiddleUI.payOrderSubmitV5")
self.logger(f"[HTTP] 发送 hex前60: {wup_data[:60].hex()}")
req = urllib.request.Request(url, data=wup_data, method='POST', headers={
'User-Agent': PC_UA, 'Origin': 'https://zt.huya.com',
'Referer': 'https://zt.huya.com/',
'Content-Type': 'application/octet-stream',
})
req = urllib.request.Request(
url,
data=wup_data,
method="POST",
headers={
"User-Agent": PC_UA,
"Origin": "https://zt.huya.com",
"Referer": "https://zt.huya.com/",
"Content-Type": "application/octet-stream",
},
)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
resp_data = resp.read()
if resp_data[:2] == b'\x1f\x8b':
if resp_data[:2] == b"\x1f\x8b":
import gzip
resp_data = gzip.decompress(resp_data)
except Exception as e:
self.logger(f"[HTTP] ❌ payOrderSubmitV5 失败: {e}")
return None
self.logger(f"[HTTP] payOrderSubmitV5 响应 {len(resp_data)} 字节, hex前60: {resp_data[:60].hex()}")
self.logger(
f"[HTTP] payOrderSubmitV5 响应 {len(resp_data)} 字节, hex前60: {resp_data[:60].hex()}"
)
wup_resp = WupResponse()
wup_resp.decode(resp_data)
if wup_resp.newdata: