added exception for transcript URLs that require PO tokens

Author: jdepoix

youtube_transcript_api/__init__.py

@@ -25,4 +25,5 @@
     InvalidVideoId,
     AgeRestricted,
     YouTubeDataUnparsable,
+    PoTokenRequired,
 )

youtube_transcript_api/_api.py

@@ -1,5 +1,4 @@
 import warnings
-from pathlib import Path
 from typing import Optional, Iterable
 
 from requests import Session

youtube_transcript_api/_cli.py

@@ -33,7 +33,6 @@ def run(self) -> str:
                 proxy_password=parsed_args.webshare_proxy_password,
             )
 
-
         transcripts = []
         exceptions = []
 

youtube_transcript_api/_errors.py

@@ -16,12 +16,16 @@ class CookieError(YouTubeTranscriptApiException):
 
 
 class CookiePathInvalid(CookieError):
-    def __init__(self, cookie_path: Path):  # pragma: no cover until cookie authentication is re-implemented
+    def __init__(
+        self, cookie_path: Path
+    ):  # pragma: no cover until cookie authentication is re-implemented
         super().__init__(f"Can't load the provided cookie file: {cookie_path}")
 
 
 class CookieInvalid(CookieError):
-    def __init__(self, cookie_path: Path):  # pragma: no cover until cookie authentication is re-implemented
+    def __init__(
+        self, cookie_path: Path
+    ):  # pragma: no cover until cookie authentication is re-implemented
         super().__init__(
             f"The cookies provided are not valid (may have expired): {cookie_path}"
         )
@@ -258,3 +262,10 @@ def cause(self) -> str:
             requested_language_codes=self._requested_language_codes,
             transcript_data=str(self._transcript_data),
         )
+
+
+class PoTokenRequired(CouldNotRetrieveTranscript):
+    CAUSE_MESSAGE = (
+        "The requested video cannot be retrieved without a PO Token. If this happens, "
+        "please open a GitHub issue!"
+    )

youtube_transcript_api/_settings.py

@@ -1,7 +1,3 @@
 WATCH_URL = "https://www.youtube.com/watch?v={video_id}"
 INNERTUBE_API_URL = "https://www.youtube.com/youtubei/v1/player?key={api_key}"
-INNERTUBE_CONTEXT = {
-"client": {
-    "clientName": "ANDROID",
-    "clientVersion": "20.10.38"
-}}
+INNERTUBE_CONTEXT = {"client": {"clientName": "ANDROID", "clientVersion": "20.10.38"}}

youtube_transcript_api/_transcripts.py

@@ -27,6 +27,7 @@
     AgeRestricted,
     VideoUnplayable,
     YouTubeDataUnparsable,
+    PoTokenRequired,
 )
 
 
