feat(protos) Add SabrContextUpdate, TimelineContext (#5)

Appears to be used for SSAP

utils/mitmproxy_sabrdump.py

@@ -21,7 +21,10 @@
     unknown_fields,
     SelectableFormats,
     PrewarmConnection,
-    AllowedCachedFormats
+    AllowedCachedFormats,
+    SabrContextUpdate,
+    SabrContextSendingPolicy,
+    TimelineContext
 )
 from yt_dlp_plugins.extractor._ytse.ump import UMPPartType, UMPParser
 
@@ -119,6 +122,22 @@ def response(self, flow: http.HTTPFlow) -> None:
                         f.write(f'Allowed Cached Formats: {acf}\n')
                         write_unknown_fields(f, acf)
 
+                    elif part.part_type == UMPPartType.SABR_CONTEXT_UPDATE:
+                        scu = protobug.loads(part.data, SabrContextUpdate)
+                        f.write(f'Sabr Context Update: {scu}\n')
+                        write_unknown_fields(f, scu)
+
+                    elif part.part_type == UMPPartType.SABR_CONTEXT_SENDING_POLICY:
+                        scsp = protobug.loads(part.data, SabrContextSendingPolicy)
+                        f.write(f'Sabr Context Sending Policy: {scsp}\n')
+                        write_unknown_fields(f, scsp)
+
+                    elif part.part_type == UMPPartType.TIMELINE_CONTEXT:
+                        tc = protobug.loads(part.data, TimelineContext)
+                        f.write(f'Timeline Context: {tc}\n')
+                        write_unknown_fields(f, tc)
+
+
                     elif part.part_type == UMPPartType.MEDIA or part.part_type == UMPPartType.MEDIA_END:
                         f.write(f'Media Header Id: {part.data[0]}\n')
 

utils/read_sabr_response.py

@@ -23,7 +23,10 @@
     unknown_fields,
     SelectableFormats,
     PrewarmConnection,
-    AllowedCachedFormats
+    AllowedCachedFormats,
+    SabrContextUpdate,
+    SabrContextSendingPolicy,
+    TimelineContext
 )
 from yt_dlp_plugins.extractor._ytse.ump import UMPPartType, UMPParser
 
@@ -105,6 +108,21 @@ def print_sabr_parts(fp):
             print(f'Allowed Cached Formats: {acf}')
             write_unknown_fields(f, acf)
 
+        elif part.part_type == UMPPartType.SABR_CONTEXT_UPDATE:
+            scu = protobug.loads(part.data, SabrContextUpdate)
+            print(f'Sabr Context Update: {scu}')
+            write_unknown_fields(f, scu)
+
+        elif part.part_type == UMPPartType.SABR_CONTEXT_SENDING_POLICY:
+            scsp = protobug.loads(part.data, SabrContextSendingPolicy)
+            print(f'Sabr Context Sending Policy: {scsp}')
+            write_unknown_fields(f, scsp)
+
+        elif part.part_type == UMPPartType.TIMELINE_CONTEXT:
+            tc = protobug.loads(part.data, TimelineContext)
+            print(f'Timeline Context: {tc}')
+            write_unknown_fields(f, tc)
+
         elif part.part_type == UMPPartType.MEDIA or part.part_type == UMPPartType.MEDIA_END:
             print(f'Media Header Id: {part.data[0]}')
 

yt_dlp_plugins/extractor/_ytse/protos/__init__.py

@@ -21,6 +21,9 @@
 from ._selectable_formats import SelectableFormats
 from ._prewarm_connection import PrewarmConnection
 from ._allowed_cached_formats import AllowedCachedFormats
+from ._sabr_context_update import SabrContextUpdate
+from ._sabr_context_sending_policy import SabrContextSendingPolicy
+from ._timeline_context import TimelineContext
 
 
 def unknown_fields(obj: typing.Any, path=()) -> typing.Iterable[tuple[tuple[str, ...], dict[int, list]]]:

yt_dlp_plugins/extractor/_ytse/protos/_sabr_context_sending_policy.py

@@ -0,0 +1,17 @@
+import protobug
+
+
+@protobug.message
+class SabrContextSendingPolicy:
+    # These contain the SabrContextUpdate.type values
+    # They are used to alter if a SabrContextUpdate is sent or not?
+
+    # Start sending the SabrContextUpdates of this type
+    start_policy: list[protobug.Int32] = protobug.field(1, default_factory=list)
+
+    # Stop sending the SabrContextUpdates of this type
+    stop_policy: list[protobug.Int32] = protobug.field(2, default_factory=list)
+
+    # Stop and discard the SabrContextUpdates of this type
+    # When type of 3, something is cleared from video data (related to SSAP and TIMELINE_CONTEXT)
+    discard_policy: list[protobug.Int32] = protobug.field(3, default_factory=list)

yt_dlp_plugins/extractor/_ytse/protos/_sabr_context_update.py

