""" 虎牙 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 json import random import struct import urllib.parse import urllib.request from typing import Optional, Callable from .taf_protocol import TafOutputStream, TafInputStream, TafType, TafStruct 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" 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 = '%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: """ 构造 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='') class HuyaHttpClient: """虎牙 HTTP POST RPC 客户端""" def __init__(self, logger: Callable[[str], None] = None): self.logger = logger or print @staticmethod def _normalize_cookie(cookie: str) -> str: """HTTP 业务 UserId.sCookie 需要带 huya_ua 前缀，避免重复写入。""" import re cookie = (cookie or "").strip() normalized = f"huya_ua={HTTP_HUYA_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) if not cookie: return normalized return f"{normalized}; {cookie}" @staticmethod def _build_user(uid: int, guid: str, cookie: str): """构造 HTTP 业务 UserId。""" from .shop_structs import UserId user = UserId() user.lUid = uid user.sGuid = guid user.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_app(source_id: str = "yellowcarlist", scene: int = 7): """构造 HTTP 业务 ShopAppInfo。""" 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(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":""}}]', "input_1": str(item_count), } @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 = cookie or "" 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): """ HTTP POST RPC 调用 Args: service: servant 名 (shopMiddleUI) method: 方法名 (getGoodsInfoV5) req_struct: 请求结构体 (TafStruct) rsp_class: 响应类 uid/guid/cookie: 用于构造 baseinfo，业务请求体也会按各接口需要携带 """ # 1. 构造 Wup 请求体 wup = WupRequest() wup.setServant(service) wup.setFunc(method) wup.setRequestId(1) wup.writeStruct("tReq", req_struct) wup_data = wup.encode() # [4字节长度][wup body] # 2. 构造 baseinfo baseinfo = generate_http_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': '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': 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()}") # 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"): log_data = result.to_log_dict() if hasattr(result, "to_log_dict") else result.to_dict() decoded = json.dumps(log_data, ensure_ascii=False, separators=(",", ":")) self.logger(f"[HTTP] 解码 tRsp: {decoded}") return result # ---------- 业务便捷方法 ---------- 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 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": bind_page_url, "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: self.logger(f"[LIVELINK] ❌ 小程序码请求失败: {type(e).__name__}: {e}") return None try: data = json.loads(resp_data.decode("utf-8", "replace")) except Exception as e: self.logger(f"[LIVELINK] ❌ 小程序码响应解析失败: {type(e).__name__}: {e}") return None if data.get("iRet") != 0: self.logger(f"[LIVELINK] ❌ 小程序码接口失败: iRet={data.get('iRet')} msg={data.get('sMsg') or '-'}") return None jdata = data.get("jData") if isinstance(data.get("jData"), dict) else {} 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 "", } @staticmethod def build_bind_urls( livelink_param: dict, b_act_id: int, game_auth_scene: str = "", nick_name: str = "", face_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"] = "" 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", 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) 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 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) # 如果指定了具体值，直接试 types_to_try = [order_type] if order_type is not None else list(range(1, 11)) last_result = None for ot in types_to_try: req = CreateOrderReqV5() req.userId = self._build_user(uid, guid, cookie) req.shopAppInfo = self._build_shop_app(source_id, 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 = scene req.scene = 0 req.sourceId = source_id req.orderType = ot req.extraParam = CreateOrderExtraParam() req.env = self._build_order_env(item_count) req.orderScene = scene req.promotionParam = CreateOrderPromotionParam() req.accountParam = CreateOrderAccountParam() req.bizType = 5 req.gameCategoryId = 507 req.ext = "" result = 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}") if result.code == 200 and result.orderId: self.logger(f"[✓] 找到有效 orderType={ot}") return result last_result = result else: self.logger(f" orderType={ot}: 无响应") return last_result 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) os.write_struct(0, self._build_user(uid, guid, cookie)) os.write_struct(1, self._build_shop_app(source_id, scene)) os.write_int64(2, order_id) os.write_string(3, "Zfb" if pay_type == 1 else str(pay_type)) os.write_string(4, "QrCode") callback_url = ( f"https://m-shop.yaoguo.com/index.html#/consumer/paycallback" f"?orderId={order_id}&pid={pid}&sourceId={source_id}" ) os.write_string(5, callback_url) os.write_map(6, self._build_order_env(item_count)) os.write_string(7, "null") 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 = generate_http_baseinfo(uid, guid, cookie) url = f"https://{CDNWS_HOST}/?baseinfo={baseinfo}" self.logger(f"[HTTP] POST shopMiddleUI.payOrderSubmitV5") self.logger(f"[HTTP] 发送 hex前60: {wup_data[:60].hex()}") 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: 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 def _self_check(): uid = 1199647239697 cookie = "yyuid=1199647239697; udb_passport=test" trace_id = "0123456789abcdef:0123456789abcdef:0:0" encoded = generate_http_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.sUA == HTTP_HUYA_UA assert parsed.sCookie == cookie assert parsed.sTraceId == trace_id user = HuyaHttpClient._build_activity_user(uid, cookie) assert user.sHuYaUA == HTTP_HUYA_UA assert user.sCookie == cookie if __name__ == "__main__": _self_check()