@@ -129,6 +130,8 @@ def fetch(self, preserve_formatting: bool = False) -> FetchedTranscript:
         Loads the actual transcript data.
         :param preserve_formatting: whether to keep select HTML text formatting
         """
+        if "&exp=xpe" in self._url:
+            raise PoTokenRequired(self.video_id)
         response = self._http_client.get(self._url)
         snippets = _TranscriptParser(preserve_formatting=preserve_formatting).parse(
             _raise_http_errors(response, self.video_id).text,
@@ -367,15 +370,14 @@ def _fetch_captions_json(self, video_id: str, try_number: int = 0) -> Dict:
                 return self._fetch_captions_json(video_id, try_number=try_number + 1)
             raise exception.with_proxy_config(self._proxy_config)
 
-
     def _extract_innertube_api_key(self, html: str, video_id: str) -> str:
         pattern = r'"INNERTUBE_API_KEY":\s*"([a-zA-Z0-9_-]+)"'
         match = re.search(pattern, html)
         if match and len(match.groups()) == 1:
             return match.group(1)
         if 'class="g-recaptcha"' in html:
             raise IpBlocked(video_id)
-        raise YouTubeDataUnparsable(video_id) # pragma: no cover
+        raise YouTubeDataUnparsable(video_id)  # pragma: no cover
 
     def _extract_captions_json(self, innertube_data: Dict, video_id: str) -> Dict:
         self._assert_playability(innertube_data.get("playabilityStatus"), video_id)

youtube_transcript_api/test/assets/youtube_po_token_required.innertube.json.static

@@ -26,6 +26,7 @@
     AgeRestricted,
     RequestBlocked,
     VideoUnplayable,
+    PoTokenRequired,
 )
 from youtube_transcript_api.proxies import GenericProxyConfig, WebshareProxyConfig
 
@@ -157,7 +158,9 @@ def test_list__url_as_video_id(self):
         )
 
         with self.assertRaises(InvalidVideoId):
-            YouTubeTranscriptApi().list("https://www.youtube.com/youtubei/v1/player?v=GJLlxj_dtq8")
+            YouTubeTranscriptApi().list(
+                "https://www.youtube.com/youtubei/v1/player?v=GJLlxj_dtq8"
+            )
 
     def test_translate_transcript(self):
         transcript = YouTubeTranscriptApi().list("GJLlxj_dtq8").find_transcript(["en"])
@@ -257,7 +260,6 @@ def test_fetch__exception_if_youtube_request_fails(self):
 
         self.assertIn("Request to YouTube failed: ", str(cm.exception))
 
-
     def test_get_transcript__exception_if_youtube_request_limit_reached(
         self,
     ):
@@ -290,6 +292,16 @@ def test_fetch__exception_if_ip_blocked(self):
         with self.assertRaises(IpBlocked):
             YouTubeTranscriptApi().fetch("abc")
 
+    def test_fetch__exception_if_po_token_required(self):
+        httpretty.register_uri(
+            httpretty.POST,
+            "https://www.youtube.com/youtubei/v1/player",
+            body=load_asset("youtube_po_token_required.innertube.json.static"),
+        )
+
+        with self.assertRaises(PoTokenRequired):
+            YouTubeTranscriptApi().fetch("GJLlxj_dtq8")
+
     def test_fetch__exception_request_blocked(self):
         httpretty.register_uri(
             httpretty.POST,
@@ -426,7 +438,7 @@ def test_fetch__with_generic_proxy_reraise_when_blocked(self, to_requests_dict):
 
     @pytest.mark.skip(
         reason="This test is temporarily disabled because cookie auth is currently not "
-               "working due to YouTube changes."
+        "working due to YouTube changes."
     )
     def test_fetch__with_cookies(self):
         cookie_path = get_asset_path("example_cookies.txt")
@@ -439,7 +451,7 @@ def test_fetch__with_cookies(self):
 
     @pytest.mark.skip(
         reason="This test is temporarily disabled because cookie auth is currently not "
-               "working due to YouTube changes."
+        "working due to YouTube changes."
     )
     def test_load_cookies(self):
         cookie_path = get_asset_path("example_cookies.txt")
@@ -454,7 +466,7 @@ def test_load_cookies(self):
 
     @pytest.mark.skip(
         reason="This test is temporarily disabled because cookie auth is currently not "
-               "working due to YouTube changes."
+        "working due to YouTube changes."
     )
     def test_load_cookies__bad_file_path(self):
         cookie_path = get_asset_path("nonexistent_cookies.txt")
@@ -463,7 +475,7 @@ def test_load_cookies__bad_file_path(self):
 
     @pytest.mark.skip(
         reason="This test is temporarily disabled because cookie auth is currently not "
-               "working due to YouTube changes."
+        "working due to YouTube changes."
     )
     def test_load_cookies__no_valid_cookies(self):
         cookie_path = get_asset_path("expired_example_cookies.txt")
@@ -743,7 +755,7 @@ def test_get_transcript__with_proxy_config__deprecated(self, to_requests_dict):
 
     @pytest.mark.skip(
         reason="This test is temporarily disabled because cookie auth is currently not "
-               "working due to YouTube changes."
+        "working due to YouTube changes."
     )
     def test_get_transcript__with_cookies__deprecated(self):
         cookies_path = get_asset_path("example_cookies.txt")
@@ -815,7 +827,7 @@ def test_get_transcripts__continue_on_error__deprecated(self, mock_get_transcrip
 
     @pytest.mark.skip(
         reason="This test is temporarily disabled because cookie auth is currently not "
-               "working due to YouTube changes."
+        "working due to YouTube changes."
     )
     @patch("youtube_transcript_api.YouTubeTranscriptApi.get_transcript")
     def test_get_transcripts__with_cookies__deprecated(self, mock_get_transcript):
@@ -832,6 +844,4 @@ def test_get_transcripts__with_proxies__deprecated(self, mock_get_transcript):
             "https": "http://localhost:8080",
         }
         YouTubeTranscriptApi.get_transcripts(["GJLlxj_dtq8"], proxies=proxies)
-        mock_get_transcript.assert_any_call(
-            "GJLlxj_dtq8", ("en",), proxies, False
-        )
+        mock_get_transcript.assert_any_call("GJLlxj_dtq8", ("en",), proxies, False)

youtube_transcript_api/test/test_api.py

@@ -329,7 +329,7 @@ def test_run__generic_proxy_config(self):
 
     @pytest.mark.skip(
         reason="This test is temporarily disabled because cookie auth is currently not "
-               "working due to YouTube changes."
+        "working due to YouTube changes."
     )
     def test_run__cookies(self):
         YouTubeTranscriptCli(