1639 lines
54 KiB
Python
1639 lines
54 KiB
Python
"""
|
||
虎牙 HTTP POST RPC 通道（cdnws.api.huya.com）
|
||
|
||
用于 WSS 业务通道被设备态 Cookie 静默拒收时的兜底:
|
||
URL: https://cdnws.api.huya.com/?baseinfo=<base64(WSConnectParaInfo)>
|
||
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,
|
||
POINTS_REDEEM_GOODS_NUM,
|
||
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<string,string>
|
||
"""
|
||
|
||
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=<redacted>")
|
||
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,
|
||
*,
|
||
max_depth: int = 5,
|
||
keep_bytes: bool = False,
|
||
):
|
||
"""已解码的 WUP 响应 → ShopRawRsp（未逐字段建模的接口用）。"""
|
||
from .shop_structs import ShopRawRsp
|
||
|
||
data = wup_resp.newdata.get("tRsp") or wup_resp.newdata.get("tResp")
|
||
payload = (
|
||
decode_struct_bytes(
|
||
data,
|
||
max_depth=max_depth,
|
||
keep_bytes=keep_bytes,
|
||
)
|
||
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,
|
||
*,
|
||
deep: bool = False,
|
||
):
|
||
"""PlayMallUI.queryAllocation（积分中心配额查询）。
|
||
|
||
``deep=True`` 时保留 ``tag2[].tag5`` 的原始字节并放宽解码深度，
|
||
供 :func:`core.huya.shop_structs.parse_points_center_gifts` 解析商品列表。
|
||
"""
|
||
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,
|
||
)
|
||
if not isinstance(resp, WupResponse):
|
||
return None
|
||
return self._raw_from_wup(
|
||
resp,
|
||
max_depth=40 if deep else 5,
|
||
keep_bytes=deep,
|
||
)
|
||
|
||
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 get_redeem_need_goods(
|
||
self,
|
||
uid: int,
|
||
guid: str,
|
||
cookie: str,
|
||
goods_num: int = POINTS_REDEEM_GOODS_NUM,
|
||
timeout: float = 15.0,
|
||
):
|
||
"""shopMiddleUI.getRedeemNeedGoods（当前积分 tag2 / 兑换门槛 tag3）。"""
|
||
from .shop_structs import GetRedeemNeedGoodsReq
|
||
|
||
req = GetRedeemNeedGoodsReq()
|
||
req.userId = self._build_shop_user(uid, cookie)
|
||
req.shopAppInfo = self._build_shop_app("yellowcarlist", SHOP_APP_SCENE)
|
||
req.goodsNum = int(goods_num or 0)
|
||
resp = self.call_rpc(
|
||
"shopMiddleUI",
|
||
"getRedeemNeedGoods",
|
||
req,
|
||
None,
|
||
uid,
|
||
guid,
|
||
cookie,
|
||
timeout=timeout,
|
||
)
|
||
return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None
|
||
|
||
def get_grow_level_user_info(
|
||
self, uid: int, guid: str, cookie: str, timeout: float = 15.0
|
||
):
|
||
"""GameBizUI.getGrowLevelUserInfo（成长值 tag1.tag0 / 等级 tag1.tag1）。"""
|
||
from .shop_structs import GetGrowLevelUserInfoReq
|
||
|
||
req = GetGrowLevelUserInfoReq()
|
||
req.userId = self._build_shop_user(uid, cookie)
|
||
resp = self.call_rpc(
|
||
"GameBizUI",
|
||
"getGrowLevelUserInfo",
|
||
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()
|