FabledCurator/backend/app/services/patreon_downloader.py

"""Native Patreon media downloader (build step 2b of the native ingester).

Given a Patreon post and its already-resolved `MediaItem`s (from
patreon_client.extract_media), download the media to the EXACT on-disk layout
gallery-dl produces, write a sidecar JSON the existing importer consumes, and
report per-media outcomes.

This module is PURE: no DB. The cross-run seen-ledger and the iter_posts
orchestration are a LATER step. The tier-1 (seen) skip is an INJECTED predicate
(`is_seen`), so this module needs no DB and is unit-testable without network.

On-disk layout (matches gallery-dl):
  <images_root>/<artist_slug>/patreon/<DIR>/<NN>_<filename>
  where <DIR> = "<YYYY-MM-DD>_<post_id>_<title40>" (date prefix omitted when the
  post's published_at is unparseable), <NN> is the 1-based index of the item in
  the post zero-padded to 2, and <filename> is MediaItem.filename. The sidecar
  is written next to the media as <NN>_<stem>.json (media_path.with_suffix).

Video: a MediaItem whose URL is a Mux/HLS stream (host stream.mux.com or path
endswith .m3u8) is fetched with yt-dlp (subprocess), passing the same
Referer/Origin headers gallery-dl forwards for Mux playback (see gallery_dl.py
_get_default_config). yt-dlp may remux to a container of its own choosing, so we
accept the actual output extension and record the real path.

FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
"""

from __future__ import annotations

import contextlib
import json
import logging
import os
import shutil
import subprocess
import time
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from urllib.parse import urlsplit

import requests

from .file_validator import is_validatable, validate_file
from .patreon_client import _load_session, _retry_after_seconds

log = logging.getLogger(__name__)

_TITLE_MAX = 40
_TIMEOUT_SECONDS = 120.0
_CHUNK = 1 << 16
# A CDN media GET rarely 429s, but if it does, a couple of backoff retries keep
# one throttled file from becoming a per-item error (plan #703).
_MAX_MEDIA_429_RETRIES = 2

# Referer/Origin yt-dlp must send for Mux-hosted Patreon video. Mux's JWT
# playback policy checks Referer/Origin on every request, so yt-dlp must send
# Patreon's, not its own default. (gallery-dl forwarded the same headers before
# the #697 cutover removed its Patreon path; this is now the only place they
# live.)
_VIDEO_HEADERS = {
    "Referer": "https://www.patreon.com/",
    "Origin": "https://www.patreon.com",
}

# Characters Windows/gallery-dl path-restrict forbids, plus path separators.
_FORBIDDEN = set('<>:"/\\|?*')


def _sanitize(name: str) -> str:
    """Make `name` safe for a single filesystem path segment.

    Replaces path separators, the Windows-forbidden set <>:"/\\|?* and control
    characters with `_`, then strips trailing dots/spaces (gallery-dl
    path-restrict behavior). Never returns empty (falls back to "_").
    """
    out = []
    for ch in name:
        if ch in _FORBIDDEN or ord(ch) < 32:
            out.append("_")
        else:
            out.append(ch)
    cleaned = "".join(out).rstrip(". ")
    return cleaned or "_"


def _is_video_url(url: str) -> bool:
    parts = urlsplit(url)
    if parts.hostname and parts.hostname.lower() == "stream.mux.com":
        return True
    return parts.path.lower().endswith(".m3u8")


def _post_dir_name(post: dict) -> str:
    """Build the post directory name matching gallery-dl's layout."""
    post_id = str(post.get("id") or "")
    attrs = post.get("attributes") or {}
    title = attrs.get("title")
    title = title if isinstance(title, str) else ""
    title40 = title[:_TITLE_MAX]

    published = attrs.get("published_at")
    date_prefix = None
    if isinstance(published, str) and published:
        s = published.strip()
        if s.endswith("Z"):
            s = s[:-1] + "+00:00"
        try:
            dt = datetime.fromisoformat(s)
        except ValueError:
            dt = None
        if dt is not None:
            date_prefix = f"{dt:%Y-%m-%d}"

    if date_prefix:
        raw = f"{date_prefix}_{post_id}_{title40}"
    else:
        raw = f"{post_id}_{title40}"
    return _sanitize(raw)


@dataclass
class MediaOutcome:
    """Per-media result of a download_post pass.

    status is one of: "downloaded", "skipped_seen", "skipped_disk", "error".
    `path` is the final on-disk path for "downloaded" (the actual yt-dlp output
    for video), or the path that already existed for "skipped_disk"; None for
    "skipped_seen" and (usually) "error". `error` carries the failure reason for
    "error", else None.
    """

    media: object  # MediaItem (avoid importing the name for a bare annotation)
    status: str
    path: Path | None
    error: str | None


