feat: support PO Token context and video_id (#11)

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Changed
+
+- Support new PO Token context requested by yt-dlp. By default, providers only support "gvs" PO Token context for backwards compatibility.
+
 ## [0.2.0]
 
 ### Changed

examples/getpot_example.py

@@ -16,6 +16,9 @@ class ExampleGetPOTProviderRH(GetPOTProvider):  # ⚠ The class name must end in
     # Supported Innertube clients, as defined in yt_dlp.extractor.youtube.INNERTUBE_CLIENTS
     _SUPPORTED_CLIENTS = ('web', 'web_embedded', 'web_music')
 
+    # Support PO Token contexts. "gvs" (Google Video Server) or "player". Default is "gvs".
+    # _SUPPORTED_CONTEXTS = ('gvs', 'player')
+
     # Optional: Define the version of the provider. Shown in debug output for debugging purposes.
     VERSION = '0.0.1'
 
@@ -27,19 +30,20 @@ class ExampleGetPOTProviderRH(GetPOTProvider):  # ⚠ The class name must end in
     # You can get the proxies for the request with `self._get_proxies(request)`
 
     # Optional
-    def _validate_get_pot(self, client: str, ydl: YoutubeDL, visitor_data=None, data_sync_id=None, player_url=None, **kwargs):
+    def _validate_get_pot(self, client: str, ydl: YoutubeDL, data_sync_id=None, **kwargs):
         # ℹ️ If you need to validate the request before making the request to the external source, do it here.
         # Raise yt_dlp.networking.exceptions.UnsupportedRequest if the request is not valid.
         if data_sync_id:
             raise UnsupportedRequest('Fetching PO Token for accounts is not supported')
 
     # ℹ️ Implement this method
-    def _get_pot(self, client: str, ydl: YoutubeDL, visitor_data=None, data_sync_id=None, player_url=None, **kwargs) -> str:
+    def _get_pot(self, client: str, ydl: YoutubeDL, visitor_data=None, data_sync_id=None, player_url=None, context=None, video_id=None, **kwargs) -> str:
         # You should use the ydl instance to make requests where possible,
         # as it will handle cookies and other networking settings passed to yt-dlp.
         response = ydl.urlopen(Request('https://example.com/get_pot', data=json.dumps({
             'client': client,
-            'visitor_data': visitor_data
+            'visitor_data': visitor_data,
+            'context': context,
         }).encode()))
 
         # If you need access to the YoutubeIE instance

tests/test_provider.py

@@ -75,11 +75,13 @@ class TestClient:
     def test_get_pot(self):
         with YoutubeDL() as ydl:
             ie = ydl.get_info_extractor('Youtube')
-            pot = json.loads(ie.fetch_po_token('web', visitor_data='visitor', data_sync_id='sync', extra_params='extra'))
+            pot = json.loads(ie.fetch_po_token('web', visitor_data='visitor', data_sync_id='sync', extra_params='extra', video_id='xyz', context='GVS'))
             assert pot['client'] == 'web'
             assert pot['visitor_data'] == 'visitor'
             assert pot['data_sync_id'] == 'sync'
             assert pot['extra_params'] == 'extra'
+            assert pot['context'] == 'gvs'
+            assert pot['video_id'] == 'xyz'
             assert pot['player_url'] is None
 
     def test_get_pot_unsupported_client(self):
@@ -95,14 +97,28 @@ def test_get_pot_request_error(self):
             assert pot is None
 
 
+    def test_default_context(self):
+        with YoutubeDL() as ydl:
+            ie = ydl.get_info_extractor('Youtube')
+            pot = json.loads(ie.fetch_po_token('web'))
+            assert pot['context'] == 'gvs'
+
+
 class TestProviderValidation:
     def test_validate_supported_clients(self):
         with YoutubeDL() as ydl, ExampleProviderRH(logger=FakeLogger()) as provider:
             provider.validate(Request('get-pot:', extensions={'getpot': {'client': 'web'}, 'ydl': ydl}))
 
-            with pytest.raises(UnsupportedRequest, match=r'^Client android is not supported$'):
+            with pytest.raises(UnsupportedRequest, match=r'^Client "android" is not supported. Supported clients: web$'):
                 provider.validate(Request('get-pot:', extensions={'getpot': {'client': 'android'}, 'ydl': ydl}))
 
+    def test_validate_supported_contexts(self):
+        with YoutubeDL() as ydl, ExampleProviderRH(logger=FakeLogger()) as provider:
+            provider.validate(Request('get-pot:', extensions={'getpot': {'client': 'web', 'context': 'gvs'}, 'ydl': ydl}))
+
+            with pytest.raises(UnsupportedRequest, match=r'^PO Token context "player" is not supported. Supported contexts: gvs$'):
+                provider.validate(Request('get-pot:', extensions={'getpot': {'client': 'web', 'context': 'player'}, 'ydl': ydl}))
+
     @pytest.mark.parametrize('extensions', [
         {'getpot': 'invalid'},
         {'getpot': {'visitor_data': 'xyz'}},

yt_dlp_plugins/extractor/getpot.py

@@ -58,6 +58,9 @@ class GetPOTProvider(RequestHandler, abc.ABC):
     # Supported Innertube clients, as defined in yt_dlp.extractor.youtube.INNERTUBE_CLIENTS
     _SUPPORTED_CLIENTS = ()
 
+    # Support PO Token contexts. "gvs" (Google Video Server) is the default for backwards compatibility.
+    _SUPPORTED_CONTEXTS = ['gvs']
+
     # Version of the provider. Shown in debug output for debugging purposes.
     VERSION = None
 
@@ -93,7 +96,11 @@ def _validate(self, request: Request):
         client = pot_request.pop('client')
 
         if client not in self._SUPPORTED_CLIENTS:
-            raise UnsupportedRequest(f'Client {client} is not supported')
+            raise UnsupportedRequest(f'Client "{client}" is not supported. Supported clients: {", ".join(self._SUPPORTED_CLIENTS)}')
+
+        context = pot_request.get('context')
+        if context and self._SUPPORTED_CONTEXTS and context not in self._SUPPORTED_CONTEXTS:
+            raise UnsupportedRequest(f'PO Token context "{context}" is not supported. Supported contexts: {", ".join(self._SUPPORTED_CONTEXTS)}')
 
         self._validate_get_pot(
             client=client,
@@ -113,29 +120,59 @@ def _send(self, request: Request):
         except NoSupportingHandlers as e:
             raise RequestError(cause=e) from e
 
-    def _validate_get_pot(self, client: str, ydl: YoutubeDL, visitor_data=None, data_sync_id=None, player_url=None,
-                          **kwargs):
+    def _validate_get_pot(
+            self,
+            client: str,
+            ydl: YoutubeDL,
+            visitor_data=None,
+            data_sync_id=None,
+            session_index=None,
+            player_url=None,
+            context=None,
+            video_id=None,
+            ytcfg=None,
+            **kwargs
+    ):
         """
         Validate and check the GetPOT request is supported.
         :param client: Innertube client, from yt_dlp.extractor.youtube.INNERTUBE_CLIENTS.
         :param ydl: YoutubeDL instance.
         :param visitor_data: Visitor Data.
         :param data_sync_id: Data Sync ID. Only provided if yt-dlp is running with an account.
+        :param session_index: Session Index.
         :param player_url: Player URL. Only provided if the client is BotGuard based (requires JS player).
+        :param context: PO Token context. "gvs" or "player".
+        :param video_id: Video ID.
+        :param ytcfg: The ytcfg yt-dlp will use for the client to make Innertube requests.
         :param kwargs: Additional arguments that may be passed in the future.
         :raises UnsupportedRequest: If the request is unsupported.
         """
 
     @abc.abstractmethod
-    def _get_pot(self, client: str, ydl: YoutubeDL, visitor_data=None, data_sync_id=None, player_url=None,
-                 **kwargs) -> str:
+    def _get_pot(
+            self,
+            client: str,
+            ydl: YoutubeDL,
+            visitor_data=None,
+            data_sync_id=None,
+            session_index=None,
+            player_url=None,
+            context=None,
+            video_id=None,
+            ytcfg=None,
+            **kwargs
+    ) -> str:
         """
         Get a PO Token
         :param client: Innertube client, from yt_dlp.extractor.youtube.INNERTUBE_CLIENTS.
         :param ydl: YoutubeDL instance.
         :param visitor_data: Visitor Data.
         :param data_sync_id: Data Sync ID. Only provided if yt-dlp is running with an account.
+        :param session_index: Session Index.
         :param player_url: Player URL. Only provided if the client is BotGuard based (requires JS player).
+        :param context: PO Token context. "gvs" or "player".
+        :param video_id: Video ID.
+        :param ytcfg: The ytcfg yt-dlp will use for the client to make Innertube requests.
         :param kwargs: Additional arguments that may be passed in the future.
         :returns: PO Token
         :raises RequestError: If the request fails.

yt_dlp_plugins/extractor/getpot_client.py

@@ -32,39 +32,43 @@ def set_downloader(self, downloader: YoutubeDL):
 
             downloader.write_debug(f'[GetPOT] PO Token Providers: {display_list}', only_once=True)
 
-    def _fetch_po_token(self, client, visitor_data=None, data_sync_id=None, player_url=None, **kwargs):
+    def _fetch_po_token(
+            self,
+            client,
+            context=None,
+            **kwargs
+        ):
         # use any existing implementation
         pot = super()._fetch_po_token(
             client=client,
-            visitor_data=visitor_data,
-            data_sync_id=data_sync_id,
-            player_url=player_url,
+            context=context,
             **kwargs
         )
 
         if pot:
             return pot
 
+        # default to gvs for compatibility with older yt-dlp versions
+        context = (context or 'gvs').lower()
+
         params = {
             'client': client,
-            'visitor_data': visitor_data,
-            'data_sync_id': data_sync_id,
-            'player_url': player_url,
+            'context': context,
             **kwargs
         }
 
         try:
-            self._downloader.write_debug(f'[GetPOT] Fetching PO Token for {client} client')
+            self._downloader.write_debug(f'[GetPOT] Fetching {context} PO Token for {client} client')
             pot_response = self._parse_json(
                 self._provider_rd.send(Request('get-pot:', extensions={'ydl': self._downloader, 'getpot': params})).read(),
                 video_id='GetPOT')
 
         except NoSupportingHandlers:
-            self._downloader.write_debug(f'[GetPOT] No provider available for {client} client')
+            self._downloader.write_debug(f'[GetPOT] No {context} PO Token provider available for {client} client')
             return
 
         except RequestError as e:
-            self._downloader.report_warning(f'[GetPOT] Failed to fetch PO Token for {client} client: {e}')
+            self._downloader.report_warning(f'[GetPOT] Failed to fetch {context} PO Token for {client} client: {e}')
             return
 
         pot = pot_response.get('po_token')