"""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): //patreon//_ where = "__" (date prefix omitted when the post's published_at is unparseable), is the 1-based index of the item in the post zero-padded to 2, and is MediaItem.filename. The sidecar is written next to the media as _.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 json import logging import os import subprocess import time from collections.abc import Callable from pathlib import Path from urllib.parse import urlsplit import requests from ..utils.prosemirror import post_body_html from .native_ingest_common import ( _BACKOFF_CAP_SECONDS, _MAX_MEDIA_RETRIES, BaseNativeDownloader, MediaOutcome, PostRecordOutcome, post_dir_name, sanitize_segment, ) log = logging.getLogger(__name__) # yt-dlp subprocess wall-clock per attempt (video only; the shared HTTP fetch # budgets live in BaseNativeDownloader). _TIMEOUT_SECONDS = 120.0 # 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", } 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") class PatreonDownloader(BaseNativeDownloader): """Download resolved Patreon media to gallery-dl's on-disk layout. Subclasses BaseNativeDownloader for the shared streaming GET (transient-retry + Range-resume) and validation/quarantine; adds the Mux/HLS yt-dlp video branch and the detail-fetch body enrichment. PURE: no DB. `_run_ytdlp` is monkeypatchable and the HTTP session is the injectable `session=` seam. """ def __init__( self, images_root: Path, cookies_path: str | None = None, *, validate: bool = True, rate_limit: float = 0.0, session: requests.Session | None = None, content_fetcher: Callable[[str], str | None] | None = None, ): super().__init__( images_root, cookies_path, platform="patreon", validate=validate, rate_limit=rate_limit, session=session, ) # Best-effort enrichment seam: (post_id) -> full HTML body, or None. The # feed endpoint often omits `content`; the adapter wires this to # PatreonClient.fetch_post_detail_content so the sidecar captures the # real body (formatting + inline + external links). # None in unit tests / when enrichment isn't wanted. self._content_fetcher = content_fetcher # -- public ------------------------------------------------------------ def download_post( self, post: dict, media_items: list, artist_slug: str, *, is_seen: Callable[[object], bool] = lambda m: False, should_stop: Callable[[], bool] = lambda: False, recapture: bool = 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. `should_stop()` is polled BEFORE each media item: a media-dense post can otherwise run a backfill chunk far past its time-box (the engine only re-checks the budget between posts), so we honour the deadline mid-post and return the items done so far — the rest re-fetch next chunk (they were never marked seen). Bounds chunk overrun to one media download. `recapture` (#830): don't re-download already-present media, but DO surface on-disk media as `skipped_disk` (with its path) even when the seen-ledger would tier-1 skip it — so the engine can backfill source_filehash for inline-image localization. Genuinely-missing seen media is NOT refetched. """ post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post) outcomes: list[MediaOutcome] = [] for i, media in enumerate(media_items, start=1): if should_stop(): break try: outcomes.append( self._download_one( post, media, post_dir, artist_slug, i, is_seen, recapture=recapture, ) ) 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], *, recapture: bool = False, ) -> MediaOutcome: # tier-1: seen ledger (injected; no DB here). In recapture mode we DON'T # short-circuit here — we fall through to the disk check so an on-disk # seen file is surfaced as skipped_disk (with its path) for source_filehash # backfill; a seen file that's NOT on disk is left alone (not refetched). seen = is_seen(media) if seen and not recapture: return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) nn = f"{index:02d}" final_name = sanitize_segment(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 ) # recapture: a seen item that isn't on disk is NOT re-downloaded (that's # recovery's job) — recapture only re-grabs post text + localizes existing # files. Returns skipped_seen so the run-of-seen / counts stay consistent. if seen: return MediaOutcome(media=media, status="skipped_seen", path=None, 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) reason, quarantine_dest = self._validate_path( out_path, artist_slug, media.url ) if reason is not None: # Quarantined (corrupt/invalid) — distinct from a download error # so the run can report a real files_quarantined count + paths. return MediaOutcome( media=media, status="quarantined", path=quarantine_dest, error=reason, ) self._write_sidecar(post, out_path, source_url=media.url) return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) # -- video (Mux/HLS via yt-dlp) ---------------------------------------- # The plain-GET streaming path (_fetch_get / _fetch_to_file) and # _validate_path are inherited from BaseNativeDownloader. 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. Mirrors the GET path's transient/permanent split (plan #705 #8): a hung fetch (TimeoutExpired) or a spawn failure (OSError) is TRANSIENT — back off and retry. A non-zero yt-dlp exit (CalledProcessError) is treated as PERMANENT for this pass — yt-dlp already does its OWN internal network retries, so a non-zero exit is effectively a real failure (private/gone/ geo-blocked), like a 4xx on the GET path: fail fast to the per-item error → dead-letter path. """ 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) attempt = 0 while True: try: subprocess.run( cmd, check=True, capture_output=True, text=True, timeout=_TIMEOUT_SECONDS, ) break except subprocess.CalledProcessError as exc: # Permanent for this pass — fail fast (no retry). log.warning( "yt-dlp failed (exit %s) for %s: %s", exc.returncode, url, (exc.stderr or "").strip() or exc, ) return None except (OSError, subprocess.TimeoutExpired) as exc: # Transient — back off and retry, like a transport blip on a GET. if attempt >= _MAX_MEDIA_RETRIES: log.warning( "yt-dlp transient failure exhausted for %s: %s", url, exc ) return None attempt += 1 delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS) log.warning( "yt-dlp transient failure (%s) — backing off %.1fs " "(retry %d/%d): %s", url, delay, attempt, _MAX_MEDIA_RETRIES, exc, ) time.sleep(delay) 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 # -- sidecar ----------------------------------------------------------- def _write_sidecar( self, post: dict, media_path: Path, *, source_url: str | None = None ) -> Path: """Write the per-media sidecar next to `media_path` — post-first (#856). On the native ingester the POST-RECORD (`write_post_record` → `_post.json`) is the sole writer of the post body/links/metadata, captured once per post BEFORE its media in the walk. So the per-media sidecar carries ONLY image-specific identity: `category` (platform) + `id` (external_post_id, to link provenance to the right Post) + this file's `source_url` (its CDN URL, #830 Phase 2 — the importer persists its filehash so the body's inline `` remaps to the local copy at render time). No body: writing it next to every image duplicated the post body N+1× and risked divergence (milestone #67). The importer skips post fields for these (post_first). """ return self._write_sidecar_data( post, media_path.with_suffix(".json"), source_url=source_url, minimal=True, ) def _write_sidecar_data( self, post: dict, sidecar_path: Path, *, source_url: str | None = None, minimal: bool = False, ) -> Path: """Serialize the post's metadata to `sidecar_path`. The post-only record (`write_post_record`) writes the FULL post (body/title/date/url); the per-media sidecar (`_write_sidecar`, minimal=True) writes only image identity (category/id/source_url) — post-first (#856). `source_url` is set only for the per-media sidecar — a media-less post has no source file.""" if minimal: data = {"category": "patreon", "id": str(post.get("id") or "")} if source_url: data["source_url"] = source_url sidecar_path.write_text(json.dumps(data, indent=2)) return sidecar_path attrs = post.get("attributes") or {} title = attrs.get("title") # Resolve the body HTML from the feed attrs: legacy flat `content`, else # convert the current `content_json_string` ProseMirror doc (#842). content = post_body_html(attrs) # The feed/list endpoint frequently returns an empty body; the full body # only comes from the per-post detail endpoint. Enrich on first write for # this post and MEMOIZE the RESOLVED HTML by mutating the shared `post` # dict — so a multi-image post fetches detail at most once, the post-record # body-length read reuses it, and a fully-seen post (no fresh download → no # sidecar write) never pays the extra GET. if (not content or not content.strip()) and self._content_fetcher: fetched = self._content_fetcher(str(post.get("id") or "")) if fetched: content = fetched if isinstance(content, str) and content.strip(): attrs["content"] = content post["attributes"] = attrs 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, } if source_url: data["source_url"] = source_url sidecar_path.write_text(json.dumps(data, indent=2)) return sidecar_path def write_post_record(self, post: dict, artist_slug: str) -> PostRecordOutcome: """Write a post-ONLY sidecar (no media file) for a media-less post, so the importer can still upsert the Post + its body — text posts often hold the only copy of an external link. Named `_post.json`: the leading underscore keeps it from colliding with a media sidecar (`_.json`) and from being resolved as some media file's sidecar by find_sidecar. Returns a PostRecordOutcome (path None when the post has no id) carrying the captured body's shape — post_type + final char count — so the engine can log per-post handling without re-reading the post itself. """ attrs = post.get("attributes") or {} title = attrs.get("title") if isinstance(attrs.get("title"), str) else None post_type = attrs.get("post_type") if isinstance(attrs.get("post_type"), str) else None pid = str(post.get("id") or "") if not pid: return PostRecordOutcome( path=None, post_type=post_type, title=title, body_chars=0, ) post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post) post_dir.mkdir(parents=True, exist_ok=True) path = self._write_sidecar_data(post, post_dir / "_post.json") # _write_sidecar_data has by now memoized any detail-fetched body onto # post["attributes"]["content"], so re-read it for the FINAL char count. body = (post.get("attributes") or {}).get("content") body_chars = len(body) if isinstance(body, str) else 0 return PostRecordOutcome( path=path, post_type=post_type, title=title, body_chars=body_chars, )