""" 虎牙 HTTP POST RPC 通道（cdnws.api.huya.com） 用于 WSS 业务通道被设备态 Cookie 静默拒收时的兜底: URL: https://cdnws.api.huya.com/?baseinfo= Body: Wup 包 (含4字节长度前缀) Response: Wup 包 (tRsp) """ import base64 import hashlib import json import random import re import time import urllib.parse import urllib.request from collections.abc import Callable from typing import Any from .cookie_utils import cookie_pairs, normalize_cookie_pairs, normalize_huya_cookie from .frame_decoder import decode_struct_bytes from .taf_protocol import TafInputStream, TafOutputStream, TafStruct # 积分中心链路常量与 WSS 通道共用（wss_client 不反向依赖本模块，无循环导入） from .wss_client import ( GAME_CATEGORY_ID_HUYA, JIFENCENTER_BIZ_PARAMS, ORDER_BIZ_TYPE, ORDER_SCENE, POINTS_PAY_CALLBACK, SHOP_APP_SCENE, ) 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" ) HTTP_HUYA_UA = "webh5&0.0.1&websocket&&diypc_52775" SHOP_BIZ_UA = "web&1.0.0&huya" class WSConnectParaInfo(TafStruct): """WSS/HTTP 连接参数（baseinfo 的内容），从前端 SDK 逆向 tag0 lUid int64 tag1 sGuid string tag2 sUA string "webh5&0.0.1&websocket&&diypc_52775" tag3 sAppSrc string "HUYA&ZH&2052" tag4 sMid string tag5 sExp string tag6 iTokenType int32 tag7 sToken string tag8 sCookie string (lUid>0 时设 document.cookie) tag9 sTraceId string "hex8:hex8:0:0" tag10 mCustomHeaders Map """ def __init__(self): self.lUid: int = 0 self.sGuid: str = "" self.sUA: str = HTTP_HUYA_UA self.sAppSrc: str = "HUYA&ZH&2052" self.sMid: str = "" self.sExp: str = "" self.iTokenType: int = 0 self.sToken: str = "" self.sCookie: str = "" self.sTraceId: str = "" self.mCustomHeaders: dict = {} def write_to(self, os: TafOutputStream): # 与浏览器一致: 可为空的字段也显式写入。 if self.lUid: os.write_int64(0, self.lUid) else: os.write_int64(0, 0) # ZERO os.write_string(1, self.sGuid) os.write_string(2, self.sUA) os.write_string(3, self.sAppSrc) os.write_string(4, self.sMid) os.write_string(5, self.sExp) os.write_int32(6, self.iTokenType) os.write_string(7, self.sToken) os.write_string(8, self.sCookie) os.write_string(9, self.sTraceId) os.write_map(10, self.mCustomHeaders) def read_from(self, ins: TafInputStream): self.lUid = ins.read_int64(0, default=self.lUid) self.sGuid = ins.read_string(1, default=self.sGuid) self.sUA = ins.read_string(2, default=self.sUA) self.sAppSrc = ins.read_string(3, default=self.sAppSrc) self.sMid = ins.read_string(4, default=self.sMid) self.sExp = ins.read_string(5, default=self.sExp) self.iTokenType = ins.read_int32(6, default=self.iTokenType) self.sToken = ins.read_string(7, default=self.sToken) self.sCookie = ins.read_string(8, default=self.sCookie) self.sTraceId = ins.read_string(9, default=self.sTraceId) self.mCustomHeaders = ins.read_map(10) def _gen_trace_id() -> str: """生成 sTraceId (格式 hex8:hex8:0:0，HAR 实证)""" h = f"{random.getrandbits(64):016x}" return f"{h}:{h}:0:0" def generate_http_baseinfo( uid: int, guid: str, cookie: str, trace_id: str | None = None ) -> str: """ 构造 HTTP POST 的 baseinfo URL 参数 HAR 实证: 当前 cdnws HTTP RPC 的 baseinfo 会带 lUid 和完整 Cookie。 """ info = WSConnectParaInfo() info.lUid = int(uid or 0) info.sGuid = guid info.sUA = HTTP_HUYA_UA info.sAppSrc = "HUYA&ZH&2052" info.sCookie = cookie or "" info.sTraceId = trace_id or _gen_trace_id() os = TafOutputStream() info.write_to(os) raw = os.get_bytes() # base64 (与 JS window.btoa 一致) b64 = base64.b64encode(raw).decode("ascii") return urllib.parse.quote(b64, safe="") def generate_ws_baseinfo( uid: int, guid: str, ua: str, exp: str = "", ) -> str: """构造 WSS 连接 URL 的 baseinfo（返回未转义的 base64）。 2026-10-13 积分中心抓包实证：握手 baseinfo 只有 lUid/sGuid/sUA/sAppSrc/sExp 有值，cookie 与 traceId 都是空的； ``sExp`` 是 getConfig 下发的版本串（实测 ``58664.107263``）。 """ info = WSConnectParaInfo() info.lUid = int(uid or 0) info.sGuid = guid or "" info.sUA = ua info.sAppSrc = "HUYA&ZH&2052" info.sExp = exp or "" info.sCookie = "" info.sTraceId = "" os = TafOutputStream() info.write_to(os) return base64.b64encode(os.get_bytes()).decode("ascii") class HuyaHttpClient: """虎牙 HTTP POST RPC 客户端""" def __init__(self, logger: Callable[[str], None] | None = None): self.logger = logger or print @staticmethod 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 = 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 or HuyaHttpClient._resolve_cookie_guid(cookie) user.sToken = "" user.sHuYaUA = HTTP_HUYA_UA user.sCookie = HuyaHttpClient._normalize_cookie(cookie) user.iTokenType = 0 user.sDeviceInfo = "" user.sQIMEI = "" return user @staticmethod def _build_shop_user(uid: int, cookie: str): """构造商城业务 UserId；与 9.1 商城 WSS 帧保持一致。""" from .shop_structs import UserId user = UserId() user.lUid = int(uid or 0) user.sGuid = "" user.sToken = "" user.sHuYaUA = SHOP_BIZ_UA user.sCookie = HuyaHttpClient._normalize_cookie(cookie) user.iTokenType = 0 user.sDeviceInfo = "" user.sQIMEI = "" return user @staticmethod def _build_shop_app(source_id: str = "yellowcarlist", scene: int = SHOP_APP_SCENE): """构造 HTTP 业务 ShopAppInfo（积分中心抓包 scene=7）。""" from .shop_structs import ShopAppInfo info = ShopAppInfo() info.sAppId = "huya" info.sBizType = "" info.scene = scene info.sourceId = source_id return info @staticmethod def _build_order_env(): """构造下单/支付环境参数。 2026-10-13 积分中心抓包：仅一个空点击序列，没有旧版坐标数组与 input_1。 """ return {"user_click_seq": "[]"} @staticmethod 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, request_id: int = 1, origin: str = "https://zt.huya.com", referer: str = "https://zt.huya.com/", ): """ HTTP POST RPC 调用 Args: service: servant 名 (shopMiddleUI) method: 方法名 (getGoodsInfoV5) req_struct: 请求结构体 (TafStruct) rsp_class: 响应类 uid/guid/cookie: 用于构造 baseinfo，业务请求体也会按各接口需要携带 request_id: WUP iRequestId（launch.queryHttpDns 固定 -1） origin/referer: 部分接口（积分中心页面）要求与页面同源 """ # 1. 构造 Wup 请求体 wup = WupRequest() wup.setServant(service) wup.setFunc(method) wup.setRequestId(request_id) wup.writeStruct("tReq", req_struct) wup_data = wup.encode() # [4字节长度][wup body] # 2. 构造 baseinfo baseinfo = self._generate_rpc_baseinfo(uid, guid, cookie) url = f"https://{CDNWS_HOST}/?baseinfo={baseinfo}" 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()}" ) # 3. POST req = urllib.request.Request( url, data=wup_data, method="POST", headers={ "User-Agent": PC_UA, "Origin": origin, "Referer": referer, "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": import gzip resp_data = gzip.decompress(resp_data) except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 self.logger(f"[HTTP] ❌ 请求失败: {type(e).__name__}: {e}") return None self.logger( f"[HTTP] 收到响应: {len(resp_data)} 字节, hex前60: {resp_data[:60].hex()}" ) # 4. 解析响应 Wup wup_resp = WupResponse() wup_resp.decode(resp_data) if wup_resp.newdata: self.logger(f"[HTTP] 响应 newdata keys: {list(wup_resp.newdata.keys())}") for k, v in wup_resp.newdata.items(): self.logger(f"[HTTP] {k} hex({len(v)}): {v.hex()[:200]}") else: self.logger(f"[HTTP] 响应无 newdata (servant={wup_resp.sServantName})") if rsp_class is None: return wup_resp result = wup_resp.readStruct("tRsp", rsp_class) if result is None: result = wup_resp.readStruct("tResp", rsp_class) if result is not None and hasattr(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 # ---------- 业务便捷方法 ---------- def query_http_dns( self, uid: int, cookie: str, guid: str = "", ua: str = HTTP_HUYA_UA, domains: list[str] | None = None, timeout: float = 10.0, ) -> list[str]: """launch.queryHttpDns — 换取 WSS host 列表（失败返回空列表）。""" req = QueryHttpDnsReq() req.uid = int(uid or 0) req.ua = ua req.domains = list(domains or [HTTP_DNS_DOMAIN]) resp = self.call_rpc( "launch", "queryHttpDns", req, None, uid=uid, guid=guid, cookie=cookie, timeout=timeout, request_id=-1, origin="https://m-shop.huya.com", referer="https://m-shop.huya.com/points.html", ) if resp is None: return [] newdata = getattr(resp, "newdata", None) or {} payload = newdata.get("tRsp") or newdata.get("tResp") or b"" return _extract_ws_hosts(payload) def get_act_info( self, act_id: int, act_uuid: str = "", app_source: str = "", timeout: float = 15.0, ): """读取活动主信息，用于驱动当前活动配置。""" from .activity_structs import GetActInfoReq, GetActInfoResp req = GetActInfoReq() req.actId = int(act_id or 0) req.actUuid = act_uuid req.appSource = app_source return self.call_rpc( "webActUI", "getActInfo", req, GetActInfoResp, timeout=timeout ) def get_act_user_task_detail( self, uid: int, cookie: str, act_id: int, timeout: float = 15.0 ): """读取当前用户任务状态，而不是只读取任务定义。""" from .activity_structs import GetActUserTaskDetailReq, GetActUserTaskDetailResp req = GetActUserTaskDetailReq() req.userId = self._build_activity_user(uid, cookie) req.actId = int(act_id or 0) return self.call_rpc( "webActUI", "getActUserTaskDetail", req, GetActUserTaskDetailResp, uid=uid, cookie=cookie, timeout=timeout, ) 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 return self.call_rpc( "webActUI", "getUserScore", req, GetUserScoreResp, uid=uid, cookie=cookie, timeout=timeout, ) 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 return self.call_rpc( "webActUI", "getActPrizeList", req, GetActPrizeListResp, uid=uid, cookie=cookie, timeout=timeout, ) 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 return self.call_rpc( "webActUI", "getUserPrizeRecords", req, GetUserPrizeRecordsResp, uid=uid, cookie=cookie, timeout=timeout, ) 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) req.pid = int(pid or 0) return self.call_rpc( "webActUI", "scoreExchangePrize", req, ScoreExchangePrizeResp, uid=uid, cookie=cookie, timeout=timeout, ) def get_act_prize_detail( self, uid: int, cookie: str, sid: int, pid: int, timeout: float = 15.0 ): """查询单个兑换奖品详情和服务端限制。""" from .activity_structs import GetActPrizeDetailReq, GetActPrizeDetailResp req = GetActPrizeDetailReq() req.userId = self._build_activity_user(uid, cookie) req.sid = int(sid or 0) req.pid = int(pid or 0) return self.call_rpc( "webActUI", "getActPrizeDetail", req, GetActPrizeDetailResp, uid=uid, cookie=cookie, timeout=timeout, ) 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) return self.call_rpc( "webActUI", "getActTaskDetail", req, GetActTaskDetailResp, uid=uid, cookie=cookie, timeout=timeout, ) def check_user_bind_game_account( self, uid: int, cookie: str, b_act_id: int, is_use_outer_act_id: int = 1, gid: int = 0, outer_act_id: str = "", scene: str = "", timeout: float = 15.0, ): """查询用户游戏账号绑定状态。""" from .activity_structs import ( CheckUserBindGameAccountReq, CheckUserBindGameAccountResp, ) req = CheckUserBindGameAccountReq() req.userId = self._build_activity_user(uid, cookie) req.gid = int(gid or 0) req.outerActId = str(outer_act_id or "") req.isInnerBind = 0 req.scene = scene or "" req.bActId = int(b_act_id or 0) req.isUseOuterActId = int(is_use_outer_act_id or 0) return self.call_rpc( "webActUI", "checkUserBindGameAccount", req, CheckUserBindGameAccountResp, uid=uid, cookie=cookie, timeout=timeout, ) def confirm_bind_act_account( self, uid: int, cookie: str, b_act_id: int, gid: int = 0, outer_act_id: str = "", timeout: float = 15.0, ): """确认虎牙号与当前活动的游戏账号绑定关系。""" from .activity_structs import ( ConfirmBindActAccountReq, ConfirmBindActAccountResp, ) req = ConfirmBindActAccountReq() req.userId = self._build_activity_user(uid, cookie) req.gid = int(gid or 0) req.outerActId = str(outer_act_id or "") req.bActId = int(b_act_id or 0) return self.call_rpc( "webActUI", "confirmBindActAccount", req, ConfirmBindActAccountResp, uid=uid, cookie=cookie, timeout=timeout, ) def get_live_link_param( self, uid: int, cookie: str, b_act_id: int, gid: int = 0, outer_act_id: int = 0, game_auth_scene: str = "", timeout: float = 15.0, ): """获取绑定二维码拉起参数。""" from .activity_structs import GetLiveLinkParamReq, GetLiveLinkParamResp req = GetLiveLinkParamReq() req.userId = self._build_activity_user(uid, cookie) req.gid = int(gid or 0) req.outerActId = int(outer_act_id or 0) req.gameAuthScene = game_auth_scene or "" req.v = "" req.bActId = int(b_act_id or 0) return self.call_rpc( "webActUI", "getLiveLinkParam", req, GetLiveLinkParamResp, uid=uid, cookie=cookie, timeout=timeout, ) 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)] if not req.uidList: return None return self.call_rpc( "huyauserui", "getUserProfileBatch", req, GetUserProfileBatchResp, uid=uid, cookie=cookie, timeout=timeout, ) @staticmethod def _livelink_qrcode_image_src(qrcode: str) -> str: """把 livelink 返回的二维码内容转成浏览器可直接展示的图片地址。""" text = (qrcode or "").strip() if not text: return "" lower = text.lower() if lower.startswith(("https://", "http://", "data:image/")): return text 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: """请求绑定页内部接口，获取真正可扫的小程序码图片。""" import gzip import time query = urllib.parse.urlsplit(bind_page_url or "").query if not query: self.logger("[LIVELINK] 获取小程序码失败: 绑定页 URL 缺少 query") return None params = urllib.parse.parse_qs(query, keep_blank_values=True) live_plat_id = (params.get("livePlatId") or [""])[0] act_id = (params.get("actId") or [""])[0] from_id = (params.get("fromId") or [""])[0] if len(from_id) > 32: from_id = "" suffix = "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=8)) session_id = f"{int(time.time())}_{suffix}" payload = json.dumps( {"queryString": f"{query}&sessionId={session_id}"}, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") endpoint = f"{LIVELINK_BIND_API_BASE}/api/qrcode/getMiniAppQrcode" 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": "", }, ) 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: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 self.logger(f"[LIVELINK] ❌ 小程序码请求失败: {type(e).__name__}: {e}") return None try: data = json.loads(resp_data.decode("utf-8", "replace")) except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 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 {} qrcode = jdata.get("qrcode") or "" image_src = self._livelink_qrcode_image_src(qrcode) if not image_src: self.logger("[LIVELINK] ❌ 小程序码接口未返回 qrcode") return None 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: """轮询 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: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 self.logger(f"[LIVELINK] ❌ 二维码状态请求失败: {type(e).__name__}: {e}") return None try: data = json.loads(resp_data.decode("utf-8", "replace")) except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 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, b_act_id: int, game_auth_scene: str = "", nick_name: str = "", face_url: str = "", redirect_url: str = "", ) -> dict: """按活动页 JS 逻辑拼绑定二维码 URL。""" from urllib.parse import quote, urlencode def join_query(params: dict) -> str: # livelink 返回的 code 已经带 %2F/%2B，不能再用 urlencode 二次转义。 return "&".join(f"{key}={value}" for key, value in params.items()) bind_query = { "t": livelink_param.get("t", ""), "gameIdList": livelink_param.get("gameIdList", ""), "actId": livelink_param.get("actId", ""), "livePlatId": livelink_param.get("livePlatId", ""), "sig": livelink_param.get("sig", ""), "code": livelink_param.get("code", ""), "faceUrl": quote(face_url or "", safe=""), "nickName": quote(nick_name or "", safe=""), } bind_query_text = join_query(bind_query) mini_path = quote(f"/pages/gameAccountBind/index?{bind_query_text}", safe="") action_url = ( "https://m.huya.com?hyaction=launchweixinminiprogram" f"&path={mini_path}&username=gh_49af166706ae" ) pc_query = dict(bind_query) pc_query["redirectUrl"] = quote(redirect_url or "", safe="") pc_url = f"https://livelink.qq.com/act/a20200527midgroupage/pc/?{join_query(pc_query)}" huya_params = { "bActId": int(b_act_id or 0), "isUseOuterActId": 1, "delayclose": 1000, "from": "diy", } if game_auth_scene: huya_params["gameAuthScene"] = game_auth_scene huya_base = "https://hd.huya.com/h5/livelinkV3/index.html" huya_container_url = f"{huya_base}?{urlencode(huya_params)}" exchange_params = dict(huya_params) exchange_params["extendParm"] = "type=qqCoinDeliver" return { "action_url": action_url, "qr_url": pc_url, "pc_url": pc_url, "huya_container_url": huya_container_url, "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", app_scene=SHOP_APP_SCENE, ): from .shop_structs import GetGoodsInfoReqV5, GoodsInfoRsp req = GetGoodsInfoReqV5() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app(source_id, app_scene) req.pid = pid req.spuId = spu_id req.skuId = sku_id req.gameId = game_id return self.call_rpc( "shopMiddleUI", "getGoodsInfoV5", req, GoodsInfoRsp, uid, guid, cookie ) def list_pay_channels( self, uid: int, guid: str, cookie: str, spu_id: str, sku_id: int, supplier_id: int = 87401, timeout: float = 15.0, ): from .shop_structs import ListPayChannelReq, ListPayChannelRsp req = ListPayChannelReq() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE) req.supplierId = int(supplier_id or 0) req.spuId = spu_id req.skuId = int(sku_id or 0) return self.call_rpc( "shopMiddleUI", "listPayChannelV5", req, ListPayChannelRsp, uid, guid, cookie, timeout=timeout, ) def check_hy_protocol( self, uid: int, guid: str, cookie: str, timeout: float = 15.0 ): from .shop_structs import CheckHyProtocolReq, ShopCodeRsp req = CheckHyProtocolReq() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE) return self.call_rpc( "shopMiddleUI", "checkHyProtocolV5", req, ShopCodeRsp, uid, guid, cookie, timeout=timeout, ) def check_user_buy_auth( self, uid: int, guid: str, cookie: str, spu_id: str, sku_id: int, timeout: float = 15.0, ): from .shop_structs import CheckUserBuyAuthReq, ShopCodeRsp req = CheckUserBuyAuthReq() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE) req.spuId = spu_id req.skuId = int(sku_id or 0) return self.call_rpc( "shopMiddleUI", "checkUserBuyAuth", req, ShopCodeRsp, uid, guid, cookie, timeout=timeout, ) 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 req = QueryUserOrderListReq() req.userId = self._build_shop_user(uid, cookie) req.offset = int(offset or 0) req.orderType = int(order_type or 1) req.pageSize = int(page_size or 10) req.status = int(status or 0) result = self.call_rpc( "shopMiddleUI", "queryUserOrderList", req, QueryUserOrderListRsp, uid, guid, cookie, timeout=timeout, ) if result is not None and getattr(result, "orders", None): return result fallback = self.call_rpc( "revenueWebUI", "queryUserOrderList", req, QueryUserOrderListRsp, uid, guid, cookie, timeout=timeout, ) return fallback if fallback is not None else result def order_detail( self, uid: int, guid: str, cookie: str, order_id: int, timeout: float = 15.0 ): """按订单号查询商城订单详情（最新商城支付链路）。""" from .shop_structs import OrderDetailReq, OrderDetailRsp req = OrderDetailReq() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE) req.orderId = int(order_id or 0) return self.call_rpc( "shopMiddleUI", "orderDetailV5", req, OrderDetailRsp, uid, guid, cookie, timeout=timeout, ) def create_order( self, uid, guid, cookie, pid, spu_id, sku_id, item_count=1, source_id="yellowcarlist", game_id="", app_scene=SHOP_APP_SCENE, order_scene=ORDER_SCENE, order_type=6, biz_params=JIFENCENTER_BIZ_PARAMS, ): """下单（HTTP 通道），参数与 2026-10-13 积分中心抓包一致。 旧实现会从 orderType=1 试到 10，失败重试有重复下单风险；抓包已确认 和平精英虚拟商品固定 orderType=6，这里改为单次调用。 """ from .shop_structs import ( CreateOrderAccountParam, CreateOrderExtraParam, CreateOrderPromotionParam, CreateOrderReqV5, CreateOrderRsp, ) req = CreateOrderReqV5() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app(source_id, app_scene) req.receiveId = 0 req.pid = pid req.spuId = spu_id req.skuId = sku_id req.itemCount = item_count req.gameId = game_id or "0" req.src = order_scene req.scene = 0 req.sourceId = source_id req.orderType = order_type extra = CreateOrderExtraParam() extra.bizParams = biz_params req.extraParam = extra req.env = self._build_order_env() req.orderScene = order_scene req.promotionParam = CreateOrderPromotionParam() req.accountParam = CreateOrderAccountParam() req.bizType = ORDER_BIZ_TYPE req.gameCategoryId = GAME_CATEGORY_ID_HUYA req.ext = "" result: Any = self.call_rpc( "shopMiddleUI", "createOrderV5", req, CreateOrderRsp, uid, guid, cookie ) if result is not None: self.logger( f" createOrderV5: code={result.code} orderId={result.orderId} msg={result.message}" ) else: self.logger(" createOrderV5: 无响应") return result @staticmethod def _normalize_pay_channel(pay_type) -> str: text = str(pay_type or "").strip() lowered = text.lower() if lowered in {"", "1", "zfb", "alipay", "支付宝"}: return "Zfb" if lowered in {"2", "wx", "weixin", "wechat", "微信"}: return "Weixin" return text def pay_order_submit( self, uid, guid, cookie, order_id, pay_type=1, pid=0, source_id="yellowcarlist", app_scene=SHOP_APP_SCENE, item_count=1, ): """发起支付（HTTP 通道），字段与 2026-10-13 积分中心抓包一致。""" from .shop_structs import PayOrderRes os = TafOutputStream() os.write_struct_begin(0) os.write_struct(0, self._build_shop_user(uid, cookie)) os.write_struct(1, self._build_shop_app(source_id, app_scene)) os.write_int64(2, order_id) os.write_string(3, self._normalize_pay_channel(pay_type)) os.write_string(4, "QrCode") callback_url = ( f"{POINTS_PAY_CALLBACK}?orderId={order_id}&pid={pid}&sourceId={source_id}" ) os.write_string(5, callback_url) os.write_map(6, self._build_order_env()) os.write_string(7, "null") os.write_string(8, "") os.write_struct_end() # 构造 Wup wup = WupRequest() wup.setServant("shopMiddleUI") wup.setFunc("payOrderSubmitV5") wup.setRequestId(1) wup.newdata["tReq"] = os.get_bytes() wup_data = wup.encode() baseinfo = self._generate_rpc_baseinfo(uid, guid, cookie) url = f"https://{CDNWS_HOST}/?baseinfo={baseinfo}" self.logger("[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", }, ) try: with urllib.request.urlopen(req, timeout=15) as resp: resp_data = resp.read() if resp_data[:2] == b"\x1f\x8b": import gzip resp_data = gzip.decompress(resp_data) except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 self.logger(f"[HTTP] ❌ payOrderSubmitV5 失败: {e}") return None 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: self.logger(f"[HTTP] 响应 keys: {list(wup_resp.newdata.keys())}") for k, v in wup_resp.newdata.items(): self.logger(f"[HTTP] {k} hex({len(v)}): {v.hex()[:200]}") result = wup_resp.readStruct("tRsp", PayOrderRes) if result is None: result = wup_resp.readStruct("tResp", PayOrderRes) return result # ============================================================ # 积分中心链路新增 RPC（HTTP 通道版本，参数与 WSS 侧一致） # ============================================================ @staticmethod def _raw_from_wup(wup_resp: WupResponse): """已解码的 WUP 响应 → ShopRawRsp（未逐字段建模的接口用）。""" from .shop_structs import ShopRawRsp data = wup_resp.newdata.get("tRsp") or wup_resp.newdata.get("tResp") payload = decode_struct_bytes(data) if isinstance(data, bytes) else {} return ShopRawRsp.from_payload(payload) def get_supplier_info( self, uid: int, guid: str, cookie: str, supplier_id: int = 87401, timeout: float = 15.0, ): """shopMiddleUI.getSupplierInfoV5（供应商/小店信息）。""" from .shop_structs import GetSupplierInfoReqV5 req = GetSupplierInfoReqV5() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE) req.supplierId = int(supplier_id or 0) resp = self.call_rpc( "shopMiddleUI", "getSupplierInfoV5", req, None, uid, guid, cookie, timeout=timeout, ) return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None def accept_hy_protocol( self, uid: int, guid: str, cookie: str, timeout: float = 15.0 ): """shopMiddleUI.acceptHyProtocolV5（同意商城协议）。""" from .shop_structs import AcceptHyProtocolReq, ShopCodeRsp req = AcceptHyProtocolReq() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE) return self.call_rpc( "shopMiddleUI", "acceptHyProtocolV5", req, ShopCodeRsp, uid, guid, cookie, timeout=timeout, ) def calc_order_promotion( self, uid: int, guid: str, cookie: str, pid: int, spu_id: str, sku_id: int = 0, item_count: int = 1, game_id: int = 9815, timeout: float = 15.0, ): """shopMiddleUI.calcOrderPromotion（下单前试算）。""" from .shop_structs import CalcOrderPromotionReqV5, shop_spu_number spu_num = shop_spu_number(spu_id) req = CalcOrderPromotionReqV5() req.userId = self._build_shop_user(uid, cookie) req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE) req.spuNum = spu_num req.skuId = int(sku_id or 0) req.itemCount = int(item_count or 1) req.pid = int(pid or 0) req.spuNum2 = spu_num req.gameId = int(game_id or 0) resp = self.call_rpc( "shopMiddleUI", "calcOrderPromotion", req, None, uid, guid, cookie, timeout=timeout, ) return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None def play_mall_query_allocation( self, uid: int, guid: str, cookie: str, source_id: str = "yellowcarlist", page_code: str = "heping_baodian_credit_center_page", timeout: float = 15.0, ): """PlayMallUI.queryAllocation（积分中心配额查询）。""" from .shop_structs import PlayMallQueryAllocationReq req = PlayMallQueryAllocationReq() req.userId = self._build_shop_user(uid, cookie) req.pageCode = page_code req.appInfo.sourceId = source_id resp = self.call_rpc( "PlayMallUI", "queryAllocation", req, None, uid, guid, cookie, timeout=timeout, ) return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None def play_mall_get_my_promotion( self, uid: int, guid: str, cookie: str, spu_id: str, sku_id: int = 0, price: int = 0, supplier_id: int = 87401, game_id: int = 9815, source_id: str = "yellowcarlist", timeout: float = 15.0, ): """PlayMallUI.getMyPromotion（返利/促销查询）。""" from .shop_structs import ( PlayMallGetMyPromotionReq, PromotionGoodsKey, shop_spu_number, ) spu_num = shop_spu_number(spu_id) req = PlayMallGetMyPromotionReq() req.userId = self._build_shop_user(uid, cookie) req.appInfo.sourceId = source_id goods = PromotionGoodsKey() goods.supplierId = int(supplier_id or 0) goods.spuNum = spu_num goods.skuId = int(sku_id or 0) goods.price = int(price or 0) goods.gameId = int(game_id or 0) goods.spuNum2 = spu_num req.goods = goods resp = self.call_rpc( "PlayMallUI", "getMyPromotion", req, None, uid, guid, cookie, timeout=timeout, ) return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None def get_user_profile( self, uid: int, guid: str, cookie: str, target_uid: int = 0, timeout: float = 15.0, ): """huyauserui.getUserProfile（单个用户资料，资料在 tag0.tag0）。""" from .shop_structs import GetUserProfileReq req = GetUserProfileReq() req.userId = self._build_shop_user(uid, cookie) req.targetUid = int(target_uid or uid or 0) resp = self.call_rpc( "huyauserui", "getUserProfile", req, None, uid, guid, cookie, timeout=timeout, ) return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None def get_follow_wechat_status( self, uid: int, guid: str, cookie: str, timeout: float = 15.0 ): """wupui.getUserFollowWeChatOfficialStatus（是否关注公众号）。""" from .shop_structs import FollowWeChatStatusReq req = FollowWeChatStatusReq() req.userId = self._build_shop_user(uid, cookie) resp = self.call_rpc( "wupui", "getUserFollowWeChatOfficialStatus", req, None, uid, guid, cookie, timeout=timeout, ) return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None def _self_check(): uid = 1199647239697 guid = "0a8973013ad3476a53029ddb9edb071f" cookie = f"yyuid=1199647239697; udb_passport=test; guid={guid}" trace_id = "0123456789abcdef:0123456789abcdef:0:0" 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.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.sGuid == "" assert user.sCookie == parsed.sCookie empty_user = HuyaHttpClient._build_activity_user(0, "") assert empty_user.sCookie == "" # ============================================================ # launch.queryHttpDns：动态解析 WSS host # # 2026-10-13「精英宝典」抓包确认：m-shop.huya.com 页面先用 HTTP cdnws 调 # launch.queryHttpDns（iRequestId 固定 -1）换取 host 列表，再连 WSS。 # 因此 host 不应长期写死；解析失败时回退到抓包实测 host。 # ============================================================ HTTP_DNS_DOMAIN = "cdnws.api.huya.com" _WSS_HOST_RE = re.compile(rb"[a-z0-9]{6,}-ws\.va\.huya\.com") #: 解析成功后的缓存时长（响应实测 ttl 1296000s，这里保守缓存）。 DNS_CACHE_TTL = 600.0 #: 解析失败后的冷却时间，避免每个任务都打一次无用的 DNS 请求。 DNS_FAILURE_COOLDOWN = 30.0 _dns_cache: dict[str, tuple[float, list[str]]] = {} class QueryHttpDnsReq(TafStruct): """launch.queryHttpDns 请求（2026-10-13 抓包逐字段）。 tag0 uid, tag1 UA, tag2 域名列表, tag3 appSrc, tag4 "", tag5 0, tag6 1 """ def __init__(self): self.uid: int = 0 # tag 0 self.ua: str = "" # tag 1 self.domains: list[str] = [] # tag 2 self.appSrc: str = "HUYA&ZH&2052" # tag 3 self.blank4: str = "" # tag 4 self.flag5: int = 0 # tag 5 self.flag6: int = 1 # tag 6 def write_to(self, os: TafOutputStream): os.write_int64(0, self.uid) os.write_string(1, self.ua) os.write_list(2, self.domains) os.write_string(3, self.appSrc) os.write_string(4, self.blank4) os.write_int32(5, self.flag5) os.write_int32(6, self.flag6) def read_from(self, ins: TafInputStream): pass def _extract_ws_hosts(payload: bytes) -> list[str]: """从 queryHttpDns 响应里按出现顺序取出 WSS host（去重）。""" hosts: list[str] = [] for match in _WSS_HOST_RE.findall(bytes(payload or b"")): host = match.decode("ascii", errors="ignore") if host and host not in hosts: hosts.append(host) return hosts def resolve_wss_host( uid: int, cookie: str, guid: str = "", logger: Callable[[str], None] | None = None, ) -> str: """解析积分中心通道的 WSS host：queryHttpDns + TTL 缓存 + 抓包 host 兜底。""" from .wss_client import SHOP_POINTS_WS_HOST log = logger or (lambda _message: None) now = time.monotonic() cached = _dns_cache.get(HTTP_DNS_DOMAIN) if cached and now < cached[0]: hosts = cached[1] else: hosts = [] try: hosts = HuyaHttpClient(logger=log).query_http_dns(uid, cookie, guid=guid) except Exception: # noqa: BLE001 外部接口失败一律回退默认 host hosts = [] ttl = DNS_CACHE_TTL if hosts else DNS_FAILURE_COOLDOWN _dns_cache[HTTP_DNS_DOMAIN] = (now + ttl, hosts) if not hosts: log("[WSS] queryHttpDns 无结果，使用抓包实测 host") return SHOP_POINTS_WS_HOST log(f"[WSS] queryHttpDns 解析 host={hosts[0]}（候选 {len(hosts)} 个）") return hosts[0] if __name__ == "__main__": _self_check()