Add yt-downloader: single-URL YouTube downloads through an isolated Mullvad WireGuard tunnel

yt_downloader/downloader.py

@@ -0,0 +1,86 @@
+"""yt-dlp integration: download a single YouTube video (or its audio)."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+import yt_dlp
+
+log = logging.getLogger(__name__)
+
+_BASE_OPTS = {
+    "quiet": True,
+    "no_warnings": True,
+    "noplaylist": True,
+    "ignoreerrors": True,
+    "no_color": True,
+}
+
+# named presets -> yt-dlp format strings. bestvideo+bestaudio requires ffmpeg
+# to mux; the trailing /best fallbacks cover formats without a separate stream.
+QUALITY_FORMATS = {
+    "best": "bestvideo+bestaudio/best",
+    "1080": "bestvideo[height<=1080]+bestaudio/best[height<=1080]",
+    "720": "bestvideo[height<=720]+bestaudio/best[height<=720]",
+    "480": "bestvideo[height<=480]+bestaudio/best[height<=480]",
+    "audio": "bestaudio/best",
+}
+
+# suffixes yt-dlp leaves behind while downloading / on interrupted runs
+_INCOMPLETE = (".part", ".ytdl", ".tmp")
+
+
+def cookie_opts(cfg) -> dict:
+    """yt-dlp options for browser cookies (bot-check mitigation)."""
+    opts: dict = {}
+    if cfg.cookies_from_browser:
+        opts["cookiesfrombrowser"] = (cfg.cookies_from_browser,)
+    elif cfg.cookies_file:
+        opts["cookiefile"] = cfg.cookies_file
+    return opts
+
+
+def _select_format(quality: str, format_override: Optional[str]) -> str:
+    if format_override:
+        return format_override
+    return QUALITY_FORMATS.get(quality, QUALITY_FORMATS["best"])
+
+
+def download_video(
+    url: str,
+    out_dir: Path,
+    quality: str = "best",
+    format_override: Optional[str] = None,
+    extra_opts: dict | None = None,
+) -> Optional[Path]:
+    """Download a single video into out_dir. Returns the resulting file path."""
+    out_dir = out_dir.expanduser()
+    out_dir.mkdir(parents=True, exist_ok=True)
+    fmt = _select_format(quality, format_override)
+    opts = {
+        **_BASE_OPTS,
+        **(extra_opts or {}),
+        "format": fmt,
+        "outtmpl": str(out_dir / "%(title).150B [%(id)s].%(ext)s"),
+        "nocheckcertificate": True,
+    }
+    if quality != "audio" and not format_override:
+        opts["merge_output_format"] = "mp4"
+    try:
+        with yt_dlp.YoutubeDL(opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+        if info is None:
+            return None
+        actual_id = info.get("id")
+        if not actual_id:
+            return None
+        marker = f" [{actual_id}]."
+        for f in out_dir.iterdir():
+            if f.is_file() and marker in f.name and f.suffix.lower() not in _INCOMPLETE:
+                return f
+    except Exception as exc:  # noqa: BLE001
+        log.warning("Download failed for %s: %s", url, exc)
+        return None
+    return None