实现虎牙绑定小程序码展示
This commit is contained in:
yml2213
@@ -104,6 +104,369 @@ class GetUserScoreResp(TafStruct):
|
||||
}
|
||||
|
||||
|
||||
class GameAccount(TafStruct):
|
||||
"""游戏账号信息。"""
|
||||
|
||||
def __init__(self):
|
||||
self.openId: str = ""
|
||||
self.appId: str = ""
|
||||
self.type: str = ""
|
||||
self.faceUrl: str = ""
|
||||
self.nick: str = ""
|
||||
self.accessToken: str = ""
|
||||
self.extendInfo: str = ""
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.openId = ins.read_string(0, default=self.openId)
|
||||
self.appId = ins.read_string(1, default=self.appId)
|
||||
self.type = ins.read_string(2, default=self.type)
|
||||
self.faceUrl = ins.read_string(3, default=self.faceUrl)
|
||||
self.nick = ins.read_string(4, default=self.nick)
|
||||
self.accessToken = ins.read_string(5, default=self.accessToken)
|
||||
self.extendInfo = ins.read_string(6, default=self.extendInfo)
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_string(0, self.openId)
|
||||
os.write_string(1, self.appId)
|
||||
os.write_string(2, self.type)
|
||||
os.write_string(3, self.faceUrl)
|
||||
os.write_string(4, self.nick)
|
||||
os.write_string(5, self.accessToken)
|
||||
os.write_string(6, self.extendInfo)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"app_id": self.appId,
|
||||
"type": self.type,
|
||||
"face_url": self.faceUrl,
|
||||
"nick": self.nick,
|
||||
"extend_info": self.extendInfo,
|
||||
"has_open_id": bool(self.openId),
|
||||
"has_access_token": bool(self.accessToken),
|
||||
}
|
||||
|
||||
|
||||
class GameRole(TafStruct):
|
||||
"""游戏角色信息。"""
|
||||
|
||||
def __init__(self):
|
||||
self.roleId: str = ""
|
||||
self.roleName: str = ""
|
||||
self.areaName: str = ""
|
||||
self.platName: str = ""
|
||||
self.partitionName: str = ""
|
||||
self.gameSerial: str = ""
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.roleId = ins.read_string(0, default=self.roleId)
|
||||
self.roleName = ins.read_string(1, default=self.roleName)
|
||||
self.areaName = ins.read_string(2, default=self.areaName)
|
||||
self.platName = ins.read_string(3, default=self.platName)
|
||||
self.partitionName = ins.read_string(4, default=self.partitionName)
|
||||
self.gameSerial = ins.read_string(5, default=self.gameSerial)
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_string(0, self.roleId)
|
||||
os.write_string(1, self.roleName)
|
||||
os.write_string(2, self.areaName)
|
||||
os.write_string(3, self.platName)
|
||||
os.write_string(4, self.partitionName)
|
||||
os.write_string(5, self.gameSerial)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"role_id": self.roleId,
|
||||
"role_name": self.roleName,
|
||||
"area_name": self.areaName,
|
||||
"plat_name": self.platName,
|
||||
"partition_name": self.partitionName,
|
||||
"game_serial": self.gameSerial,
|
||||
}
|
||||
|
||||
|
||||
class GameAccountData(TafStruct):
|
||||
"""绑定账号状态。"""
|
||||
|
||||
def __init__(self):
|
||||
self.isBindAcount: int = 0
|
||||
self.isBindRole: int = 0
|
||||
self.isNeedActCheck: int = 0
|
||||
self.gameAccount = GameAccount()
|
||||
self.gameRole = GameRole()
|
||||
self.changBindTime: int = 0
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.isBindAcount = ins.read_int32(0, default=self.isBindAcount)
|
||||
self.isBindRole = ins.read_int32(1, default=self.isBindRole)
|
||||
self.isNeedActCheck = ins.read_int32(2, default=self.isNeedActCheck)
|
||||
self.gameAccount = ins.read_struct(3, GameAccount) or self.gameAccount
|
||||
self.gameRole = ins.read_struct(4, GameRole) or self.gameRole
|
||||
self.changBindTime = ins.read_int32(5, default=self.changBindTime)
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_int32(0, self.isBindAcount)
|
||||
os.write_int32(1, self.isBindRole)
|
||||
os.write_int32(2, self.isNeedActCheck)
|
||||
os.write_struct(3, self.gameAccount)
|
||||
os.write_struct(4, self.gameRole)
|
||||
os.write_int32(5, self.changBindTime)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"is_bind_account": self.isBindAcount,
|
||||
"is_bind_role": self.isBindRole,
|
||||
"is_need_act_check": self.isNeedActCheck,
|
||||
"game_account": self.gameAccount.to_dict(),
|
||||
"game_role": self.gameRole.to_dict(),
|
||||
"change_bind_time": self.changBindTime,
|
||||
}
|
||||
|
||||
|
||||
class CheckUserBindGameAccountReq(TafStruct):
|
||||
"""webActUI.checkUserBindGameAccount 请求。"""
|
||||
|
||||
def __init__(self):
|
||||
self.userId = ActivityUserId()
|
||||
self.gid: int = 0
|
||||
self.outerActId: str = ""
|
||||
self.isInnerBind: int = 0
|
||||
self.scene: str = ""
|
||||
self.bActId: int = 0
|
||||
self.isUseOuterActId: int = 0
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_struct(0, self.userId)
|
||||
os.write_int32(1, self.gid)
|
||||
os.write_string(2, self.outerActId)
|
||||
os.write_int32(3, self.isInnerBind)
|
||||
os.write_string(4, self.scene)
|
||||
os.write_int32(5, self.bActId)
|
||||
os.write_int32(6, self.isUseOuterActId)
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.userId = ins.read_struct(0, ActivityUserId) or self.userId
|
||||
self.gid = ins.read_int32(1, default=self.gid)
|
||||
self.outerActId = ins.read_string(2, default=self.outerActId)
|
||||
self.isInnerBind = ins.read_int32(3, default=self.isInnerBind)
|
||||
self.scene = ins.read_string(4, default=self.scene)
|
||||
self.bActId = ins.read_int32(5, default=self.bActId)
|
||||
self.isUseOuterActId = ins.read_int32(6, default=self.isUseOuterActId)
|
||||
|
||||
|
||||
class CheckUserBindGameAccountResp(TafStruct):
|
||||
"""webActUI.checkUserBindGameAccount 响应。"""
|
||||
|
||||
def __init__(self):
|
||||
self.status: int = 0
|
||||
self.msg: str = ""
|
||||
self.accountData = GameAccountData()
|
||||
self.gameName: str = ""
|
||||
self.gameAuthScene: str = ""
|
||||
self.changeBindDay: int = 0
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.status = ins.read_int32(0, default=self.status)
|
||||
self.msg = ins.read_string(1, default=self.msg)
|
||||
self.accountData = ins.read_struct(2, GameAccountData) or self.accountData
|
||||
self.gameName = ins.read_string(3, default=self.gameName)
|
||||
self.gameAuthScene = ins.read_string(4, default=self.gameAuthScene)
|
||||
self.changeBindDay = ins.read_int32(5, default=self.changeBindDay)
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_int32(0, self.status)
|
||||
os.write_string(1, self.msg)
|
||||
os.write_struct(2, self.accountData)
|
||||
os.write_string(3, self.gameName)
|
||||
os.write_string(4, self.gameAuthScene)
|
||||
os.write_int32(5, self.changeBindDay)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"status": self.status,
|
||||
"msg": self.msg,
|
||||
"account_data": self.accountData.to_dict(),
|
||||
"game_name": self.gameName,
|
||||
"game_auth_scene": self.gameAuthScene,
|
||||
"change_bind_day": self.changeBindDay,
|
||||
}
|
||||
|
||||
|
||||
class GetLiveLinkParamReq(TafStruct):
|
||||
"""webActUI.getLiveLinkParam 请求。"""
|
||||
|
||||
def __init__(self):
|
||||
self.userId = ActivityUserId()
|
||||
self.gid: int = 0
|
||||
self.outerActId: int = 0
|
||||
self.gameAuthScene: str = ""
|
||||
self.v: str = ""
|
||||
self.bActId: int = 0
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_struct(0, self.userId)
|
||||
os.write_int32(1, self.gid)
|
||||
os.write_int32(2, self.outerActId)
|
||||
os.write_string(3, self.gameAuthScene)
|
||||
os.write_string(4, self.v)
|
||||
os.write_int32(5, self.bActId)
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.userId = ins.read_struct(0, ActivityUserId) or self.userId
|
||||
self.gid = ins.read_int32(1, default=self.gid)
|
||||
self.outerActId = ins.read_int32(2, default=self.outerActId)
|
||||
self.gameAuthScene = ins.read_string(3, default=self.gameAuthScene)
|
||||
self.v = ins.read_string(4, default=self.v)
|
||||
self.bActId = ins.read_int32(5, default=self.bActId)
|
||||
|
||||
|
||||
class GetLiveLinkParamResp(TafStruct):
|
||||
"""webActUI.getLiveLinkParam 响应。"""
|
||||
|
||||
def __init__(self):
|
||||
self.status: int = 0
|
||||
self.msg: str = ""
|
||||
self.livelinkParam: dict[str, str] = {}
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.status = ins.read_int32(0, default=self.status)
|
||||
self.msg = ins.read_string(1, default=self.msg)
|
||||
self.livelinkParam = ins.read_map(2)
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_int32(0, self.status)
|
||||
os.write_string(1, self.msg)
|
||||
os.write_map(2, self.livelinkParam)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"status": self.status,
|
||||
"msg": self.msg,
|
||||
"livelink_param": dict(self.livelinkParam),
|
||||
}
|
||||
|
||||
def to_log_dict(self) -> dict:
|
||||
data = dict(self.livelinkParam)
|
||||
if data.get("code"):
|
||||
data["code"] = f"<redacted len={len(data['code'])}>"
|
||||
if data.get("sig"):
|
||||
data["sig"] = f"<redacted len={len(data['sig'])}>"
|
||||
return {
|
||||
"status": self.status,
|
||||
"msg": self.msg,
|
||||
"livelink_param": data,
|
||||
}
|
||||
|
||||
|
||||
class GetUserProfileBatchReq(TafStruct):
|
||||
"""huyauserui.getUserProfileBatch 请求。"""
|
||||
|
||||
def __init__(self):
|
||||
self.userId = ActivityUserId()
|
||||
self.uidList: list[int] = []
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_struct(0, self.userId)
|
||||
os.write_list(1, self.uidList)
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
self.userId = ins.read_struct(0, ActivityUserId) or self.userId
|
||||
self.uidList = ins.read_list(1)
|
||||
|
||||
|
||||
class HuyaUserProfile(TafStruct):
|
||||
"""虎牙用户资料，仅读取绑定链接需要的字段。"""
|
||||
|
||||
def __init__(self):
|
||||
self.uid: int = 0
|
||||
self.nick: str = ""
|
||||
self.avatar: str = ""
|
||||
self.passport: str = ""
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
while True:
|
||||
tag, dtype = ins.peek_head()
|
||||
if dtype == TafType.STRUCT_END:
|
||||
break
|
||||
ins.read_head()
|
||||
if tag == 0:
|
||||
self.uid = ins._read_int_value(dtype)
|
||||
elif tag == 1:
|
||||
self.nick = self._read_string_value(ins, dtype)
|
||||
elif tag == 2:
|
||||
self.avatar = self._read_string_value(ins, dtype)
|
||||
elif tag == 17:
|
||||
self.passport = self._read_string_value(ins, dtype)
|
||||
else:
|
||||
ins.skip_field(dtype)
|
||||
|
||||
@staticmethod
|
||||
def _read_string_value(ins: TafInputStream, dtype: int) -> str:
|
||||
if dtype == TafType.STRING1:
|
||||
length = int.from_bytes(ins.buf.read(1), "big")
|
||||
elif dtype == TafType.STRING4:
|
||||
length = int.from_bytes(ins.buf.read(4), "big")
|
||||
else:
|
||||
ins.skip_field(dtype)
|
||||
return ""
|
||||
return ins.buf.read(length).decode("utf-8", errors="replace")
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_int64(0, self.uid)
|
||||
os.write_string(1, self.nick)
|
||||
os.write_string(2, self.avatar)
|
||||
os.write_string(17, self.passport)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"uid": self.uid,
|
||||
"nick": self.nick,
|
||||
"avatar": self.avatar,
|
||||
"passport": self.passport,
|
||||
}
|
||||
|
||||
|
||||
class GetUserProfileBatchResp(TafStruct):
|
||||
"""huyauserui.getUserProfileBatch 响应。"""
|
||||
|
||||
def __init__(self):
|
||||
self.profiles: list[HuyaUserProfile] = []
|
||||
|
||||
def read_from(self, ins: TafInputStream):
|
||||
found = ins._find_tag(0, False)
|
||||
if not found:
|
||||
return
|
||||
if found[1] != TafType.LIST:
|
||||
ins.skip_field(found[1])
|
||||
return
|
||||
count = ins._read_int_len()
|
||||
profiles = []
|
||||
for _ in range(count):
|
||||
_, dtype = ins.read_head()
|
||||
if dtype != TafType.STRUCT_BEGIN:
|
||||
ins.skip_field(dtype)
|
||||
continue
|
||||
item_profile = None
|
||||
while True:
|
||||
tag, field_type = ins.peek_head()
|
||||
if field_type == TafType.STRUCT_END:
|
||||
ins.read_head()
|
||||
break
|
||||
if tag == 0 and field_type == TafType.STRUCT_BEGIN:
|
||||
item_profile = ins.read_struct(0, HuyaUserProfile)
|
||||
else:
|
||||
ins.read_head()
|
||||
ins.skip_field(field_type)
|
||||
if item_profile is not None:
|
||||
profiles.append(item_profile)
|
||||
self.profiles = profiles
|
||||
|
||||
def write_to(self, os: TafOutputStream):
|
||||
os.write_list(0, self.profiles)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {"profiles": [profile.to_dict() for profile in self.profiles]}
|
||||
|
||||
|
||||
def _self_check():
|
||||
rsp = GetUserScoreResp()
|
||||
rsp.status = 200
|
||||
@@ -125,6 +488,26 @@ def _self_check():
|
||||
assert parsed.status == 200
|
||||
assert parsed.available_score == 180
|
||||
|
||||
link_rsp = GetLiveLinkParamResp()
|
||||
link_rsp.status = 200
|
||||
link_rsp.livelinkParam = {
|
||||
"actId": "17096",
|
||||
"gameIdList": "cjm",
|
||||
"livePlatId": "huya",
|
||||
"code": "abc",
|
||||
"sig": "def",
|
||||
"t": "1783159287",
|
||||
}
|
||||
os = TafOutputStream()
|
||||
os.write_struct(0, link_rsp)
|
||||
ins = TafInputStream(os.get_bytes())
|
||||
_, dtype = ins.read_head()
|
||||
assert dtype == TafType.STRUCT_BEGIN
|
||||
parsed_link = GetLiveLinkParamResp()
|
||||
parsed_link.read_from(ins)
|
||||
assert parsed_link.status == 200
|
||||
assert parsed_link.livelinkParam["actId"] == "17096"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
_self_check()
|
||||
|
||||
+224
-1
@@ -18,6 +18,7 @@ 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")
|
||||
@@ -246,7 +247,8 @@ class HuyaHttpClient:
|
||||
if result is None:
|
||||
result = wup_resp.readStruct("tResp", rsp_class)
|
||||
if result is not None and hasattr(result, "to_dict"):
|
||||
decoded = json.dumps(result.to_dict(), ensure_ascii=False, separators=(",", ":"))
|
||||
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
|
||||
|
||||
@@ -268,6 +270,227 @@ class HuyaHttpClient:
|
||||
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 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
|
||||
|
||||
Reference in New Issue
Block a user