added retry mechanism when blocked while using proxy

Author: jdepoix

youtube_transcript_api/_api.py

@@ -48,9 +48,9 @@ def __init__(
             http_client.cookies = _load_cookie_jar(cookie_path)
         if proxy_config is not None:
             http_client.proxies = proxy_config.to_requests_dict()
-            if proxy_config.prevent_keeping_connections_alive():
+            if proxy_config.prevent_keeping_connections_alive:
                 http_client.headers.update({"Connection": "close"})
-        self._fetcher = TranscriptListFetcher(http_client)
+        self._fetcher = TranscriptListFetcher(http_client, proxy_config=proxy_config)
 
     def fetch(
         self,

youtube_transcript_api/_transcripts.py

@@ -4,14 +4,15 @@
 from itertools import chain
 
 from html import unescape
-from typing import List, Dict, Iterator, Iterable, Pattern
+from typing import List, Dict, Iterator, Iterable, Pattern, Optional
 
 from defusedxml import ElementTree
 
 import re
 
 from requests import HTTPError, Session, Response
 
+from .proxies import ProxyConfig
 from ._errors import (
     VideoUnavailable,
     YouTubeRequestFailed,
@@ -339,16 +340,32 @@ def _get_language_description(self, transcript_strings: Iterable[str]) -> str:
 
 
 class TranscriptListFetcher:
-    def __init__(self, http_client: Session):
+    def __init__(self, http_client: Session, proxy_config: Optional[ProxyConfig]):
         self._http_client = http_client
+        self._proxy_config = proxy_config
 
     def fetch(self, video_id: str) -> TranscriptList:
         return TranscriptList.build(
             self._http_client,
             video_id,
-            self._extract_captions_json(self._fetch_video_html(video_id), video_id),
+            self._fetch_captions_json(video_id),
         )
 
+    def _fetch_captions_json(self, video_id: str, try_number: int = 0) -> Dict:
+        try:
+            return self._extract_captions_json(
+                self._fetch_video_html(video_id), video_id
+            )
+        except RequestBlocked as exception:
+            retries = (
+                0
+                if self._proxy_config is None
+                else self._proxy_config.retries_when_blocked
+            )
+            if try_number + 1 < retries:
+                return self._fetch_captions_json(video_id, try_number=try_number + 1)
+            raise exception
+
     def _extract_captions_json(self, html: str, video_id: str) -> Dict:
         splitted_html = html.split("var ytInitialPlayerResponse = ")
 

youtube_transcript_api/proxies.py

@@ -32,6 +32,7 @@ def to_requests_dict(self) -> RequestsProxyConfigDict:
         """
         pass
 
+    @property
     def prevent_keeping_connections_alive(self) -> bool:
         """
         If you are using rotating proxies, it can be useful to prevent the HTTP
@@ -40,6 +41,16 @@ def prevent_keeping_connections_alive(self) -> bool:
         """
         return False
 
+    @property
+    def retries_when_blocked(self) -> int:
+        """
+        Defines how many times we should retry if a request is blocked. When using
+        rotating residential proxies with a large IP pool it can make sense to retry a
+        couple of times when a blocked IP is encountered, since a retry will trigger
+        an IP rotation and the next IP might not be blocked.
+        """
+        return 0
+
 
 class GenericProxyConfig(ProxyConfig):
     """
@@ -83,8 +94,9 @@ class WebshareProxyConfig(GenericProxyConfig):
     most reliable way to work around being blocked by YouTube.
 
     If you don't have a Webshare account yet, you will have to create one
-    at https://www.webshare.io/?referral_code=w0xno53eb50g and purchase a residential
-    proxy package that suits your workload, to be able to use this proxy config.
+    at https://www.webshare.io/?referral_code=w0xno53eb50g and purchase a "Residential"
+    proxy package that suits your workload, to be able to use this proxy config (make
+    sure NOT to purchase "Proxy Server" or "Static Residential"!).
 
     Once you have created an account you only need the "Proxy Username" and
     "Proxy Password" that you can find in your Webshare settings
@@ -105,24 +117,33 @@ def __init__(
         self,
         proxy_username: str,
         proxy_password: str,
+        retries_when_blocked: int = 10,
         domain_name: str = DEFAULT_DOMAIN_NAME,
         proxy_port: int = DEFAULT_PORT,
     ):
         """
         Once you have created a Webshare account at
-        https://www.webshare.io/?referral_code=w0xno53eb50g and purchased a residential
-        proxy package, this config class allows you to easily use it, by defaulting to
-        the most reliable proxy settings (rotating residential proxies).
+        https://www.webshare.io/?referral_code=w0xno53eb50g and purchased a
+        "Residential" package (make sure NOT to purchase "Proxy Server" or
+        "Static Residential"!), this config class allows you to easily use it,
+        by defaulting to the most reliable proxy settings (rotating residential
+        proxies).
 
         :param proxy_username: "Proxy Username" found at
             https://dashboard.webshare.io/proxy/settings
         :param proxy_password: "Proxy Password" found at
             https://dashboard.webshare.io/proxy/settings
+        :param retries_when_blocked: Define how many times we should retry if a request
+            is blocked. When using rotating residential proxies with a large IP pool it
+            makes sense to retry a couple of times when a blocked IP is encountered,
+            since a retry will trigger an IP rotation and the next IP might not be
+            blocked. Defaults to 10.
         """
         self.proxy_username = proxy_username
         self.proxy_password = proxy_password
         self.domain_name = domain_name
         self.proxy_port = proxy_port
+        self._retries_when_blocked = retries_when_blocked
 
     @property
     def url(self) -> str:
@@ -139,5 +160,10 @@ def http_url(self) -> str:
     def https_url(self) -> str:
         return self.url
 
+    @property
     def prevent_keeping_connections_alive(self) -> bool:
         return True
+
+    @property
+    def retries_when_blocked(self) -> int:
+        return self._retries_when_blocked

youtube_transcript_api/test/test_api.py

@@ -341,6 +341,43 @@ def test_fetch__with_proxy_prevent_alive_connections(self, to_requests_dict):
         request = httpretty.last_request()
         self.assertEqual(request.headers.get("Connection"), "close")
 
+    @patch("youtube_transcript_api.proxies.GenericProxyConfig.to_requests_dict")
+    def test_fetch__with_proxy_retry_when_blocked(self, to_requests_dict):
+        for _ in range(3):
+            httpretty.register_uri(
+                httpretty.GET,
+                "https://www.youtube.com/watch",
+                body=load_asset("youtube_request_blocked.html.static"),
+            )
+        proxy_config = WebshareProxyConfig(
+            proxy_username="username",
+            proxy_password="password",
+        )
+
+        YouTubeTranscriptApi(proxy_config=proxy_config).fetch("Njp5uhTorCo")
+
+        self.assertEqual(len(httpretty.latest_requests()), 3 + 2)
+
+    @patch("youtube_transcript_api.proxies.GenericProxyConfig.to_requests_dict")
+    def test_fetch__with_proxy_reraise_when_blocked(self, to_requests_dict):
+        retries = 5
+        for _ in range(retries):
+            httpretty.register_uri(
+                httpretty.GET,
+                "https://www.youtube.com/watch",
+                body=load_asset("youtube_request_blocked.html.static"),
+            )
+        proxy_config = WebshareProxyConfig(
+            proxy_username="username",
+            proxy_password="password",
+            retries_when_blocked=retries,
+        )
+
+        with self.assertRaises(RequestBlocked):
+            YouTubeTranscriptApi(proxy_config=proxy_config).fetch("Njp5uhTorCo")
+
+        self.assertEqual(len(httpretty.latest_requests()), retries)
+
     def test_fetch__with_cookies(self):
         cookie_path = get_asset_path("example_cookies.txt")
         transcript = YouTubeTranscriptApi(cookie_path=cookie_path).fetch("GJLlxj_dtq8")