From 6439dc19420eca820e0f62caaf6700ae7fd9a8ed Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 30 Aug 2026 20:04:08 +0800 Subject: [PATCH] =?UTF-8?q?type:=20=E6=94=B6=E7=AA=84=E8=99=8E=E7=89=99=20?= =?UTF-8?q?HTTP=20=E4=B8=8E=20WSS=20=E5=AE=A2=E6=88=B7=E7=AB=AF=E7=B1=BB?= =?UTF-8?q?=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/huya/http_client.py | 324 ++++++++++++++++++++++++++++----------- core/huya/wss_client.py | 322 ++++++++++++++++++++++++-------------- 2 files changed, 445 insertions(+), 201 deletions(-) diff --git a/core/huya/http_client.py b/core/huya/http_client.py index bdabe4e..66271b6 100644 --- a/core/huya/http_client.py +++ b/core/huya/http_client.py @@ -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 """ + 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=") - 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: diff --git a/core/huya/wss_client.py b/core/huya/wss_client.py index 7fa04ec..77c3bec 100644 --- a/core/huya/wss_client.py +++ b/core/huya/wss_client.py @@ -14,13 +14,14 @@ - 服务端响应 seq ≠ 请求 seq → 必须用 FIFO 匹配 - 初始化 wsLaunch iRequestId=-1，业务 requestId 从 8 开始递增 """ + import asyncio import random import re import struct import time from collections import deque -from typing import Optional, Callable +from typing import Any, Optional, Callable, cast import websockets @@ -41,19 +42,19 @@ WSS_COOKIE_UA = "webh5&0.0.1&websocket&&diypc_52775" WSS_SHOP_SCENE = 4 # seq 按消息类型固定 (HAR 实证，跨两次抓包一致) -SEQ_HEARTBEAT = 0x1d00000d -SEQ_WSLAUNCH = 0x1d000078 -SEQ_GETCONFIG = 0x1d000076 -SEQ_REGISTER = 0x1d00003c -SEQ_CONFIRM = 0x1d000002 -SEQ_BUSINESS = 0x1d000106 # 业务 RPC (HAR 7.4 实证：业务帧固定使用该 seq) +SEQ_HEARTBEAT = 0x1D00000D +SEQ_WSLAUNCH = 0x1D000078 +SEQ_GETCONFIG = 0x1D000076 +SEQ_REGISTER = 0x1D00003C +SEQ_CONFIRM = 0x1D000002 +SEQ_BUSINESS = 0x1D000106 # 业务 RPC (HAR 7.4 实证：业务帧固定使用该 seq) class WssCommand: RPC_REQUEST = 0x03 RPC_RESPONSE = 0x04 - AUTH = 0x0a - SERVER_PUSH1 = 0x0b + AUTH = 0x0A + SERVER_PUSH1 = 0x0B HEARTBEAT_SEND = 0x10 HEARTBEAT_RECV = 0x11 CONFIRM_SEND = 0x17 @@ -68,15 +69,15 @@ class WssMessage: self.body = body def encode(self) -> bytes: - header = struct.pack('>BBI', 0x00, self.command, self.sequence) + header = struct.pack(">BBI", 0x00, self.command, self.sequence) return header + self.body @classmethod - def decode(cls, data: bytes) -> 'WssMessage': + def decode(cls, data: bytes) -> "WssMessage": if len(data) < 6: raise ValueError(f"消息太短: {len(data)} bytes") - version, command = struct.unpack('>BB', data[0:2]) - sequence = struct.unpack('>I', data[2:6])[0] + version, command = struct.unpack(">BB", data[0:2]) + sequence = struct.unpack(">I", data[2:6])[0] body = data[6:] return cls(command=command, sequence=sequence, body=body) @@ -84,9 +85,11 @@ class WssMessage: class HuyaWssClient: """虎牙 WSS 客户端 — 商城通道""" - def __init__(self, baseinfo: str = None, logger: Callable[[str], None] = None): + def __init__( + self, baseinfo: str | None = None, logger: Callable[[str], None] | None = None + ): self.baseinfo = baseinfo or SHOP_BASEINFO - self.ws = None + self.ws: Any = None self._biz_seq = SEQ_BUSINESS # 业务 RPC 递增用 self._wup_req_id = 8 self.rpc_queue = deque() @@ -135,7 +138,12 @@ class HuyaWssClient: # 大包头里的长度使用原始 WUP 包长度（含 4B 长度前缀），但正文只放裸 WUP body； # 浏览器随后用 trace 尾部的前 4 字节补足该长度窗口。 packet_len = len(wup_data) - return bytes([packet_len & 0xFF]) + packet_len.to_bytes(4, "big") + wup_body + HuyaWssClient._make_big_tail() + return ( + bytes([packet_len & 0xFF]) + + packet_len.to_bytes(4, "big") + + wup_body + + HuyaWssClient._make_big_tail() + ) @staticmethod def _normalize_biz_cookie(cookie: str) -> str: @@ -143,7 +151,12 @@ class HuyaWssClient: cookie = (cookie or "").strip() normalized = f"huya_ua={WSS_COOKIE_UA}" if re.search(r"(?:^|;\s*)huya_ua=", cookie): - return re.sub(r"(^|;\s*)huya_ua=[^;]*", lambda m: f"{m.group(1)}{normalized}", cookie, count=1) + return re.sub( + r"(^|;\s*)huya_ua=[^;]*", + lambda m: f"{m.group(1)}{normalized}", + cookie, + count=1, + ) if not cookie: return normalized return f"{normalized}; {cookie}" @@ -152,6 +165,7 @@ class HuyaWssClient: def _build_biz_user(uid: int, cookie: str): """构造 WSS 业务 UserId（HAR 实证：guid 为空，cookie 不额外加前缀）""" from .shop_structs import UserId + user = UserId() user.lUid = uid user.sGuid = "" @@ -167,6 +181,7 @@ class HuyaWssClient: def _build_shop_app(source_id: str = "yellowcarlist", scene: int = WSS_SHOP_SCENE): """构造 WSS 业务 ShopAppInfo""" from .shop_structs import ShopAppInfo + info = ShopAppInfo() info.sAppId = "huya" info.sBizType = "" @@ -183,28 +198,33 @@ class HuyaWssClient: "input_1": str(item_count), } - async def connect(self, host: str = SHOP_WS_HOST, timeout: float = 15.0, - cookie: str = ""): + async def connect( + self, host: str = SHOP_WS_HOST, timeout: float = 15.0, cookie: str = "" + ): from urllib.parse import quote + url = f"wss://{host}/?baseinfo={quote(self.baseinfo)}" self.logger(f"[WSS] 正在连接 {host} ...") self.logger(f"[WSS] URL: {url[:100]}...") - BROWSER_UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/149.0.0.0 Safari/537.36") + BROWSER_UA = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/149.0.0.0 Safari/537.36" + ) headers = {"Accept-Language": "zh-CN,zh;q=0.9"} try: self.ws = await asyncio.wait_for( websockets.connect( url, - origin="https://m-shop.yaoguo.com", + origin=cast(Any, "https://m-shop.yaoguo.com"), user_agent_header=BROWSER_UA, additional_headers=headers, open_timeout=timeout, ), - timeout=timeout) + timeout=timeout, + ) except asyncio.TimeoutError: self.logger(f"[WSS] 连接超时（{timeout}s）") raise @@ -230,11 +250,16 @@ class HuyaWssClient: try: async for raw in self.ws: try: - msg = WssMessage.decode(raw) - self.logger(format_wss_log(msg.body, msg.command, msg.sequence, "收")) + raw_bytes = raw if isinstance(raw, bytes) else raw.encode() + msg = WssMessage.decode(raw_bytes) + self.logger( + format_wss_log(msg.body, msg.command, msg.sequence, "收") + ) await self._handle_message(msg) except Exception as e: - self.logger(f"[WSS] 解析消息失败: {e} raw_hex={raw[:50].hex()}") + self.logger( + f"[WSS] 解析消息失败: {e} raw_hex={raw_bytes[:50].hex()}" + ) except asyncio.CancelledError: pass except websockets.exceptions.ConnectionClosed as e: @@ -270,14 +295,17 @@ class HuyaWssClient: async def send_heartbeat(self): # body = LIST(tag0,count=1) + 元素{tag0="live:0", tag1=""} + TAIL os = TafOutputStream() - os.write_head(0, TafType.LIST) # 09 - os.write_int32(0, 1) # 00 01 (count=1) - os.write_string(0, "live:0") # 06 06 live:0 - os.buf.write(b'\x16\x00') # 16 00 (tag1 "") + os.write_head(0, TafType.LIST) # 09 + os.write_int32(0, 1) # 00 01 (count=1) + os.write_string(0, "live:0") # 06 06 live:0 + os.buf.write(b"\x16\x00") # 16 00 (tag1 "") body = os.get_bytes() + TAIL_BYTES - msg = WssMessage(command=WssCommand.HEARTBEAT_SEND, - sequence=SEQ_HEARTBEAT, body=body) - self.logger(format_wss_log(body, WssCommand.HEARTBEAT_SEND, SEQ_HEARTBEAT, "发")) + msg = WssMessage( + command=WssCommand.HEARTBEAT_SEND, sequence=SEQ_HEARTBEAT, body=body + ) + self.logger( + format_wss_log(body, WssCommand.HEARTBEAT_SEND, SEQ_HEARTBEAT, "发") + ) await self.ws.send(msg.encode()) async def initialize(self, uid: int, guid: str, cookie: str): @@ -295,10 +323,14 @@ class HuyaWssClient: if launch_rsp is None: self.logger("[✗] wsLaunch 无响应，初始化失败") return False - self.logger(f"[初始化] wsLaunch OK guid={self._launch_guid} ip={self._launch_ip}") + self.logger( + f"[初始化] wsLaunch OK guid={self._launch_guid} ip={self._launch_ip}" + ) config_rsp = await self.call_get_config_shop() - self.logger(f"[初始化] getConfig {'OK' if config_rsp is not None else '无响应(可继续)'}") + self.logger( + f"[初始化] getConfig {'OK' if config_rsp is not None else '无响应(可继续)'}" + ) if self._launch_guid: await self.send_register(self._launch_guid) @@ -316,11 +348,11 @@ class HuyaWssClient: os = TafOutputStream() os.write_struct_begin(0) - os.write_int64(0, 0) # lUid = 0 - os.write_string(1, "") # sGuid = "" - os.write_string(2, ua) # sUA - os.write_string(3, app_src) # sAppSrc - os.write_struct_begin(4) # tDeviceInfo + os.write_int64(0, 0) # lUid = 0 + os.write_string(1, "") # sGuid = "" + os.write_string(2, ua) # sUA + os.write_string(3, app_src) # sAppSrc + os.write_struct_begin(4) # tDeviceInfo for i in range(5): os.write_string(i, "") os.write_struct_end() @@ -336,8 +368,7 @@ class HuyaWssClient: body = self._encode_rpc_body(wup_data) seq = SEQ_WSLAUNCH - msg = WssMessage(command=WssCommand.RPC_REQUEST, - sequence=seq, body=body) + msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=seq, body=body) future = asyncio.Future() self.rpc_queue.append(future) self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, seq, "发")) @@ -363,11 +394,13 @@ class HuyaWssClient: wup_resp.decode(wup) treq = wup_resp.newdata.get("tRsp") or wup_resp.newdata.get("tResp") if not treq: - self.logger(f"[RPC] wsLaunch 响应无 tRsp, keys={list(wup_resp.newdata.keys())}") + self.logger( + f"[RPC] wsLaunch 响应无 tRsp, keys={list(wup_resp.newdata.keys())}" + ) return ins = TafInputStream(treq) tag, dtype = ins.peek_head() - if dtype != 0x0a: # STRUCT_BEGIN + if dtype != 0x0A: # STRUCT_BEGIN self.logger(f"[RPC] wsLaunch tRsp 非结构体 dtype=0x{dtype:02x}") return ins.read_head() @@ -376,19 +409,29 @@ class HuyaWssClient: ftag, ftype = ins.peek_head() except EOFError: break - if ftype == 0x0b: # STRUCT_END + if ftype == 0x0B: # STRUCT_END ins.read_head() break ins.read_head() if ftag == 0 and ftype in (0x06, 0x07): - ln = ins.buf.read(1)[0] if ftype == 0x06 else int.from_bytes(ins.buf.read(4), 'big') - self._launch_guid = ins.buf.read(ln).decode('utf-8', 'replace') + ln = ( + ins.buf.read(1)[0] + if ftype == 0x06 + else int.from_bytes(ins.buf.read(4), "big") + ) + self._launch_guid = ins.buf.read(ln).decode("utf-8", "replace") elif ftag == 1 and ftype in (0x06, 0x07): - ln = ins.buf.read(1)[0] if ftype == 0x06 else int.from_bytes(ins.buf.read(4), 'big') - self._launch_ip = ins.buf.read(ln).decode('utf-8', 'replace') + ln = ( + ins.buf.read(1)[0] + if ftype == 0x06 + else int.from_bytes(ins.buf.read(4), "big") + ) + self._launch_ip = ins.buf.read(ln).decode("utf-8", "replace") else: ins.skip_field(ftype) - self.logger(f"[RPC] wsLaunch 解析: guid={self._launch_guid} ip={self._launch_ip}") + self.logger( + f"[RPC] wsLaunch 解析: guid={self._launch_guid} ip={self._launch_ip}" + ) except Exception as e: self.logger(f"[RPC] wsLaunch 响应解析失败: {e}") @@ -403,27 +446,36 @@ class HuyaWssClient: if len(body) < 5: return body # 大包格式需优先判断；prefix 可能是 0x00，容易被误判成小包长度 8。 - wup_len = int.from_bytes(body[1:5], 'big') - if 5 + wup_len <= len(body) and body[5:7] == b'\x10\x03': - return body[5:5 + wup_len] + wup_len = int.from_bytes(body[1:5], "big") + if 5 + wup_len <= len(body) and body[5:7] == b"\x10\x03": + return body[5 : 5 + wup_len] # 尝试 4B total_len 小包格式 - total_len = int.from_bytes(body[0:4], 'big') - if 8 <= total_len <= len(body) and body[4:6] == b'\x10\x03': + total_len = int.from_bytes(body[0:4], "big") + if 8 <= total_len <= len(body) and body[4:6] == b"\x10\x03": # total_len 包含自身 4B return body[:total_len] if 5 + wup_len <= len(body): - return body[5:5 + wup_len] + return body[5 : 5 + wup_len] return body async def send_auth(self, cookie: str): """cmd 0x0a AUTH — 发送 cookie""" ua = "webh5&0.0.1&websocket&&diypc_52775" auth_text = f"huya_ua={ua}; {cookie}" - msg = WssMessage(command=WssCommand.AUTH, - sequence=SEQ_WSLAUNCH, - body=auth_text.encode('utf-8') + TAIL_BYTES) + msg = WssMessage( + command=WssCommand.AUTH, + sequence=SEQ_WSLAUNCH, + body=auth_text.encode("utf-8") + TAIL_BYTES, + ) await self.ws.send(msg.encode()) - self.logger(format_wss_log(auth_text.encode('utf-8') + TAIL_BYTES, WssCommand.AUTH, SEQ_WSLAUNCH, "发")) + self.logger( + format_wss_log( + auth_text.encode("utf-8") + TAIL_BYTES, + WssCommand.AUTH, + SEQ_WSLAUNCH, + "发", + ) + ) async def call_get_config_shop(self, timeout: float = 10.0): """商城版 mobileui.getConfig @@ -441,18 +493,18 @@ class HuyaWssClient: os.write_struct_begin(0) # tag0: UserId (匿名) os.write_struct_begin(0) - os.write_int64(0, 0) # lUid = 0 - os.write_string(1, "") # sGuid - os.write_string(2, "") # sUA - os.write_string(3, ua) # sHuYaUA - os.write_string(4, "") # sToken - os.write_int32(5, 0) # iTokenType - os.write_string(6, "") # sDeviceInfo - os.write_string(7, "") # sCookie + os.write_int64(0, 0) # lUid = 0 + os.write_string(1, "") # sGuid + os.write_string(2, "") # sUA + os.write_string(3, ua) # sHuYaUA + os.write_string(4, "") # sToken + os.write_int32(5, 0) # iTokenType + os.write_string(6, "") # sDeviceInfo + os.write_string(7, "") # sCookie os.write_struct_end() - os.write_map(1, {}) # tag1: 空Map - os.write_string(2, "huya") # tag2: sAppId - os.write_head(3, TafType.LIST) # tag3: 空List + os.write_map(1, {}) # tag1: 空Map + os.write_string(2, "huya") # tag2: sAppId + os.write_head(3, TafType.LIST) # tag3: 空List os.write_int32(0, 0) os.write_struct_end() @@ -466,8 +518,7 @@ class HuyaWssClient: body = self._encode_rpc_body(wup_data) seq = SEQ_GETCONFIG - msg = WssMessage(command=WssCommand.RPC_REQUEST, - sequence=seq, body=body) + msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=seq, body=body) future = asyncio.Future() self.rpc_queue.append(future) self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, seq, "发")) @@ -503,8 +554,7 @@ class HuyaWssClient: os.write_map(7, {}) os.write_int32(8, 0) body = os.get_bytes() + TAIL_BYTES - msg = WssMessage(command=WssCommand.REGISTER, - sequence=SEQ_REGISTER, body=body) + msg = WssMessage(command=WssCommand.REGISTER, sequence=SEQ_REGISTER, body=body) self.logger(format_wss_log(body, WssCommand.REGISTER, SEQ_REGISTER, "发")) await self.ws.send(msg.encode()) @@ -513,14 +563,20 @@ class HuyaWssClient: os = TafOutputStream() os.write_map(0, {}) body = os.get_bytes() + TAIL_BYTES - msg = WssMessage(command=WssCommand.CONFIRM_SEND, - sequence=SEQ_CONFIRM, body=body) + msg = WssMessage( + command=WssCommand.CONFIRM_SEND, sequence=SEQ_CONFIRM, body=body + ) self.logger(format_wss_log(body, WssCommand.CONFIRM_SEND, SEQ_CONFIRM, "发")) await self.ws.send(msg.encode()) - async def call_rpc(self, service: str, method: str, - req_struct: TafStruct, rsp_class=None, - timeout: float = 10.0): + async def call_rpc( + self, + service: str, + method: str, + req_struct: TafStruct, + rsp_class=None, + timeout: float = 10.0, + ): seq = self._next_biz_seq() wup = WupRequest() @@ -532,8 +588,7 @@ class HuyaWssClient: wup_data = wup.encode() body = self._encode_rpc_body(wup_data) - msg = WssMessage(command=WssCommand.RPC_REQUEST, - sequence=seq, body=body) + msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=seq, body=body) future = asyncio.Future() self.rpc_queue.append(future) self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, seq, "发")) @@ -559,7 +614,9 @@ class HuyaWssClient: if dtype == TafType.STRUCT_BEGIN: ins.read_head() decoded = _decode_taf_struct(ins) - self.logger(f"[←] {service}.{method} {key}: {_truncate(decoded)}") + self.logger( + f"[←] {service}.{method} {key}: {_truncate(decoded)}" + ) except Exception: pass @@ -571,10 +628,18 @@ class HuyaWssClient: result = wup_resp.readStruct("tResp", rsp_class) return result - async def get_goods_info(self, uid: int, guid: str, cookie: str, - pid: int, spu_id: str, sku_id: int = 0, - game_id: str = "", source_id: str = "yellowcarlist", - scene: int = WSS_SHOP_SCENE): + async def get_goods_info( + self, + uid: int, + guid: str, + cookie: str, + pid: int, + spu_id: str, + sku_id: int = 0, + game_id: str = "", + source_id: str = "yellowcarlist", + scene: int = WSS_SHOP_SCENE, + ): from .shop_structs import GetGoodsInfoReqV5, GoodsInfoRsp req = GetGoodsInfoReqV5() @@ -585,12 +650,20 @@ class HuyaWssClient: req.skuId = sku_id req.gameId = game_id - return await self.call_rpc("shopMiddleUI", "getGoodsInfoV5", - req, GoodsInfoRsp, timeout=15.0) + return await self.call_rpc( + "shopMiddleUI", "getGoodsInfoV5", req, GoodsInfoRsp, timeout=15.0 + ) - async def query_user_order_list(self, uid: int, guid: str, cookie: str, - offset: int = 0, page_size: int = 10, - order_type: int = 1, status: int = 0): + async def query_user_order_list( + self, + uid: int, + guid: str, + cookie: str, + offset: int = 0, + page_size: int = 10, + order_type: int = 1, + status: int = 0, + ): from .shop_structs import QueryUserOrderListReq, QueryUserOrderListRsp req = QueryUserOrderListReq() @@ -600,17 +673,35 @@ class HuyaWssClient: req.pageSize = page_size req.status = status - return await self.call_rpc("revenueWebUI", "queryUserOrderList", - req, QueryUserOrderListRsp, timeout=15.0) + return await self.call_rpc( + "revenueWebUI", + "queryUserOrderList", + req, + QueryUserOrderListRsp, + timeout=15.0, + ) - async def create_order(self, uid: int, guid: str, cookie: str, - pid: int, spu_id: str, sku_id: int, - item_count: int = 1, source_id: str = "yellowcarlist", - game_id: str = "", scene: int = WSS_SHOP_SCENE, - order_type: int = 6): - from .shop_structs import (CreateOrderReqV5, CreateOrderRsp, - CreateOrderExtraParam, CreateOrderPromotionParam, - CreateOrderAccountParam) + async def create_order( + self, + uid: int, + guid: str, + cookie: str, + pid: int, + spu_id: str, + sku_id: int, + item_count: int = 1, + source_id: str = "yellowcarlist", + game_id: str = "", + scene: int = WSS_SHOP_SCENE, + order_type: int = 6, + ): + from .shop_structs import ( + CreateOrderReqV5, + CreateOrderRsp, + CreateOrderExtraParam, + CreateOrderPromotionParam, + CreateOrderAccountParam, + ) req = CreateOrderReqV5() req.userId = self._build_biz_user(uid, cookie) @@ -635,14 +726,22 @@ class HuyaWssClient: req.ext = "" self.logger(f"[下单] orderType={order_type} (HUYA_VIRTUAL)") - return await self.call_rpc("shopMiddleUI", "createOrderV5", - req, CreateOrderRsp, timeout=15.0) + return await self.call_rpc( + "shopMiddleUI", "createOrderV5", req, CreateOrderRsp, timeout=15.0 + ) - async def pay_order_submit(self, uid: int, guid: str, cookie: str, - order_id: int, pay_type: int = 1, - pid: int = 0, source_id: str = "yellowcarlist", - scene: int = WSS_SHOP_SCENE, - item_count: int = 1): + async def pay_order_submit( + self, + uid: int, + guid: str, + cookie: str, + order_id: int, + pay_type: int = 1, + pid: int = 0, + source_id: str = "yellowcarlist", + scene: int = WSS_SHOP_SCENE, + item_count: int = 1, + ): from .shop_structs import PayOrderRes os = TafOutputStream() @@ -674,8 +773,7 @@ class HuyaWssClient: wup_data = wup.encode() body = self._encode_rpc_body(wup_data) - msg = WssMessage(command=WssCommand.RPC_REQUEST, - sequence=seq, body=body) + msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=seq, body=body) future = asyncio.Future() self.rpc_queue.append(future) self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, seq, "发"))