feat: protobuf ctoken generation

this commit introduces on-demand ctoken generation with a wrapper for more concise, protodec-like syntax (see producePlaylistContinuation)
2025-09-26 22:40:29 +02:00
parent 1c9174c888
commit 30850a7ce0
2 changed files with 87 additions and 1 deletions
--- a/ythdd_proto.py
+++ b/ythdd_proto.py
@@ -0,0 +1,85 @@
+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