@@ -0,0 +1,37 @@
+import typing
+import protobug
+
+
+# @protobug.message
+# class ExampleTypeTwoContextUpdate:
+#     # At least appears to match BufferedRange pb...
+#     buffered_ranges: list[BufferedRange] = protobug.field(1, default_factory=list)
+
+# May be provided in:
+# - mediaCommonConfig in Innertube config
+# - SABR_CONTEXT_UPDATE
+# - TIMELINE_CONTEXT
+
+@protobug.message
+class SabrContextUpdate:
+
+    class SabrContextScope(protobug.Enum):
+        SABR_CONTEXT_SCOPE_UNKNOWN = 0
+        SABR_CONTEXT_SCOPE_PLAYBACK = 1
+        SABR_CONTEXT_SCOPE_REQUEST = 2
+        SABR_CONTEXT_SCOPE_WATCH_ENDPOINT = 3
+        SABR_CONTEXT_SCOPE_CONTENT_ADS = 4
+
+    class SabrContextWritePolicy(protobug.Enum):
+        # Whether to override existing sabr context updates?
+        SABR_CONTEXT_WRITE_POLICY_UNSPECIFIED = 0
+        SABR_CONTEXT_WRITE_POLICY_OVERWRITE = 1
+        SABR_CONTEXT_WRITE_POLICY_KEEP_EXISTING = 2
+
+    type: typing.Optional[protobug.Int32] = protobug.field(1, default=None)  # seen = 2
+    scope: typing.Optional[SabrContextScope] = protobug.field(2, default=None) # seen = 2 (SABR_CONTEXT_SCOPE_REQUEST?)
+
+    # note: may be base64 encoded
+    value: typing.Optional[protobug.Bytes] = protobug.field(3, default=None)
+    send_by_default: typing.Optional[protobug.Bool] = protobug.field(4, default=None)  # seen = True
+    write_policy: typing.Optional[SabrContextWritePolicy] = protobug.field(5, default=None)

yt_dlp_plugins/extractor/_ytse/protos/_streamer_context.py

@@ -43,8 +43,10 @@ class ClientInfo:
     unknown_field_98: typing.Optional[protobug.String] = protobug.field(98, default=None)  # seen on android = "google"
     gl_device_info: typing.Optional[GLDeviceInfo] = protobug.field(102, default=None)
 
+
 @protobug.message
-class Fqa:
+class ContextUpdate:
+    # Type and Value from a SabrContextUpdate
     type: typing.Optional[protobug.Int32] = protobug.field(1, default=None)
     value: typing.Optional[protobug.Bytes] = protobug.field(2, default=None)
 
@@ -67,7 +69,9 @@ class StreamerContext:
     po_token: typing.Optional[protobug.Bytes] = protobug.field(2, default=None)
     playback_cookie: typing.Optional[protobug.Bytes] = protobug.field(3, default=None)
     gp: typing.Optional[protobug.Bytes] = protobug.field(4, default=None)
-    field5: list[Fqa] = protobug.field(5, default_factory=list)
-    field6: list[protobug.Int32] = protobug.field(6, default_factory=list)
+    # referred to as "stmctxt". Seems to be the SABR context updates to apply?
+    context_updates: list[ContextUpdate] = protobug.field(5, default_factory=list)
+    # referred to as "unsntctxt". Is the type in the SABR Context update that was not sent (e.g. sendByDefault is False, or excluded by a sending policy)
+    unsent_context_updates: list[protobug.Int32] = protobug.field(6, default_factory=list)
     field7: typing.Optional[protobug.String] = protobug.field(7, default=None)
     field8: typing.Optional[Gqa] = protobug.field(8, default=None)

yt_dlp_plugins/extractor/_ytse/protos/_timeline_context.py

@@ -0,0 +1,39 @@
+import typing
+import protobug
+from ._sabr_context_update import SabrContextUpdate
+
+
+@protobug.message
+class rkH:
+    video_id: typing.Optional[protobug.String] = protobug.field(1, default=None) # videoId
+    pst: typing.Optional[protobug.Int32] = protobug.field(2, default=None)  # time - playback start time?
+    lst: typing.Optional[protobug.Int32] = protobug.field(3, default=None) # lst
+    ld: typing.Optional[protobug.Int32] = protobug.field(4, default=None) # duration
+    ls: typing.Optional[protobug.Int32] = protobug.field(5, default=None) # scale for ld
+    ps: typing.Optional[protobug.Int32] = protobug.field(6, default=None) # scale for pst
+
+
+
+@protobug.message
+class i8V:
+    L4: typing.Optional[protobug.Int32] = protobug.field(1, default=None) # nonv
+
+
+@protobug.message
+class Clip:
+    clip_id: typing.Optional[protobug.String] = protobug.field(1, default=None)
+    T7: typing.Optional[rkH] = protobug.field(2, default=None)
+    h2: typing.Optional[i8V] = protobug.field(3, default=None)
+
+
+@protobug.message
+class Timeline:
+    clip: list[Clip] = protobug.field(1, default_factory=list)
+    version: typing.Optional[protobug.String] = protobug.field(2, default=None)
+
+
+# Looks like this may be used for SSAP to update the visible timeline + provide a SABR context override to switch to the ad?
+@protobug.message
+class TimelineContext:
+    timeline: typing.Optional[Timeline] = protobug.field(1, default=None)
+    sabr_context_update: typing.Optional[SabrContextUpdate] = protobug.field(2, default=None)