feat: enhance YouTubeDownloader with cookie handling and error normalization

app/services/youtube_downloader.py

@@ -9,8 +9,9 @@ from __future__ import annotations
 
 import base64
 import os
+import re
 import tempfile
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List, Optional
 
@@ -21,7 +22,7 @@ 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 to try in order. Android/iOS often bypass stricter web checks.
 _PLAYER_CLIENTS: List[List[str]] = [
     ["android", "web"],
     ["ios", "web"],
@@ -29,23 +30,65 @@ _PLAYER_CLIENTS: List[List[str]] = [
     ["default"],
 ]
 
+_BROWSER_COOKIE_SPEC_RE = re.compile(
+    r"""(?x)
+    (?P<name>[^+:]+)
+    (?:\s*\+\s*(?P<keyring>[^:]+))?
+    (?:\s*:\s*(?!:)(?P<profile>.+?))?
+    (?:\s*::\s*(?P<container>.+))?
+    """
+)
 
-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)
-    """
+def _normalize_error_message(message: object) -> str:
+    text = str(message).replace("\n", " ").strip()
+    text = re.sub(r"^\s*ERROR:\s*", "", text, flags=re.IGNORECASE)
+    return " ".join(text.split())
+
+
+def _normalized_error_lower(message: str) -> str:
+    return _normalize_error_message(message).replace("’", "'").lower()
+
+
+def _is_auth_required_message(message: str) -> bool:
+    normalized = _normalized_error_lower(message)
+    return (
+        "sign in to confirm you're not a bot" in normalized
+        or "use --cookies-from-browser or --cookies" in normalized
+    )
+
+
+def _is_ffmpeg_missing_message(message: str) -> bool:
+    normalized = _normalized_error_lower(message)
+    return "ffmpeg" in normalized and (
+        "not found" in normalized
+        or "--ffmpeg-location" in normalized
+        or "please install" in normalized
+    )
+
+
+def _is_ffmpeg_postprocess_message(message: str) -> bool:
+    normalized = _normalized_error_lower(message)
+    return "ffmpeg" in normalized and (
+        "postprocessing" in normalized
+        or "post-process" in normalized
+        or "extractaudio" in normalized
+    )
+
+
+def _get_cookie_path() -> Optional[str]:
+    """Resolve cookie file path from environment variables."""
     cookie_file = os.getenv("YOUTUBE_COOKIES_FILE")
-    if cookie_file and Path(cookie_file).is_file():
-        return cookie_file
+    if cookie_file:
+        if Path(cookie_file).is_file():
+            return cookie_file
+        logger.warning(f"YOUTUBE_COOKIES_FILE does not exist: {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 = Path(tempfile.gettempdir()) / "crowncode_yt_cookies.txt"
             tmp.write_bytes(raw)
             return str(tmp)
         except Exception as exc:  # noqa: BLE001
@@ -54,6 +97,30 @@ def _get_cookie_path() -> Optional[str]:
     return None
 
 
+def _get_browser_cookie_config() -> Optional[tuple[str, Optional[str], Optional[str], Optional[str]]]:
+    """Resolve yt-dlp cookies-from-browser config from environment variables."""
+    raw = os.getenv("YOUTUBE_COOKIES_FROM_BROWSER")
+    if not raw:
+        return None
+
+    match = _BROWSER_COOKIE_SPEC_RE.fullmatch(raw.strip())
+    if match is None:
+        raise ValueError(
+            "Invalid YOUTUBE_COOKIES_FROM_BROWSER format. "
+            "Expected BROWSER[+KEYRING][:PROFILE][::CONTAINER]."
+        )
+
+    browser_name, keyring, profile, container = match.group(
+        "name", "keyring", "profile", "container"
+    )
+    return (
+        browser_name.lower(),
+        profile,
+        keyring.upper() if keyring else None,
+        container,
+    )
+
+
 @dataclass
 class DownloadResult:
     file_path: Path
@@ -63,6 +130,45 @@ class DownloadResult:
     warnings: List[str]
 
 
+@dataclass
+class DownloadAttemptFailure:
+    client_label: str
+    message: str
+
+
+@dataclass
+class DownloadAttemptResult:
+    info: Optional[dict]
+    failures: List[DownloadAttemptFailure] = field(default_factory=list)
+
+
+class YouTubeDownloadError(RuntimeError):
+    def __init__(
+        self,
+        error_code: str,
+        message: str,
+        warnings: Optional[List[str]] = None,
+    ) -> None:
+        super().__init__(message)
+        self.error_code = error_code
+        self.warnings = warnings or []
+
+
+class _YtDlpLogCollector:
+    def __init__(self) -> None:
+        self.last_error: Optional[str] = None
+
+    def debug(self, message: str) -> None:
+        logger.debug(f"yt-dlp: {_normalize_error_message(message)}")
+
+    def warning(self, message: str) -> None:
+        logger.debug(f"yt-dlp warning: {_normalize_error_message(message)}")
+
+    def error(self, message: str) -> None:
+        self.last_error = _normalize_error_message(message)
+        logger.debug(f"yt-dlp error: {self.last_error}")
+
+
 class YouTubeDownloader:
     def __init__(self, output_dir: Path) -> None:
         self.output_dir = output_dir
@@ -80,10 +186,20 @@ class YouTubeDownloader:
 
         warnings: List[str] = []
 
-        info = self._download_with_ffmpeg(url, video_id, warnings)
+        ffmpeg_result = self._download_with_ffmpeg(url, video_id)
+        info = ffmpeg_result.info
+
         if info is None:
-            info = self._download_without_ffmpeg(url, video_id, warnings)
-            warnings.append("ffmpeg_unavailable")
+            raw_result = self._download_without_ffmpeg(url, video_id)
+            info = raw_result.info
+
+            if info is not None:
+                if any(_is_ffmpeg_missing_message(f.message) for f in ffmpeg_result.failures):
+                    warnings.append("ffmpeg_unavailable")
+                elif any(_is_ffmpeg_postprocess_message(f.message) for f in ffmpeg_result.failures):
+                    warnings.append("ffmpeg_postprocess_failed")
+            else:
+                self._raise_download_error(ffmpeg_result.failures + raw_result.failures)
 
         file_path = self._resolve_output_path(video_id)
         audio_format = file_path.suffix.lstrip(".") or (info.get("ext") if info else None)
@@ -123,13 +239,24 @@ class YouTubeDownloader:
         if cookie_path:
             opts["cookiefile"] = cookie_path
             logger.debug("Using cookie file for YouTube authentication")
+            return opts
+
+        browser_cookie_config = _get_browser_cookie_config()
+        if browser_cookie_config:
+            opts["cookiesfrombrowser"] = browser_cookie_config
+            logger.debug(
+                "Using browser cookies for YouTube authentication "
+                f"({browser_cookie_config[0]})"
+            )
 
         return opts
 
     def _try_with_clients(
         self, url: str, options: dict, with_postprocessor: bool
-    ) -> Optional[dict]:
+    ) -> DownloadAttemptResult:
         """Try downloading with different player client combinations."""
+        failures: List[DownloadAttemptFailure] = []
+
         if with_postprocessor:
             options["format"] = "bestaudio/best"
             options["postprocessors"] = [
@@ -145,35 +272,66 @@ class YouTubeDownloader:
         for clients in _PLAYER_CLIENTS:
             opts = {**options}
             opts["extractor_args"] = {"youtube": {"player_client": clients}}
+            opts["logger"] = _YtDlpLogCollector()
             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)
+                    return DownloadAttemptResult(
+                        info=ydl.extract_info(url, download=True),
+                        failures=failures,
+                    )
             except Exception as exc:  # noqa: BLE001
-                logger.debug(f"player_client={client_label} failed: {exc}")
+                collector = opts["logger"]
+                error_message = collector.last_error or _normalize_error_message(exc)
+                failures.append(
+                    DownloadAttemptFailure(
+                        client_label=client_label,
+                        message=error_message,
+                    )
+                )
+                logger.debug(f"player_client={client_label} failed: {error_message}")
                 continue
 
-        return None
+        return DownloadAttemptResult(info=None, failures=failures)
 
-    def _download_with_ffmpeg(
-        self, url: str, video_id: str, warnings: List[str]
-    ) -> Optional[dict]:
+    def _download_with_ffmpeg(self, url: str, video_id: str) -> DownloadAttemptResult:
         opts = self._base_options(video_id)
         result = self._try_with_clients(url, opts, with_postprocessor=True)
-        if result is None:
+        if result.info 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]:
+    def _download_without_ffmpeg(self, url: str, video_id: str) -> DownloadAttemptResult:
         opts = self._base_options(video_id)
         result = self._try_with_clients(url, opts, with_postprocessor=False)
-        if result is None:
+        if result.info is None:
             logger.warning("All player clients failed without ffmpeg postprocessor")
         return result
 
+    def _raise_download_error(self, failures: List[DownloadAttemptFailure]) -> None:
+        if failures:
+            auth_failure = next(
+                (failure for failure in reversed(failures) if _is_auth_required_message(failure.message)),
+                None,
+            )
+            if auth_failure is not None:
+                raise YouTubeDownloadError(
+                    error_code="youtube_authentication_required",
+                    message=auth_failure.message,
+                    warnings=["youtube_authentication_required"],
+                )
+
+            raise YouTubeDownloadError(
+                error_code="youtube_download_failed",
+                message=failures[-1].message,
+            )
+
+        raise YouTubeDownloadError(
+            error_code="youtube_download_failed",
+            message="YouTube download failed without an explicit yt-dlp error.",
+        )
+
     def _resolve_output_path(self, video_id: str) -> Path:
         safe_video_id = sanitize_filename(video_id)
         candidates = list(self.output_dir.glob(f"{safe_video_id}.*"))