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