571 lines
19 KiB
Python
571 lines
19 KiB
Python
"""虎牙活动积分相关 JCE 结构。"""
|
|
|
|
from .taf_protocol import TafInputStream, TafOutputStream, TafStruct, TafType
|
|
|
|
|
|
class ActivityUserId(TafStruct):
|
|
"""活动组件使用的用户标识。"""
|
|
|
|
def __init__(self):
|
|
self.lUid: int = 0
|
|
self.sGuid: str = ""
|
|
self.sToken: str = ""
|
|
self.sHuYaUA: str = "webh5&0.0.1&websocket&&diypc_52775"
|
|
self.sCookie: str = ""
|
|
self.iTokenType: int = 0
|
|
self.sDeviceInfo: str = ""
|
|
|
|
def write_to(self, os: TafOutputStream):
|
|
os.write_int64(0, self.lUid)
|
|
os.write_string(1, self.sGuid)
|
|
os.write_string(2, self.sToken)
|
|
os.write_string(3, self.sHuYaUA)
|
|
os.write_string(4, self.sCookie)
|
|
os.write_int32(5, self.iTokenType)
|
|
os.write_string(6, self.sDeviceInfo)
|
|
|
|
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.sToken = ins.read_string(2, default=self.sToken)
|
|
self.sHuYaUA = ins.read_string(3, default=self.sHuYaUA)
|
|
self.sCookie = ins.read_string(4, default=self.sCookie)
|
|
self.iTokenType = ins.read_int32(5, default=self.iTokenType)
|
|
self.sDeviceInfo = ins.read_string(6, default=self.sDeviceInfo)
|
|
|
|
|
|
class GetUserScoreReq(TafStruct):
|
|
"""webActUI.getUserScore 请求。"""
|
|
|
|
def __init__(self):
|
|
self.userId = ActivityUserId()
|
|
self.sid: int = 0
|
|
|
|
def write_to(self, os: TafOutputStream):
|
|
os.write_struct(0, self.userId)
|
|
os.write_int32(1, self.sid)
|
|
|
|
def read_from(self, ins: TafInputStream):
|
|
self.userId = ins.read_struct(0, ActivityUserId) or self.userId
|
|
self.sid = ins.read_int32(1, default=self.sid)
|
|
|
|
|
|
class GetUserScoreResp(TafStruct):
|
|
"""webActUI.getUserScore 响应。"""
|
|
|
|
def __init__(self):
|
|
self.status: int = 0
|
|
self.msg: str = ""
|
|
self.gainScore: int = 0
|
|
self.usedScore: int = 0
|
|
self.newGainScore: int = 0
|
|
self.newUsedScore: int = 0
|
|
self.score: int = 0
|
|
self.isOpenExchangeTip: int = 0
|
|
|
|
@property
|
|
def available_score(self) -> int:
|
|
"""按现网页组件逻辑计算可用积分。"""
|
|
if self.newGainScore or self.newUsedScore:
|
|
return max(0, int(self.newGainScore) - int(self.newUsedScore))
|
|
return int(self.score)
|
|
|
|
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.gainScore = ins.read_int32(2, default=self.gainScore)
|
|
self.usedScore = ins.read_int32(3, default=self.usedScore)
|
|
self.newGainScore = ins.read_int64(4, default=self.newGainScore)
|
|
self.newUsedScore = ins.read_int64(5, default=self.newUsedScore)
|
|
self.score = ins.read_int64(6, default=self.score)
|
|
self.isOpenExchangeTip = ins.read_int32(7, default=self.isOpenExchangeTip)
|
|
|
|
def write_to(self, os: TafOutputStream):
|
|
os.write_int32(0, self.status)
|
|
os.write_string(1, self.msg)
|
|
os.write_int32(2, self.gainScore)
|
|
os.write_int32(3, self.usedScore)
|
|
os.write_int64(4, self.newGainScore)
|
|
os.write_int64(5, self.newUsedScore)
|
|
os.write_int64(6, self.score)
|
|
os.write_int32(7, self.isOpenExchangeTip)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"status": self.status,
|
|
"msg": self.msg,
|
|
"gain_score": self.gainScore,
|
|
"used_score": self.usedScore,
|
|
"new_gain_score": self.newGainScore,
|
|
"new_used_score": self.newUsedScore,
|
|
"score": self.score,
|
|
"available_score": self.available_score,
|
|
"is_open_exchange_tip": self.isOpenExchangeTip,
|
|
}
|
|
|
|
|
|
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 ConfirmBindActAccountReq(TafStruct):
|
|
"""webActUI.confirmBindActAccount 请求。"""
|
|
|
|
def __init__(self):
|
|
self.userId = ActivityUserId()
|
|
self.gid: int = 0
|
|
self.outerActId: 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_string(2, self.outerActId)
|
|
os.write_int32(3, 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_string(2, default=self.outerActId)
|
|
self.bActId = ins.read_int32(3, default=self.bActId)
|
|
|
|
|
|
class ConfirmBindActAccountResp(TafStruct):
|
|
"""webActUI.confirmBindActAccount 响应。"""
|
|
|
|
def __init__(self):
|
|
self.status: int = 0
|
|
self.msg: 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)
|
|
|
|
def write_to(self, os: TafOutputStream):
|
|
os.write_int32(0, self.status)
|
|
os.write_string(1, self.msg)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"status": self.status,
|
|
"msg": self.msg,
|
|
}
|
|
|
|
|
|
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
|
|
rsp.msg = "请求成功"
|
|
rsp.gainScore = 300
|
|
rsp.usedScore = 120
|
|
rsp.newGainScore = 300
|
|
rsp.newUsedScore = 120
|
|
rsp.score = 180
|
|
|
|
os = TafOutputStream()
|
|
os.write_struct(0, rsp)
|
|
ins = TafInputStream(os.get_bytes())
|
|
_, dtype = ins.read_head()
|
|
assert dtype == TafType.STRUCT_BEGIN
|
|
|
|
parsed = GetUserScoreResp()
|
|
parsed.read_from(ins)
|
|
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"
|
|
|
|
confirm_rsp = ConfirmBindActAccountResp()
|
|
confirm_rsp.status = 200
|
|
confirm_rsp.msg = "请求成功"
|
|
os = TafOutputStream()
|
|
os.write_struct(0, confirm_rsp)
|
|
ins = TafInputStream(os.get_bytes())
|
|
_, dtype = ins.read_head()
|
|
assert dtype == TafType.STRUCT_BEGIN
|
|
parsed_confirm = ConfirmBindActAccountResp()
|
|
parsed_confirm.read_from(ins)
|
|
assert parsed_confirm.status == 200
|
|
assert parsed_confirm.msg == "请求成功"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_self_check()
|