Migrate YouTube transcript clients to v1.x with proxy support

shared/libs/youtube_proxy.py

@@ -45,8 +45,18 @@ def _get_webshare_proxy_url() -> str | None:
     url = os.getenv("WEBSHARE_PROXY_URL", "").strip()
     if not url:
         return None
-    parsed = urllib.parse.urlparse(url)
-    if parsed.scheme not in ("http", "https", "socks5") or not parsed.hostname:
+    parsed: urllib.parse.ParseResult | None = None
+    try:
+        parsed = urllib.parse.urlparse(url)
+        hostname = parsed.hostname
+        parsed.port
+    except ValueError:
+        hostname = None
+    if (
+        parsed is None
+        or parsed.scheme not in ("http", "https", "socks5")
+        or not hostname
+    ):
         logger.warning(
             "WEBSHARE_PROXY_URL is set but malformed — falling back to direct connection"
         )
@@ -438,7 +448,9 @@ async def _transcript_operation():
                     logger.info(f"✅ Direct transcript extraction: {len(transcript)} segments")
                     return transcript
             except Exception as e:
-                logger.debug(f"Direct transcript failed: {e}")
+                logger.debug(
+                    f"Direct transcript failed: {_redact_proxy_credentials(e)}"
+                )
 
             # Method 2: Alternative language codes
             # ``list_transcripts`` class method is now the instance ``list``;
@@ -464,7 +476,9 @@ async def _transcript_operation():
                         logger.info(f"✅ Alternative language transcript: {len(transcript)} segments")
                         return transcript
             except Exception as e:
-                logger.debug(f"Alternative transcript failed: {e}")
+                logger.debug(
+                    f"Alternative transcript failed: {_redact_proxy_credentials(e)}"
+                )
 
             # Method 3: yt-dlp fallback
             try:
@@ -489,7 +503,9 @@ async def _transcript_operation():
                         # Convert to transcript format
                         return [{'text': 'Transcript extracted via yt-dlp', 'start': 0, 'duration': 1}]
             except Exception as e:
-                logger.debug(f"yt-dlp extraction failed: {e}")
+                logger.debug(
+                    f"yt-dlp extraction failed: {_redact_proxy_credentials(e)}"
+                )
 
             # CouldNotRetrieveTranscript(>=1.0) takes a bare video_id and builds
             # its own message/URL; passing a sentence corrupts the generated URL.

src/agents/interactive_metadata_extractor.py

@@ -16,6 +16,8 @@
 from dotenv import load_dotenv
 from youtube_transcript_api import YouTubeTranscriptApi
 
+from youtube_extension.utils.proxy import get_transcript_proxy_config
+
 load_dotenv()
 logger = logging.getLogger(__name__)
 
@@ -84,7 +86,10 @@ async def extract_transcript(self, video_id: str) -> list[dict[str, Any]]:
             # event loop free.
             loop = asyncio.get_event_loop()
             transcript = await loop.run_in_executor(
-                None, lambda: YouTubeTranscriptApi().fetch(video_id).to_raw_data()
+                None,
+                lambda: YouTubeTranscriptApi(
+                    proxy_config=get_transcript_proxy_config()
+                ).fetch(video_id).to_raw_data(),
             )
 
             for i, entry in enumerate(transcript):

src/agents/markdown_video_processor.py

@@ -15,6 +15,8 @@
 import aiohttp
 from dotenv import load_dotenv
 
+from youtube_extension.utils.proxy import redact_proxy_credentials
+
 # Load environment variables
 load_dotenv()
 
@@ -54,10 +56,14 @@ async def get_video_metadata(self, video_id: str) -> dict[str, Any]:
             try:
                 transcript_data = await proxy.get_transcript(video_id)
             except Exception as e:
-                logger.warning(f"Transcript not available: {e}")
+                logger.warning(
+                    "Transcript not available: %s", redact_proxy_credentials(e)
+                )
 
         except Exception as e:
