96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
from ythdd_globals import safeTraverse
|
|
import base64
|
|
import blackboxprotobuf as bbpb
|
|
import json
|
|
import urllib.parse
|
|
import ythdd_globals
|
|
|
|
def bbpbToB64(msg_and_typedef: tuple, urlsafe: bool = False, padding: bool = False) -> str:
|
|
encoded_protobuf = bbpb.encode_message(*msg_and_typedef)
|
|
if urlsafe:
|
|
b64_protobuf = base64.urlsafe_b64encode(encoded_protobuf)
|
|
else:
|
|
b64_protobuf = base64.b64encode(encoded_protobuf)
|
|
if padding:
|
|
url_encoded_b64 = urllib.parse.quote(b64_protobuf.decode())
|
|
else:
|
|
url_encoded_b64 = b64_protobuf.decode().rstrip('=')
|
|
return url_encoded_b64
|
|
|
|
def fdictToBbpb(msg: dict) -> tuple:
|
|
# Requires Python 3.7+ or CPython 3.6+,
|
|
# as these versions preserve dictionary insertion order.
|
|
# Structural matching (match, case) requires Python 3.10+.
|
|
clean_msg = {}
|
|
clean_type = {}
|
|
for key in msg:
|
|
num, type = key.split(":")
|
|
|
|
match type:
|
|
case "message":
|
|
# if the type is an embedded message
|
|
internal_msg, internal_type = fdictToBbpb(msg[key])
|
|
# msg can just be appended as usual
|
|
clean_msg[num] = internal_msg
|
|
# type contains more fields than normally
|
|
clean_type[num] = {
|
|
'field_order': list(internal_msg.keys()),
|
|
'message_typedef': internal_type,
|
|
'type': type
|
|
}
|
|
|
|
case "base64" | "base64u" | "base64p" | "base64up":
|
|
# if the type is a base64-embedded message
|
|
internal_msg, internal_type = fdictToBbpb(msg[key])
|
|
match type.removeprefix("base64"):
|
|
case "":
|
|
b64_encoded_msg = bbpbToB64((internal_msg, internal_type))
|
|
case "u":
|
|
b64_encoded_msg = bbpbToB64((internal_msg, internal_type), urlsafe=True)
|
|
case "p":
|
|
b64_encoded_msg = bbpbToB64((internal_msg, internal_type), padding=True)
|
|
case "up":
|
|
b64_encoded_msg = bbpbToB64((internal_msg, internal_type), urlsafe=True, padding=True)
|
|
clean_msg[num] = b64_encoded_msg
|
|
clean_type[num] = {'type': 'string'}
|
|
|
|
case "int" | "string":
|
|
clean_msg[num] = msg[key]
|
|
clean_type[num] = {'type': type}
|
|
|
|
case _:
|
|
raise KeyError(f'error(fmsgToBBPBTuple): invalid key "{type}"')
|
|
|
|
|
|
return (clean_msg, clean_type)
|
|
|
|
def producePlaylistContinuation(plid: str, offset: int = 0) -> str:
|
|
msge = {
|
|
'80226972:message': {
|
|
'2:string': f'VL{plid}',
|
|
'3:base64': {
|
|
'1:int': int(offset / 100),
|
|
'15:string': f'PT:{bbpbToB64(fdictToBbpb({"1:int": offset}))}',
|
|
'104:message': {
|
|
'1:int': 0
|
|
}
|
|
},
|
|
'35:string': plid
|
|
}
|
|
}
|
|
|
|
bbpb_dicts = fdictToBbpb(msge)
|
|
b64_ctoken = bbpbToB64(bbpb_dicts, urlsafe=True, padding=True)
|
|
|
|
return b64_ctoken
|
|
|
|
def produceSearchParams(page: int = 1) -> str:
|
|
msge = {
|
|
"9:int": 20 * (page - 1), # pagination
|
|
"30:int": 1 # no self-harm censorship
|
|
}
|
|
|
|
bbpb_dicts = fdictToBbpb(msge)
|
|
b64_params = bbpbToB64(bbpb_dicts, urlsafe=True, padding=True)
|
|
|
|
return b64_params |