709 lines
26 KiB
Python
709 lines
26 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 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<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 = '%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=<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': '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 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": 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
|
||
|
||
@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", 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, self._normalize_pay_channel(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()
|