feat: enhance YouTubeDownloader with cookie support and player client cycling

app/services/youtube_downloader.py

@@ -1,17 +1,58 @@
 """
 YouTube audio download helper using yt-dlp with validation.
+
+Handles YouTube bot detection by cycling through player clients
+and supporting cookie-based authentication via environment variables.
 """
 
 from __future__ import annotations
 
+import base64
+import os
+import tempfile
 from dataclasses import dataclass
 from pathlib import Path
 from typing import List, Optional
 
 import yt_dlp
 
+from .logging_config import get_logger
 from .validation import sanitize_filename, validate_video_id
 
+logger = get_logger(__name__)
+
+# Player clients to try in order — android/ios often bypass bot detection
+_PLAYER_CLIENTS: List[List[str]] = [
+    ["android", "web"],
+    ["ios", "web"],
+    ["tv", "web"],
+    ["default"],
+]
+
+
+def _get_cookie_path() -> Optional[str]:
+    """Resolve cookie file path from environment variables.
+
+    Supports:
+      - YOUTUBE_COOKIES_FILE: direct path to a Netscape-format cookies.txt
+      - YOUTUBE_COOKIES_BASE64: base64-encoded cookies.txt content (for Spaces secrets)
+    """
+    cookie_file = os.getenv("YOUTUBE_COOKIES_FILE")
+    if cookie_file and Path(cookie_file).is_file():
+        return cookie_file
+
+    cookie_b64 = os.getenv("YOUTUBE_COOKIES_BASE64")
+    if cookie_b64:
+        try:
+            raw = base64.b64decode(cookie_b64)
+            tmp = Path(tempfile.gettempdir()) / "yt_cookies.txt"
+            tmp.write_bytes(raw)
+            return str(tmp)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning(f"Failed to decode YOUTUBE_COOKIES_BASE64: {exc}")
+
+    return None
+
 
 @dataclass
 class DownloadResult:
@@ -30,23 +71,23 @@ class YouTubeDownloader:
     def download(self, url: str, video_id: str) -> DownloadResult:
         if not url or not url.strip():
             raise ValueError("URL cannot be empty")
-        
+
         if not video_id or not video_id.strip():
             raise ValueError("Video ID cannot be empty")
-        
+
         if not validate_video_id(video_id):
             raise ValueError("Invalid video ID format")
-        
+
         warnings: List[str] = []
 
-        info = self._download_with_ffmpeg(url, video_id)
+        info = self._download_with_ffmpeg(url, video_id, warnings)
         if info is None:
-            info = self._download_without_ffmpeg(url, video_id)
+            info = self._download_without_ffmpeg(url, video_id, warnings)
             warnings.append("ffmpeg_unavailable")
 
         file_path = self._resolve_output_path(video_id)
         audio_format = file_path.suffix.lstrip(".") or (info.get("ext") if info else None)
-        
+
         title = info.get("title") if info else None
         if title:
             title = sanitize_filename(title)
@@ -59,42 +100,79 @@ class YouTubeDownloader:
             warnings=warnings,
         )
 
-    def _download_with_ffmpeg(self, url: str, video_id: str) -> Optional[dict]:
+    def _base_options(self, video_id: str) -> dict:
+        """Build base yt-dlp options with bot-detection mitigations."""
         safe_video_id = sanitize_filename(video_id)
-        options = {
-            "format": "bestaudio/best",
+        opts: dict = {
             "outtmpl": str(self.output_dir / f"{safe_video_id}.%(ext)s"),
             "noplaylist": True,
             "quiet": True,
             "no_warnings": True,
-            "postprocessors": [
+            "socket_timeout": 30,
+            "retries": 3,
+            "http_headers": {
+                "User-Agent": (
+                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) "
+                    "Chrome/131.0.0.0 Safari/537.36"
+                ),
+            },
+        }
+
+        cookie_path = _get_cookie_path()
+        if cookie_path:
+            opts["cookiefile"] = cookie_path
+            logger.debug("Using cookie file for YouTube authentication")
+
+        return opts
+
+    def _try_with_clients(
+        self, url: str, options: dict, with_postprocessor: bool
+    ) -> Optional[dict]:
+        """Try downloading with different player client combinations."""
+        if with_postprocessor:
+            options["format"] = "bestaudio/best"
+            options["postprocessors"] = [
                 {
                     "key": "FFmpegExtractAudio",
                     "preferredcodec": "wav",
                     "preferredquality": "192",
                 }
-            ],
-        }
-        try:
-            with yt_dlp.YoutubeDL(options) as ydl:
-                return ydl.extract_info(url, download=True)
-        except Exception:
-            return None
+            ]
+        else:
+            options["format"] = "bestaudio[ext=m4a]/bestaudio[ext=mp3]/bestaudio"
 
-    def _download_without_ffmpeg(self, url: str, video_id: str) -> Optional[dict]:
-        safe_video_id = sanitize_filename(video_id)
-        options = {
-            "format": "bestaudio[ext=m4a]/bestaudio[ext=mp3]/bestaudio",
-            "outtmpl": str(self.output_dir / f"{safe_video_id}.%(ext)s"),
-            "noplaylist": True,
-            "quiet": True,
-            "no_warnings": True,
-        }
-        try:
-            with yt_dlp.YoutubeDL(options) as ydl:
-                return ydl.extract_info(url, download=True)
-        except Exception:
-            return None
+        for clients in _PLAYER_CLIENTS:
+            opts = {**options}
+            opts["extractor_args"] = {"youtube": {"player_client": clients}}
+            client_label = "+".join(clients)
+            try:
+                logger.debug(f"Trying player_client={client_label}")
+                with yt_dlp.YoutubeDL(opts) as ydl:
+                    return ydl.extract_info(url, download=True)
+            except Exception as exc:  # noqa: BLE001
+                logger.debug(f"player_client={client_label} failed: {exc}")
+                continue
+
+        return None
+
+    def _download_with_ffmpeg(
+        self, url: str, video_id: str, warnings: List[str]
+    ) -> Optional[dict]:
+        opts = self._base_options(video_id)
+        result = self._try_with_clients(url, opts, with_postprocessor=True)
+        if result is None:
+            logger.warning("All player clients failed with ffmpeg postprocessor")
+        return result
+
+    def _download_without_ffmpeg(
+        self, url: str, video_id: str, warnings: List[str]
+    ) -> Optional[dict]:
+        opts = self._base_options(video_id)
+        result = self._try_with_clients(url, opts, with_postprocessor=False)
+        if result is None:
+            logger.warning("All player clients failed without ffmpeg postprocessor")
+        return result
 
     def _resolve_output_path(self, video_id: str) -> Path:
         safe_video_id = sanitize_filename(video_id)