Skip to content

Instantly share code, notes, and snippets.

@Trung0246

Trung0246/nimoTVDecode.py

Created July 18, 2021 05:44
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save Trung0246/b32375c96958e7c83dc2c0303fc4f2b0 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save Trung0246/b32375c96958e7c83dc2c0303fc4f2b0 to your computer and use it in GitHub Desktop.
Download ZIP
Extract stream url from data blob
Raw
nimoTVDecode.py
from jce import types, JceStruct, JceField
# from jce.types import JceType
# from typing import Dict, Tuple
# from typing import Dict, Tuple, Optional, Type, Any
# import struct
# get the raw data blob from G_roomBaseInfo.mStreamPkg
# we use an example data blob grabbed from the site
raw_blob = "000002e110022c3c4c56074d656469615549661367657453747265616d496e666f4279526f6f6d7d000102b008000206001800010605696e7433321d0000010c060474527370180001061b485559412e47657453747265616d496e666f4279526f6f6d5273701d000102710a0a08000106307375313539393531373730393934397235613831613430346163663532666464316336373432616365363234323934371a0900010a060254581630737531353939353137373039393439723561383161343034616366353266646431633637343261636536323432393437261b687474703a2f2f74782e666c762e6e696d6f2e74762f6c6976652f36042e666c76461b687474703a2f2f74782e686c732e6e696d6f2e74762f6c6976652f56052e6d3375386005790c8c9600a6062e736c696365bcccd90ce064f00f64f01064f61100f81200020603666c761a0603666c76161b687474703a2f2f74782e666c762e6e696d6f2e74762f6c6976652f26042e666c76390c4001506466007c0b0603686c731a0603686c73161b687474703a2f2f74782e686c732e6e696d6f2e74762f6c6976652f26052e6d337538390c40015c663369643d7375313539393531373730393934397235613831613430346163663532666464316336373432616365363234323934377c0b0b1900020a0604333630701101f42c30ff4a0101e01101680b56086e656564776d3d31680c7601340b0a0604373230701c2c30ff4a0105001102d00b56086e656564776d3d30680c7601320b2610343434363830373135383136343738343c463977735365637265743d653339323437656634383638663035343264306332343239343135363664373926777354696d653d363066353062363753000fcc58f90c0d3063000001746aaf567d700181058f9006a101f4b101f4c101f4d600e61961707069643d38312674703d313632363538353537353639390b0b1607383437313935382607383437313935383260f3b9e70b8c980ca80c"
data_blob = bytes.fromhex(raw_blob[8:]) # remove the header?
class NimoTVTop(JceStruct):
iVersion: types.INT16 = JceField(jce_id=1, jce_type=types.INT16)
cPacketType: types.INT8 = JceField(jce_id=2, jce_type=types.INT8)
iMessageType: types.INT32 = JceField(jce_id=3, jce_type=types.INT32)
iRequestId: types.INT32 = JceField(jce_id=4, jce_type=types.INT32)
sServantName: types.STRING = JceField(jce_id=5, jce_type=types.STRING)
sFuncName: types.STRING = JceField(jce_id=6, jce_type=types.STRING)
sBuffer: types.BYTES = JceField(jce_id=7, jce_type=types.BYTES)
iTimeout: types.INT32 = JceField(jce_id=8, jce_type=types.INT32)
context: types.MAP = JceField(jce_id=9, jce_type=types.MAP)
status: types.MAP = JceField(jce_id=10, jce_type=types.MAP)
class NimoTVMap(JceStruct):
data: types.MAP = JceField(jce_id=0, jce_type=types.MAP)
# If anyone know how to use these JceStruct instead of using that hack I'm really appreciated
# class NimoTVStream(JceStruct):
# vMediaStreamInfo: types.LIST = JceField(jce_id=0, jce_type=types.LIST) # Vector
# vStreamGearInfo: types.LIST = JceField(jce_id=1, jce_type=types.LIST) # Vector
# sStreamGroupId: types.STRING = JceField(jce_id=2, jce_type=types.STRING)
# lLiveCompatibleFlag: types.INT64 = JceField(jce_id=3, jce_type=types.INT64)
# sAntiCode: types.STRING = JceField(jce_id=4, jce_type=types.STRING)
# lStreamId: types.INT64 = JceField(jce_id=5, jce_type=types.INT64)
# lUid: types.INT64 = JceField(jce_id=6, jce_type=types.INT64)
# iStreamProperty: types.INT32 = JceField(jce_id=7, jce_type=types.INT32)
# iBitRate: types.INT32 = JceField(jce_id=8, jce_type=types.INT32)
# iStreamType: types.INT32 = JceField(jce_id=9, jce_type=types.INT32)
# iPcDefaultBitRate: types.INT32 = JceField(jce_id=10, jce_type=types.INT32)
# iWebDefaultBitRate: types.INT32 = JceField(jce_id=11, jce_type=types.INT32)
# iMobileDefaultBitRate: types.INT32 = JceField(jce_id=12, jce_type=types.INT32)
# sStreamBusiProperty: types.STRING = JceField(jce_id=13, jce_type=types.STRING)
# sAdditionalParam: types.STRING = JceField(jce_id=14, jce_type=types.STRING)
# class NimoTVStreamInfo(JceStruct): # vMediaStreamInfo element
# sCdnName: types.STRING = JceField(jce_id=0, jce_type=types.STRING)
# sStreamName: types.STRING = JceField(jce_id=1, jce_type=types.STRING)
# sFlvUrl: types.STRING = JceField(jce_id=2, jce_type=types.STRING)
# sFlvSuffix: types.STRING = JceField(jce_id=3, jce_type=types.STRING)
# sHlsUrl: types.STRING = JceField(jce_id=4, jce_type=types.STRING)
# sHlsSuffix: types.STRING = JceField(jce_id=5, jce_type=types.STRING)
# iLineIndex: types.INT32 = JceField(jce_id=6, jce_type=types.INT32)
# vFlvIPList: types.LIST = JceField(jce_id=7, jce_type=types.LIST) # Vector
# iIsP2PSupport: types.INT32 = JceField(jce_id=8, jce_type=types.INT32)
# sP2pUrl: types.STRING = JceField(jce_id=9, jce_type=types.STRING)
# sP2pSuffix: types.STRING = JceField(jce_id=10, jce_type=types.STRING)
# lFreeFlag: types.INT64 = JceField(jce_id=11, jce_type=types.INT64)
# iIsHEVCSupport: types.INT32 = JceField(jce_id=12, jce_type=types.INT32)
# vP2pIPList: types.LIST = JceField(jce_id=13, jce_type=types.LIST) # Vector
# iPCPriorityRate: types.INT32 = JceField(jce_id=14, jce_type=types.INT32)
# iWebPriorityRate: types.INT32 = JceField(jce_id=15, jce_type=types.INT32)
# iMobilePriorityRate: types.INT32 = JceField(jce_id=16, jce_type=types.INT32)
# sAdditionalParam: types.STRING = JceField(jce_id=17, jce_type=types.STRING)
# mMediaStreamFormatInfo: types.MAP = JceField(jce_id=18, jce_type=types.MAP)
# class NimoTVStreamFormatInfo(JceStruct): # mMediaStreamFormatInfo element
# sFormatName: types.STRING = JceField(jce_id=0, jce_type=types.STRING)
# sUrl: types.STRING = JceField(jce_id=1, jce_type=types.STRING)
# sSuffix: types.STRING = JceField(jce_id=2, jce_type=types.STRING)
# vIPList: types.LIST = JceField(jce_id=3, jce_type=types.LIST) # Vector
# iIsSupport: types.INT32 = JceField(jce_id=4, jce_type=types.INT32)
# iRate: types.INT32 = JceField(jce_id=5, jce_type=types.INT32)
# sAdditionalParam: types.STRING = JceField(jce_id=6, jce_type=types.STRING)
# lFormatPropertyFlag: types.INT64 = JceField(jce_id=7, jce_type=types.INT64)
# class NimoTVResolution(JceStruct):
# iWidth: types.INT32 = JceField(jce_id=0, jce_type=types.INT32)
# iHeight: types.INT32 = JceField(jce_id=1, jce_type=types.INT32)
# class NimoTVGearInfo(JceStruct): # vStreamGearInfo element
# sDisplayName: types.STRING = JceField(jce_id=0, jce_type=types.STRING)
# iBitRate: types.INT32 = JceField(jce_id=1, jce_type=types.INT32)
# iCodecType: types.INT32 = JceField(jce_id=2, jce_type=types.INT32)
# iHEVCBitRate: types.INT32 = JceField(jce_id=3, jce_type=types.INT32)
# tResolution: NimoTVResolution = JceField(jce_id=4, jce_type=NimoTVResolution)
# sAdditionalParam: types.STRING = JceField(jce_id=5, jce_type=types.STRING)
# mSupportFormat: types.MAP = JceField(jce_id=6, jce_type=types.MAP) # No idea
# sBusiGearIndex: types.STRING = JceField(jce_id=7, jce_type=types.STRING)
id_stream = [
"vMediaStreamInfo",
"vStreamGearInfo",
"sStreamGroupId",
"lLiveCompatibleFlag",
"sAntiCode",
"lStreamId",
"lUid",
"iStreamProperty",
"iBitRate",
"iStreamType",
"iPcDefaultBitRate",
"iWebDefaultBitRate",
"iMobileDefaultBitRate",
"sStreamBusiProperty",
"sAdditionalParam"
]
id_stream_info = [
"sCdnName",
"sStreamName",
"sFlvUrl",
"sFlvSuffix",
"sHlsUrl",
"sHlsSuffix",
"iLineIndex",
"vFlvIPList",
"iIsP2PSupport",
"sP2pUrl",
"sP2pSuffix",
"lFreeFlag",
"iIsHEVCSupport",
"vP2pIPList",
"iPCPriorityRate",
"iWebPriorityRate",
"iMobilePriorityRate",
"sAdditionalParam",
"mMediaStreamFormatInfo"
]
id_stream_format = [
"sFormatName",
"sUrl",
"sSuffix",
"vIPList",
"iIsSupport",
"iRate",
"sAdditionalParam",
"lFormatPropertyFlag"
]
id_gear = [
"sDisplayName",
"iBitRate",
"iCodecType",
"iHEVCBitRate",
"tResolution",
"sAdditionalParam",
"mSupportFormat",
"sBusiGearIndex"
]
id_resolution = [
"iWidth",
"iHeight"
]
def replace_key (data_dict, key_dict, old_key):
try:
data_dict[key_dict[old_key]] = data_dict.pop(old_key)
except KeyError:
pass
class STREAMMAP(types.MAP):
__jce_type__ = (8,)
# The most extreme hack so far to replace number with proper key
# probably will break in the future unless someone can use those JceStruct properly
@classmethod
def validate(cls, v):
for id, data in v.items():
for i in range(len(id_stream)):
if (0 in v[id]) :
for j in range(len(v[id][0])):
for k in range(len(id_stream_info)):
if (18 in v[id][0][j]) :
for protocol in v[id][0][j][18]:
for l in range(len(id_stream_format)):
replace_key(v[id][0][j][18][protocol], id_stream_format, l)
replace_key(v[id][0][j], id_stream_info, k)
if (1 in v[id]):
for j in range(len(v[id][1])):
for k in range(len(id_stream_info)):
if (4 in v[id][1][j]) :
for l in range(len(v[id][1][j][4])):
replace_key(v[id][1][j][4], id_resolution, l)
replace_key(v[id][1][j], id_gear, k)
replace_key(v[id], id_stream, i)
return v
class NimoTVMediaMap(JceStruct):
mMediaStreamInfoPack: STREAMMAP = JceField(jce_id=0, jce_type=STREAMMAP)
class NimoTVInfo(JceStruct):
tStreamInfoPack: NimoTVMediaMap = JceField(jce_id=0, jce_type=NimoTVMediaMap)
sRoomId: types.STRING = JceField(jce_id=1, jce_type=types.STRING)
sRoomGroupId: types.STRING = JceField(jce_id=2, jce_type=types.STRING)
sServantName: types.INT64 = JceField(jce_id=3, jce_type=types.INT64)
class NimoTVRSP(JceStruct):
tRsp: NimoTVInfo = JceField(jce_id=0, jce_type=NimoTVInfo)
data_top = NimoTVTop.decode(data_blob)
data_map = NimoTVMap.decode(data_top.sBuffer)
data_rsp = NimoTVRSP.decode(data_map.data.get("tRsp").get("HUYA.GetStreamInfoByRoomRsp"))
pack = data_rsp.tRsp.tStreamInfoPack.mMediaStreamInfoPack
stream_names = list(pack.keys())
stream_url_data = pack[stream_names[0]].get("vMediaStreamInfo")[0]
print(stream_url_data.get("sHlsUrl") + stream_url_data.get("sStreamName") + stream_url_data.get("sHlsSuffix"))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment