feat(external): file-host fetcher subsystem (Phase 4a)

Shared, reusable fetchers for the 5 off-platform hosts behind one signature
(fetch_external(host, url, dest_dir, ...) -> FetchResult):
- dropbox    : force dl=1 + stream GET
- pixeldrain : GET /api/file/{id}
- mediafire  : scrape the download page for the direct link + stream GET
- gdrive     : gdown (confirm-token + virus-scan interstitial); added to reqs
- mega       : MEGAcmd `mega-get` subprocess (public link incl. #key)

HTTP/gdown/subprocess go through module seams so unit tests run without
network/gdown/MEGAcmd. fetch_external never raises — every backend failure
(transport, non-200, scrape miss, subprocess error, stop) is captured on
.error so the worker (next slice) records it and moves on. mega's binary lands
in the runtime image in a later slice; the code is complete + tested now.

Refs FC #830 (Phase 4a).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/services/external_fetch.py

@@ -0,0 +1,247 @@
+"""Fetchers for off-platform file hosts (mega / gdrive / mediafire / dropbox /
+pixeldrain).
+
+A shared, reusable subsystem: given an external_link URL, fetch the file(s) into
+a destination directory and report the outcome. The download worker (separate
+slice) drives these off the external_link ledger; any in-house downloader can
+call `fetch_external()` directly.
+
+No single tool covers all five hosts, so a small registry maps host → fetch
+function behind one signature:
+
+    fetch_external(host, url, dest_dir, *, timeout, should_stop) -> FetchResult
+
+Backends:
+  - dropbox    : force the direct-download variant (dl=1) + stream GET.
+  - pixeldrain : GET the /api/file/{id} endpoint.
+  - mediafire  : scrape the download page for the direct link + stream GET.
+  - gdrive     : gdown (handles the confirm-token + virus-scan interstitial).
+  - mega       : MEGAcmd `mega-get` subprocess (public link incl. #key); needs
+                 the binary in the runtime image.
+
+Public links work credential-free (rule 26); per-host creds are a later Settings
+concern. Plain-HTTP homelab — no secure-context API. The HTTP / gdown /
+subprocess calls go through module-level seams so unit tests run without
+network, gdown, or MEGAcmd.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import re
+import subprocess
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import requests
+
+log = logging.getLogger(__name__)
+
+_CHUNK = 1 << 16
+_DEFAULT_TIMEOUT = 600.0
+_USER_AGENT = (
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
+)
+# MediaFire's download page embeds the real file URL in a download-button href.
+_MEDIAFIRE_RE = re.compile(
+    r'href="(https://download[^"]+?\.mediafire\.com/[^"]+)"', re.IGNORECASE
+)
+_CD_FILENAME_RE = re.compile(r'filename\*?=(?:UTF-8\'\')?"?([^";]+)"?', re.IGNORECASE)
+
+
+@dataclass
+class FetchResult:
+    files: list[Path] = field(default_factory=list)
+    bytes: int = 0
+    error: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.error is None and bool(self.files)
+
+
+class ExternalFetchError(Exception):
+    """A fetch failed in a way worth recording on the link's last_error."""
+
+
+# -- seams (monkeypatched in tests) ----------------------------------------
+
+def _http_get(url: str, *, timeout: float, headers: dict | None = None,
+              stream: bool = True) -> requests.Response:
+    hdrs = {"User-Agent": _USER_AGENT}
+    if headers:
+        hdrs.update(headers)
+    return requests.get(url, timeout=timeout, headers=hdrs, stream=stream)
+
+
+def _gdown_download(url: str, out_dir: str) -> str | None:
+    """Download a Google Drive url into out_dir via gdown; return the written
+    path. Imported lazily so the dep is optional at module-import time."""
+    # Lazy import: keeps gdown an optional dep that isn't needed to import this
+    # module (e.g. the no-DB unit lane that never exercises a real fetch).
+    import gdown
+    # A trailing sep tells gdown to keep the server-side filename inside out_dir.
+    return gdown.download(url, output=out_dir + os.sep, quiet=True, fuzzy=True)
+
+
+def _run_mega_get(url: str, out_dir: str, *, timeout: float) -> None:
+    """Download a mega.nz public link (key in the #fragment) into out_dir via
+    MEGAcmd's `mega-get`. Raises ExternalFetchError on non-zero exit."""
+    # Fixed argv (not shell): only `url` is external input, and it's a positional
+    # arg to mega-get, so there's no shell-injection surface.
+    proc = subprocess.run(
+        ["mega-get", url, out_dir],
+        capture_output=True, text=True, timeout=timeout, check=False,
+    )
+    if proc.returncode != 0:
+        raise ExternalFetchError(
+            f"mega-get exit {proc.returncode}: {(proc.stderr or '').strip()[:300]}"
+        )
+
+
+# -- helpers ---------------------------------------------------------------
+
+def _safe_name(name: str, fallback: str) -> str:
+    name = os.path.basename((name or "").strip().strip('"'))
+    # Strip path separators / control chars; never empty.
+    name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", name).strip(". ")
+    return name or fallback
+
+
+def _filename_from(resp: requests.Response, url: str, fallback: str) -> str:
+    cd = resp.headers.get("Content-Disposition", "")
+    m = _CD_FILENAME_RE.search(cd)
+    if m:
+        return _safe_name(m.group(1), fallback)
+    path_name = os.path.basename(urlsplit(url).path)
+    return _safe_name(path_name, fallback)
+
+
+def _stream_to_file(resp: requests.Response, dest: Path,
+                    should_stop: Callable[[], bool]) -> int:
+    """Stream a response body to `dest` (atomic via .part). Returns byte count.
+    Honors should_stop between chunks (partial file removed)."""
+    part = dest.with_name(dest.name + ".part")
+    total = 0
+    try:
+        with part.open("wb") as fh:
+            for chunk in resp.iter_content(chunk_size=_CHUNK):
+                if should_stop():
+                    raise ExternalFetchError("stopped")
+                if chunk:
+                    fh.write(chunk)
+                    total += len(chunk)
+    except BaseException:
+        part.unlink(missing_ok=True)
+        raise
+    os.replace(part, dest)
+    return total
+
+
+def _get_to_dir(url: str, dest_dir: Path, *, timeout: float,
+                should_stop: Callable[[], bool], fallback: str,
+                headers: dict | None = None) -> FetchResult:
+    resp = _http_get(url, timeout=timeout, headers=headers, stream=True)
+    if resp.status_code != 200:
+        return FetchResult(error=f"HTTP {resp.status_code} for {url}")
+    name = _filename_from(resp, url, fallback)
+    dest = dest_dir / name
+    written = _stream_to_file(resp, dest, should_stop)
+    return FetchResult(files=[dest], bytes=written)
+
+
+# -- per-host fetchers -----------------------------------------------------
+
+def _fetch_dropbox(url: str, dest_dir: Path, *, timeout: float,
+                   should_stop: Callable[[], bool]) -> FetchResult:
+    # Force the direct-download variant: dl=1 (Dropbox serves an HTML preview
+    # for dl=0). Rewrite/insert the param rather than string-replace so ?dl=0,
+    # &dl=0, and a missing param all resolve.
+    parts = urlsplit(url)
+    q = dict(parse_qsl(parts.query))
+    q["dl"] = "1"
+    direct = urlunsplit(parts._replace(query=urlencode(q)))
+    return _get_to_dir(direct, dest_dir, timeout=timeout,
+                       should_stop=should_stop, fallback="dropbox-file")
+
+
+def _fetch_pixeldrain(url: str, dest_dir: Path, *, timeout: float,
+                      should_stop: Callable[[], bool]) -> FetchResult:
+    # /u/{id} (and /l/{id}) → the API file endpoint.
+    file_id = urlsplit(url).path.rstrip("/").split("/")[-1]
+    if not file_id:
+        return FetchResult(error=f"no pixeldrain id in {url}")
+    api = f"https://pixeldrain.com/api/file/{file_id}"
+    return _get_to_dir(api, dest_dir, timeout=timeout,
+                       should_stop=should_stop, fallback=f"{file_id}.bin")
+
+
+def _fetch_mediafire(url: str, dest_dir: Path, *, timeout: float,
+                     should_stop: Callable[[], bool]) -> FetchResult:
+    page = _http_get(url, timeout=timeout, stream=False)
+    if page.status_code != 200:
+        return FetchResult(error=f"HTTP {page.status_code} for mediafire page")
+    m = _MEDIAFIRE_RE.search(page.text or "")
+    if not m:
+        return FetchResult(error="mediafire direct link not found on page")
+    return _get_to_dir(m.group(1), dest_dir, timeout=timeout,
+                       should_stop=should_stop, fallback="mediafire-file")
+
+
+def _fetch_gdrive(url: str, dest_dir: Path, *, timeout: float,
+                  should_stop: Callable[[], bool]) -> FetchResult:
+    out = _gdown_download(url, str(dest_dir))
+    if not out:
+        return FetchResult(error="gdown returned no file (quota / private?)")
+    p = Path(out)
+    if not p.exists():
+        return FetchResult(error=f"gdown reported {out} but it is missing")
+    return FetchResult(files=[p], bytes=p.stat().st_size)
+
+
+def _fetch_mega(url: str, dest_dir: Path, *, timeout: float,
+                should_stop: Callable[[], bool]) -> FetchResult:
+    before = set(dest_dir.iterdir()) if dest_dir.exists() else set()
+    _run_mega_get(url, str(dest_dir), timeout=timeout)
+    new = [p for p in dest_dir.iterdir() if p not in before and p.is_file()]
+    if not new:
+        return FetchResult(error="mega-get wrote no new file")
+    return FetchResult(files=new, bytes=sum(p.stat().st_size for p in new))
+
+
+_REGISTRY: dict[str, Callable[..., FetchResult]] = {
+    "dropbox": _fetch_dropbox,
+    "pixeldrain": _fetch_pixeldrain,
+    "mediafire": _fetch_mediafire,
+    "gdrive": _fetch_gdrive,
+    "mega": _fetch_mega,
+}
+
+SUPPORTED_HOSTS = tuple(_REGISTRY)
+
+
+def fetch_external(host: str, url: str, dest_dir: Path, *,
+                   timeout: float = _DEFAULT_TIMEOUT,
+                   should_stop: Callable[[], bool] = lambda: False) -> FetchResult:
+    """Fetch `url` (a `host` link) into `dest_dir`. Returns a FetchResult; never
+    raises — any backend error (transport, non-200, scrape miss, subprocess
+    failure, stop) is captured on `.error` so the worker can record it and move
+    on."""
+    fetcher = _REGISTRY.get(host)
+    if fetcher is None:
+        return FetchResult(error=f"unsupported host {host!r}")
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    try:
+        return fetcher(url, dest_dir, timeout=timeout, should_stop=should_stop)
+    except requests.RequestException as exc:
+        return FetchResult(error=f"transport error: {exc}")
+    except subprocess.TimeoutExpired:
+        return FetchResult(error="timed out")
+    except ExternalFetchError as exc:
+        return FetchResult(error=str(exc))
+    except Exception as exc:  # never let a backend quirk kill the worker
+        log.warning("external fetch (%s) failed for %s: %s", host, url, exc)
+        return FetchResult(error=f"{type(exc).__name__}: {exc}")

requirements.txt

@@ -38,3 +38,8 @@ nh3>=0.2,<0.3
 # Archive extraction (FC-2d-iii filesystem-import aid)
 rarfile>=4.2,<5
 py7zr>=1,<2
+
+# Google Drive fetcher for off-platform file-host links (external downloads,
+# #830). Handles Drive's confirm-token + virus-scan interstitial. mega.nz uses
+# the MEGAcmd `mega-get` binary instead (added to the runtime image, not pip).
+gdown>=5,<6

tests/test_external_fetch.py

@@ -0,0 +1,141 @@
+"""Unit tests for external_fetch — no network/gdown/MEGAcmd (seams stubbed)."""
+
+from pathlib import Path
+
+import pytest
+import requests
+
+import backend.app.services.external_fetch as ef
+from backend.app.services.external_fetch import fetch_external
+
+
+def _resp(*, status=200, headers=None, chunks=(b"hello",), text=""):
+    class _R:
+        def __init__(self):
+            self.status_code = status
+            self.headers = headers or {}
+            self.text = text
+
+        def iter_content(self, chunk_size=65536):
+            yield from chunks
+
+    return _R()
+
+
+def test_dropbox_forces_dl1_and_downloads(tmp_path, monkeypatch):
+    seen = {}
+
+    def fake_get(url, *, timeout, headers=None, stream=True):
+        seen["url"] = url
+        return _resp(headers={"Content-Disposition": 'attachment; filename="film.zip"'})
+
+    monkeypatch.setattr(ef, "_http_get", fake_get)
+    res = fetch_external(
+        "dropbox", "https://www.dropbox.com/s/abc/film.zip?dl=0", tmp_path
+    )
+    assert res.ok
+    assert "dl=1" in seen["url"]
+    assert res.files[0].name == "film.zip"
+    assert res.files[0].read_bytes() == b"hello"
+
+
+def test_pixeldrain_hits_api_endpoint(tmp_path, monkeypatch):
+    seen = {}
+
+    def fake_get(url, *, timeout, headers=None, stream=True):
+        seen["url"] = url
+        return _resp()
+
+    monkeypatch.setattr(ef, "_http_get", fake_get)
+    res = fetch_external("pixeldrain", "https://pixeldrain.com/u/ems4UomN", tmp_path)
+    assert res.ok
+    assert seen["url"] == "https://pixeldrain.com/api/file/ems4UomN"
+
+
+def test_mediafire_scrapes_then_downloads(tmp_path, monkeypatch):
+    calls = []
+
+    def fake_get(url, *, timeout, headers=None, stream=True):
+        calls.append(url)
+        if "mediafire.com/file" in url:
+            return _resp(
+                text='<a id="downloadButton" '
+                'href="https://download1234.mediafire.com/abc/film.zip">dl</a>'
+            )
+        return _resp(headers={"Content-Disposition": 'filename="film.zip"'})
+
+    monkeypatch.setattr(ef, "_http_get", fake_get)
+    res = fetch_external("mediafire", "https://www.mediafire.com/file/x/film.zip", tmp_path)
+    assert res.ok
+    assert res.files[0].name == "film.zip"
+    assert calls[1].startswith("https://download1234.mediafire.com/")
+
+
+def test_mediafire_missing_link_errors(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_http_get", lambda *a, **k: _resp(text="<html>none</html>"))
+    res = fetch_external("mediafire", "https://www.mediafire.com/file/x", tmp_path)
+    assert not res.ok
+    assert "not found" in res.error
+
+
+def test_gdrive_uses_gdown(tmp_path, monkeypatch):
+    def fake_gdown(url, out_dir):
+        p = Path(out_dir) / "drive.bin"
+        p.write_bytes(b"xyz")
+        return str(p)
+
+    monkeypatch.setattr(ef, "_gdown_download", fake_gdown)
+    res = fetch_external("gdrive", "https://drive.google.com/file/d/ID/view", tmp_path)
+    assert res.ok
+    assert res.files[0].name == "drive.bin"
+    assert res.bytes == 3
+
+
+def test_gdrive_no_file_errors(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_gdown_download", lambda url, out_dir: None)
+    res = fetch_external("gdrive", "https://drive.google.com/file/d/ID/view", tmp_path)
+    assert not res.ok
+    assert "gdown" in res.error
+
+
+def test_mega_collects_new_files(tmp_path, monkeypatch):
+    def fake_mega(url, out_dir, *, timeout):
+        (Path(out_dir) / "film.mp4").write_bytes(b"vid")
+
+    monkeypatch.setattr(ef, "_run_mega_get", fake_mega)
+    res = fetch_external("mega", "https://mega.nz/file/x#k", tmp_path)
+    assert res.ok
+    assert res.files[0].name == "film.mp4"
+
+
+def test_unsupported_host(tmp_path):
+    res = fetch_external("nope", "https://x", tmp_path)
+    assert not res.ok
+    assert "unsupported" in res.error
+
+
+def test_non_200_errors(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_http_get", lambda *a, **k: _resp(status=404))
+    res = fetch_external("pixeldrain", "https://pixeldrain.com/u/x", tmp_path)
+    assert not res.ok
+    assert "404" in res.error
+
+
+def test_transport_error_captured(tmp_path, monkeypatch):
+    def boom(*a, **k):
+        raise requests.ConnectionError("down")
+
+    monkeypatch.setattr(ef, "_http_get", boom)
+    res = fetch_external("dropbox", "https://www.dropbox.com/s/x?dl=0", tmp_path)
+    assert not res.ok
+    assert "transport" in res.error
+
+
+def test_should_stop_aborts_and_cleans_part(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_http_get", lambda *a, **k: _resp(chunks=(b"a", b"b")))
+    res = fetch_external(
+        "pixeldrain", "https://pixeldrain.com/u/x", tmp_path,
+        should_stop=lambda: True,
+    )
+    assert not res.ok
+    assert list(tmp_path.glob("*.part")) == []