-            logger.warning(f"MCP proxy failed, using direct API: {e}")
+            logger.warning(
+                "MCP proxy failed, using direct API: %s", redact_proxy_credentials(e)
+            )
             # Fallback to direct API
             url = "https://www.googleapis.com/youtube/v3/videos"
             params = {

src/agents/process_video_with_mcp.py

@@ -32,6 +32,8 @@
 from pathlib import Path
 from typing import Any
 
+from youtube_extension.utils.proxy import get_proxy_url, get_transcript_proxy_config
+
 # Load environment variables from project root .env if present
 try:
     from dotenv import load_dotenv  # type: ignore
@@ -219,7 +221,10 @@ async def _extract_transcript_with_rotation(self, video_id: str) -> list[dict[st
             if YouTubeTranscriptApi is not None:
                 # youtube-transcript-api >=1.0 instance API
                 transcript = await loop.run_in_executor(
-                    None, lambda: YouTubeTranscriptApi().fetch(video_id).to_raw_data()
+                    None,
+                    lambda: YouTubeTranscriptApi(
+                        proxy_config=get_transcript_proxy_config()
+                    ).fetch(video_id).to_raw_data(),
                 )
                 if transcript:
                     return transcript
@@ -230,7 +235,10 @@ async def _extract_transcript_with_rotation(self, video_id: str) -> list[dict[st
         try:
             if YouTubeTranscriptApi is not None:
                 transcript_list = await loop.run_in_executor(
-                    None, lambda: YouTubeTranscriptApi().list(video_id)  # type: ignore[union-attr]
+                    None,
+                    lambda: YouTubeTranscriptApi(
+                        proxy_config=get_transcript_proxy_config()
+                    ).list(video_id),  # type: ignore[union-attr]
                 )
                 fetch_tasks = [
                     loop.run_in_executor(None, lambda t=t: t.fetch().to_raw_data())
@@ -248,7 +256,11 @@ async def _extract_transcript_with_rotation(self, video_id: str) -> list[dict[st
 
         # 3) yt-dlp fallback (mocked in tests)
         try:
-            with yt_dlp.YoutubeDL({"quiet": True}) as ydl:  # type: ignore[attr-defined]
+            ydl_options: dict[str, Any] = {"quiet": True}
+            proxy_url = get_proxy_url()
+            if proxy_url:
+                ydl_options["proxy"] = proxy_url
+            with yt_dlp.YoutubeDL(ydl_options) as ydl:  # type: ignore[attr-defined]
                 _ = ydl.extract_info(f"https://www.youtube.com/watch?v={video_id}", download=False)
                 return [{"text": "Transcript extracted via yt-dlp", "start": 0.0, "duration": 0.0}]
         except Exception:
@@ -390,4 +402,3 @@ async def main() -> dict[str, Any]:
 if __name__ == "__main__":
     asyncio.run(main())
 
-

src/integration/youtube_api.py

@@ -12,6 +12,8 @@
 import httpx
 from youtube_transcript_api import YouTubeTranscriptApi
 
+from youtube_extension.utils.proxy import get_transcript_proxy_config
+
 
 @dataclass
 class VideoMetadata:
@@ -96,7 +98,9 @@ async def get_transcript(
         loop = asyncio.get_event_loop()
         transcript = await loop.run_in_executor(
             None,
-            lambda: YouTubeTranscriptApi().fetch(video_id, languages=languages).to_raw_data()
+            lambda: YouTubeTranscriptApi(
+                proxy_config=get_transcript_proxy_config()
+            ).fetch(video_id, languages=languages).to_raw_data()
         )
 
         return [

src/mcp/mcp_video_processor.py

@@ -20,6 +20,7 @@
 from typing import Any
 
 from utils.path_utils import select_readable_file, select_writable_dir
+from youtube_extension.utils.proxy import get_transcript_proxy_config
 
 # MCP integration imports
 try:
@@ -691,7 +692,9 @@ async def _direct_extraction():
             # executor — otherwise it stalls the event loop and defeats the
             # @timeout_protection / circuit-breaker hanging protection.
             loop = asyncio.get_event_loop()
-            yt_api = YouTubeTranscriptApi()
+            yt_api = YouTubeTranscriptApi(
+                proxy_config=get_transcript_proxy_config()
+            )
             transcript = await loop.run_in_executor(
                 None,
                 lambda: yt_api.fetch(
@@ -715,7 +718,9 @@ async def _routed_extraction():
             # These are blocking network calls — run them in an executor to keep
             # the event loop free and let the timeout protection work.
             loop = asyncio.get_event_loop()
-            yt_api = YouTubeTranscriptApi()
+            yt_api = YouTubeTranscriptApi(
+                proxy_config=get_transcript_proxy_config()
+            )
             transcript_list = await loop.run_in_executor(
                 None, lambda: yt_api.list(video_id)
             )

src/youtube_extension/backend/services/youtube/adapters/official_api.py

@@ -17,7 +17,10 @@
 import httpx
 
 from youtube_extension.utils import extract_video_id
-from youtube_extension.utils.proxy import get_transcript_proxy_config
+from youtube_extension.utils.proxy import (
+    get_transcript_proxy_config,
+    redact_proxy_credentials,
+)
 
 # Fallback transcript retrieval
 try:
@@ -262,7 +265,11 @@ async def get_video_transcript(self, video_id_or_url: str, language: str = 'en')
                 except CouldNotRetrieveTranscript as e:
                     logger.error(f"❌ Could not retrieve transcript for {video_id}: {e}")
                 except Exception as e:
-                    logger.warning(f"youtube-transcript-api fetch failed for {video_id}: {e}")
+                    logger.warning(
+                        "youtube-transcript-api fetch failed for %s: %s",
+                        video_id,
+                        redact_proxy_credentials(e),
+                    )
 
             # Fallback via robust service if needed
             if (not transcript_data) and HAS_ROBUST_TRANSCRIPT_FALLBACK:

src/youtube_extension/backend/services/youtube/adapters/robust.py

@@ -20,7 +20,11 @@
 
 import httpx
 
-from youtube_extension.utils.proxy import get_proxy_url, get_transcript_proxy_config
+from youtube_extension.utils.proxy import (
+    get_proxy_url,
+    get_transcript_proxy_config,
+    redact_proxy_credentials,
+)
 
 # Import our cost monitor
 try:
@@ -487,7 +491,10 @@ def _list_and_fetch() -> Any:
                 segments = transcript_data.get("segments", [])
                 return True, len(segments)
         except Exception as e:
-            logger.debug(f"Transcript availability check failed: {e}")
+            logger.debug(
+                "Transcript availability check failed: %s",
+                redact_proxy_credentials(e),
+            )
 
         return False, 0
 
@@ -521,7 +528,11 @@ async def get_transcript(
                         f"YouTubeTranscriptApi.fetch() returned {len(transcript) if transcript else 0} segments"
                     )
                 except Exception as fetch_err:
-                    api_error = f"YouTubeTranscriptApi.fetch failed: {type(fetch_err).__name__}: {fetch_err}"
+                    api_error = (
+                        "YouTubeTranscriptApi.fetch failed: "
+                        f"{type(fetch_err).__name__}: "
+                        f"{redact_proxy_credentials(fetch_err)}"
+                    )
                     logger.warning(api_error)
                     transcript_errors.append(api_error)
                     # Try instance list() as fallback — reuse the same proxy
@@ -543,7 +554,11 @@ def _list_fallback() -> Any:
                             f"YouTubeTranscriptApi.list() returned {len(transcript) if transcript else 0} segments"
                         )
                     except Exception as list_err:
-                        api_error = f"YouTubeTranscriptApi.list() fallback failed: {type(list_err).__name__}: {list_err}"
+                        api_error = (
+                            "YouTubeTranscriptApi.list() fallback failed: "
+                            f"{type(list_err).__name__}: "
+                            f"{redact_proxy_credentials(list_err)}"
+                        )
                         logger.warning(api_error)
                         transcript_errors.append(api_error)
                         transcript = []
@@ -592,7 +607,8 @@ def _list_fallback() -> Any:
 
             except Exception as e:
                 error_msg = (
-                    f"YouTube Transcript API outer exception: {type(e).__name__}: {e}"
+                    "YouTube Transcript API outer exception: "
+                    f"{type(e).__name__}: {redact_proxy_credentials(e)}"
                 )
                 logger.warning(error_msg)
                 transcript_errors.append(error_msg)

src/youtube_extension/processors/enhanced_extractor.py

@@ -22,7 +22,6 @@
     import yt_dlp
     from googleapiclient.discovery import build
     from googleapiclient.errors import HttpError
-    from youtube_transcript_api import get_transcript
     from youtube_transcript_api._errors import (
         CouldNotRetrieveTranscript,
         NoTranscriptFound,

src/youtube_extension/utils/proxy.py

@@ -67,8 +67,18 @@ def get_proxy_url() -> str | None:
     url = os.getenv(_PROXY_ENV_VAR, "").strip()
     if not url:
         return None
-    parsed = urllib.parse.urlparse(url)
-    if parsed.scheme not in ("http", "https", "socks5") or not parsed.hostname:
+    parsed: urllib.parse.ParseResult | None = None
+    try:
+        parsed = urllib.parse.urlparse(url)
+        hostname = parsed.hostname
+        parsed.port
+    except ValueError:
+        hostname = None
+    if (
+        parsed is None
+        or parsed.scheme not in ("http", "https", "socks5")
+        or not hostname
+    ):
         logger.warning(
             "%s is set but malformed — falling back to direct connection",
             _PROXY_ENV_VAR,