feat: support PO Token context and video_id

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Changed
+
+- Support new PO Token context requested by yt-dlp.
+
 ## [0.2.0]
 
 ### Changed

examples/getpot_example.py

@@ -27,19 +27,23 @@ 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, context=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')
 
+        if context != 'gvs':
+            raise UnsupportedRequest('Only GVS PO Tokens are 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,6 +97,13 @@ 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:

yt_dlp_plugins/extractor/getpot.py

@@ -113,29 +113,51 @@ 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,
+            player_url=None,
+            context=None,
+            video_id=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 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 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,
+            player_url=None,
+            context=None,
+            video_id=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 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 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,55 @@ 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,
+            visitor_data=None,
+            data_sync_id=None,
+            player_url=None,
+            context=None,
+            video_id=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,
+            video_id=video_id,
             **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,
+            'video_id': video_id,
             **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')