class PatreonDownloader:
    """Download resolved Patreon media to gallery-dl's on-disk layout.

    PURE: no DB. The HTTP session and the yt-dlp invocation are injectable seams
    so tests run without network or a real subprocess:
      - pass `session=` to stub `session.get`, or monkeypatch `_fetch_to_file`.
      - monkeypatch `_run_ytdlp` to avoid spawning yt-dlp.
    """

    def __init__(
        self,
        images_root: Path,
        cookies_path: str | None = None,
        *,
        validate: bool = True,
        rate_limit: float = 0.0,
        session: requests.Session | None = None,
    ):
        self.images_root = Path(images_root)
        self.cookies_path = str(cookies_path) if cookies_path else None
        self._validate = validate
        # Politeness: seconds to sleep before each actual media download (paces
        # the CDN; honors ImportSettings.download_rate_limit_seconds, the same
        # value gallery-dl used as its between-downloads `sleep`). 0 = no pacing.
        # Applied only to real downloads, not to seen/disk skips. plan #703.
        self._rate_limit = rate_limit or 0.0
        # Build a cookie-loaded session the same way patreon_client does, so the
        # CDN GETs carry the creator's auth.
        self.session = session if session is not None else _load_session(cookies_path)

    # -- public ------------------------------------------------------------

    def download_post(
        self,
        post: dict,
        media_items: list,
        artist_slug: str,
        *,
        is_seen: Callable[[object], bool] = lambda m: False,
    ) -> list[MediaOutcome]:
        """Download every media item of one post; return per-item outcomes.

        Builds the post directory, iterates media (1-based NN), applies the
        two-tier skip (injected is_seen, then disk), downloads (plain GET or
        yt-dlp for video), writes the sidecar for each freshly-downloaded item,
        validates, and returns outcomes. Resilient: one media's failure yields
        an "error" outcome for that item; the rest proceed.
        """
        post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post)
        outcomes: list[MediaOutcome] = []

        for i, media in enumerate(media_items, start=1):
            try:
                outcomes.append(
                    self._download_one(post, media, post_dir, artist_slug, i, is_seen)
                )
            except Exception as exc:  # resilient: isolate one item's failure
                log.warning(
                    "Patreon media failed (post %s, item %d): %s",
                    post.get("id"), i, exc,
                )
                outcomes.append(
                    MediaOutcome(media=media, status="error", path=None, error=str(exc))
                )
        return outcomes

    # -- per-item ----------------------------------------------------------

    def _download_one(
        self,
        post: dict,
        media,
        post_dir: Path,
        artist_slug: str,
        index: int,
        is_seen: Callable[[object], bool],
    ) -> MediaOutcome:
        # tier-1: seen ledger (injected; no DB here).
        if is_seen(media):
            return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)

        nn = f"{index:02d}"
        final_name = _sanitize(f"{nn}_{media.filename}")
        media_path = post_dir / final_name

        # tier-2: already on disk.
        if media_path.exists():
            return MediaOutcome(
                media=media, status="skipped_disk", path=media_path, error=None
            )
        # Video may land at a different extension; honor a pre-existing remux.
        if _is_video_url(media.url):
            existing = self._existing_video_output(media_path)
            if existing is not None:
                return MediaOutcome(
                    media=media, status="skipped_disk", path=existing, error=None
                )

        post_dir.mkdir(parents=True, exist_ok=True)

        # Pace real downloads only (the skips above already returned). plan #703.
        if self._rate_limit > 0:
            time.sleep(self._rate_limit)

        if _is_video_url(media.url):
            out_path = self._run_ytdlp(media.url, media_path, _VIDEO_HEADERS)
            if out_path is None or not Path(out_path).exists():
                return MediaOutcome(
                    media=media,
                    status="error",
                    path=None,
                    error="yt-dlp produced no output",
                )
            out_path = Path(out_path)
        else:
            out_path = self._fetch_get(media.url, media_path)
            invalid = self._validate_path(out_path, artist_slug)
            if invalid is not None:
                return MediaOutcome(
                    media=media, status="error", path=None, error=invalid
                )

        self._write_sidecar(post, out_path)
        return MediaOutcome(media=media, status="downloaded", path=out_path, error=None)

    # -- download seams ----------------------------------------------------

    def _fetch_get(self, url: str, dest: Path) -> Path:
        """Stream `url` to a .part file then atomic-rename to `dest`.

        Thin wrapper over `_fetch_to_file` so tests can stub either the whole
        GET path (`_fetch_to_file`) or just `session.get`.
        """
        part = dest.with_name(dest.name + ".part")
        try:
            self._fetch_to_file(url, part)
        except Exception:
            with contextlib.suppress(OSError):
                part.unlink()
            raise
        os.replace(part, dest)
        return dest

    def _fetch_to_file(self, url: str, dest: Path) -> None:
        """Stream a non-video URL to `dest` via the (stubbable) session.

        Retries a transient 429 with backoff (plan #703); any other non-2xx
        falls through to raise_for_status → HTTPError, which download_post turns
        into a resilient per-item error outcome.
        """
        attempt = 0
        while True:
            resp = self.session.get(url, stream=True, timeout=_TIMEOUT_SECONDS)
            if resp.status_code == 429 and attempt < _MAX_MEDIA_429_RETRIES:
                attempt += 1
                delay = _retry_after_seconds(resp, attempt)
                log.warning(
                    "Patreon media 429 (%s) — backing off %.1fs (retry %d/%d)",
                    url, delay, attempt, _MAX_MEDIA_429_RETRIES,
                )
                time.sleep(delay)
                continue
            break
        resp.raise_for_status()
        with open(dest, "wb") as fh:
            for chunk in resp.iter_content(chunk_size=_CHUNK):
                if chunk:
                    fh.write(chunk)

    def _run_ytdlp(self, url: str, dest: Path, headers: dict) -> Path | None:
        """Invoke yt-dlp to fetch a Mux/HLS stream to (around) `dest`.

        Returns the actual output path (yt-dlp may remux to a different
        container, so we resolve the real file afterward). Overridden/
        monkeypatched in tests to avoid spawning a real subprocess.

        The output template uses `dest` without its extension; yt-dlp appends
        the chosen container extension. We pass Referer/Origin (Mux JWT policy)
        and the cookies file.
        """
        dest = Path(dest)
        out_template = str(dest.with_suffix("")) + ".%(ext)s"
        cmd = ["yt-dlp", "--no-progress", "-o", out_template]
        for key, value in headers.items():
            cmd += ["--add-header", f"{key}:{value}"]
        if self.cookies_path and os.path.isfile(self.cookies_path):
            cmd += ["--cookies", self.cookies_path]
        cmd.append(url)
        try:
            subprocess.run(
                cmd,
                check=True,
                capture_output=True,
                text=True,
                timeout=_TIMEOUT_SECONDS,
            )
        except (OSError, subprocess.SubprocessError) as exc:
            log.warning("yt-dlp failed for %s: %s", url, exc)
            return None
        return self._existing_video_output(dest)

    def _existing_video_output(self, dest: Path) -> Path | None:
        """Find a yt-dlp output for `dest` regardless of chosen extension.

        Returns `dest` itself if present, else any sibling sharing the same
        stem (the remuxed container). None if nothing matched.
        """
        dest = Path(dest)
        if dest.exists():
            return dest
        stem = dest.with_suffix("").name
        parent = dest.parent
        if not parent.is_dir():
            return None
        for cand in sorted(parent.iterdir()):
            if cand.is_file() and cand.stem == stem and cand.name != dest.name:
                return cand
        return None

    # -- validation --------------------------------------------------------

    def _validate_path(self, path: Path, artist_slug: str) -> str | None:
        """Validate a freshly-written file; quarantine + return reason if bad.

        Mirrors gallery_dl._validate_and_quarantine: fail-open for unknown
        formats, move the corrupt file to _quarantine/<slug>/patreon, and return
        the failure reason string (None when ok / not validatable / disabled).
        """
        if not self._validate or not is_validatable(path):
            return None
        try:
            result = validate_file(path)
        except Exception as exc:
            log.warning("Validator raised on %s: %s", path, exc)
            return None
        if result.ok:
            return None
        quarantine_root = self.images_root / "_quarantine" / artist_slug / "patreon"
        try:
            quarantine_root.mkdir(parents=True, exist_ok=True)
            dest = quarantine_root / path.name
            counter = 1
            while dest.exists():
                dest = quarantine_root / f"{path.stem}.{counter}{path.suffix}"
                counter += 1
            shutil.move(str(path), str(dest))
        except OSError as exc:
            log.error("Failed to quarantine %s: %s. File left in place.", path, exc)
        return result.reason or "validation failed"

    # -- sidecar -----------------------------------------------------------

    def _write_sidecar(self, post: dict, media_path: Path) -> Path:
        """Write the importer-consumed sidecar next to `media_path`.

        Patreon uses base-default sidecar keys (parse_sidecar maps
        category->platform, id->external_post_id, title->post_title,
        content->description, published_at->post_date, url->post_url). Patreon
        registers no derive_post_url, so `url` is trusted as the permalink — we
        pass the post's attributes.url.
        """
        attrs = post.get("attributes") or {}
        title = attrs.get("title")
        content = attrs.get("content")
        published = attrs.get("published_at")
        url = attrs.get("url")
        data = {
            "category": "patreon",
            "id": str(post.get("id") or ""),
            "title": title if isinstance(title, str) else "",
            "content": content if isinstance(content, str) else "",
            "published_at": published if isinstance(published, str) else None,
            "url": url if isinstance(url, str) else None,
        }
        sidecar_path = media_path.with_suffix(".json")
        sidecar_path.write_text(json.dumps(data, indent=2))
        return sidecar_path