Files
FabledCurator/backend/app/services/patreon_downloader.py
T
bvandeusen 976107bbe8
CI / lint (push) Successful in 4s
CI / frontend-build (push) Successful in 24s
CI / backend-lint-and-test (push) Successful in 35s
CI / integration (push) Successful in 3m13s
fix(patreon): read post body from content_json_string (ProseMirror), not the dead content field (#842)
THE empty-body root cause. Patreon deprecated the flat `content` HTML field —
it returns null on the feed AND the detail endpoint, for every post type
(confirmed against the live API: all 135 StickySpoodge posts, text_only/
image_file/poll alike). The real body now lives in `content_json_string` (a
ProseMirror/TipTap doc), returned only under the DEFAULT post fieldset — a sparse
fields[post]=content request omits it. Not credential, not post_type: a request
shape gone stale.

- NEW utils/prosemirror.py: ProseMirror doc -> HTML (paragraphs, marks
  bold/italic/underline/strike/code/link, hardBreak, inline images, lists,
  headings; unknown nodes degrade to children). post_body_html(attrs) = the one
  resolver: legacy content HTML else convert content_json_string.
- patreon_client: add content_json_string to the feed _FIELDS_POST; rewrite
  fetch_post_detail_content to use the DEFAULT fieldset (no sparse fields[post])
  and resolve via post_body_html (replaces the wrong sparse req + full-fetch
  fallback).
- patreon_downloader._write_sidecar_data: resolve body via post_body_html
  (feed content_json_string) before the detail-fetch; memoize resolved HTML.
- tests: prosemirror converter unit tests; client legacy + content_json_string
  paths; contract pins content_json_string.

Inline <img> nodes carry the CDN filehash → bodies now feed Phase-2 localization.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-15 00:14:52 -04:00

618 lines
27 KiB
Python

"""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 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 ..utils.prosemirror import post_body_html
from .file_validator import is_validatable, quarantine_file, validate_file
from .patreon_client import (
_BACKOFF_CAP_SECONDS,
_load_session,
_retry_after_seconds,
)
log = logging.getLogger(__name__)
_TITLE_MAX = 40
_TIMEOUT_SECONDS = 120.0
_CHUNK = 1 << 16
# Retry a media GET that hits a TRANSIENT failure within the same pass (plan
# #705 #8): a transport blip (connection reset / timeout / truncated stream), a
# 429, or a 5xx. PERMANENT failures (404 gone, 403 forbidden) fail fast straight
# to the error/dead-letter path — no point re-fetching them. Keeps a momentary
# network hiccup from becoming a per-item error that waits for the next walk.
_MAX_MEDIA_RETRIES = 3
# requests transport errors worth retrying (vs. an HTTPError, which is a real
# server response and is classified by status code).
_TRANSIENT_TRANSPORT_EXC = (
requests.ConnectionError,
requests.Timeout,
requests.exceptions.ChunkedEncodingError,
)
# 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",
"quarantined", "error". `path` is the final on-disk path for "downloaded"
(the actual yt-dlp output for video), the path that already existed for
"skipped_disk", or the _quarantine destination for "quarantined"; None for
"skipped_seen" and (usually) "error". `error` carries the failure/validation
reason for "error"/"quarantined", else None.
"""
media: object # MediaItem (avoid importing the name for a bare annotation)
status: str
path: Path | None
error: str | None
@dataclass
class PostRecordOutcome:
"""Result of write_post_record — mirrors the download_post → MediaOutcome
contract so the engine reports per-post handling without re-reading the post.
`path` is the _post.json sidecar (None when the post had no id); the rest is
the captured body's shape (post_type + final char count) for the run log.
"""
path: Path | None
post_type: str | None
title: str | None
body_chars: int
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,
content_fetcher: Callable[[str], str | None] | None = None,
):
self.images_root = Path(images_root)
self.cookies_path = str(cookies_path) if cookies_path else None
self._validate = validate
# 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 <img> + external <a href> links).
# None in unit tests / when enrichment isn't wanted.
self._content_fetcher = content_fetcher
# 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,
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(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)
# -- 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, retrying
TRANSIENT failures within the same pass (plan #705 #8) and RESUMING from
the bytes already on disk via a Range request when a retry follows a
mid-download cut (plan #708 B5).
Retried (backoff): transport blips (connection reset / timeout /
truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After),
and 5xx. Failed fast (no retry → HTTPError → per-item error → dead-letter
path): 4xx other than 429 (404 gone, 403 forbidden) — re-fetching a
permanent failure is pointless.
Resume: on a retry, if bytes already landed in `dest`, ask for the rest
with `Range: bytes=<have>-`. A 206 means the server honored it → append; a
200 means it ignored it (served the whole file) → start clean. The caller
(_fetch_get) stages into a `.part`, so a non-range server never corrupts
the output — the worst case is re-downloading from zero, as before.
"""
attempt = 0
while True:
have = dest.stat().st_size if dest.exists() else 0
headers = {"Range": f"bytes={have}-"} if have > 0 else None
try:
resp = self.session.get(
url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers,
)
if (resp.status_code == 429 or resp.status_code >= 500) \
and attempt < _MAX_MEDIA_RETRIES:
attempt += 1
delay = _retry_after_seconds(resp, attempt)
log.warning(
"Patreon media transient HTTP %d (%s) — backing off "
"%.1fs (retry %d/%d)",
resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES,
)
time.sleep(delay)
continue
# A Range that starts at/past EOF (we already have the whole file)
# comes back 416 — the bytes we kept ARE the file.
if have > 0 and resp.status_code == 416:
return
# 2xx → ok; 4xx-non-429 (or an exhausted 429/5xx) → HTTPError
# (permanent for this pass) → not caught below → per-item error.
resp.raise_for_status()
# 206 → server honored the Range; append after the kept bytes.
# Anything else (200) → it served the whole file → start clean.
mode = "ab" if (have > 0 and resp.status_code == 206) else "wb"
with open(dest, mode) as fh:
for chunk in resp.iter_content(chunk_size=_CHUNK):
if chunk:
fh.write(chunk)
return
except _TRANSIENT_TRANSPORT_EXC as exc:
if attempt >= _MAX_MEDIA_RETRIES:
raise # exhausted → terminal error outcome
attempt += 1
delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS)
log.warning(
"Patreon media transport error (%s) — backing off %.1fs "
"(retry %d/%d): %s",
url, delay, attempt, _MAX_MEDIA_RETRIES, exc,
)
time.sleep(delay)
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
# -- validation --------------------------------------------------------
def _validate_path(
self, path: Path, artist_slug: str, source_url: str | None = None
) -> tuple[str | None, Path | None]:
"""Validate a freshly-written file; quarantine if bad.
Uses the shared `file_validator.quarantine_file` — same move + provenance
sidecar gallery-dl writes (the native path used to skip the sidecar; that
parity gap is closed here). Returns `(reason, quarantine_dest)` when
quarantined (dest is the original path if the move itself failed), else
`(None, None)` (ok / not validatable / disabled). plan #704: the dest is
surfaced so the run reports a real quarantined-paths list.
"""
if not self._validate or not is_validatable(path):
return None, None
try:
result = validate_file(path)
except Exception as exc:
log.warning("Validator raised on %s: %s", path, exc)
return None, None
if result.ok:
return None, None
dest = quarantine_file(
self.images_root, path, artist_slug, "patreon",
url=source_url, result=result,
)
return (result.reason or "validation failed"), (dest or path)
# -- sidecar -----------------------------------------------------------
def _write_sidecar(
self, post: dict, media_path: Path, *, source_url: str | None = None
) -> 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.
`source_url` is THIS file's CDN URL (#830 Phase 2). The importer persists
its filehash on the ImageRecord so the post body's inline `<img src>` can
be remapped to the local copy at render time.
"""
return self._write_sidecar_data(
post, media_path.with_suffix(".json"), source_url=source_url
)
def _write_sidecar_data(
self, post: dict, sidecar_path: Path, *, source_url: str | None = None
) -> Path:
"""Serialize the post's metadata to `sidecar_path`. Shared by the
per-media sidecar (next to each file) and the post-only sidecar
(`write_post_record`, for media-less posts). `source_url` is set only for
the per-media sidecar — a media-less post has no source file."""
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 <a href> link. Named `_post.json`: the
leading underscore keeps it from colliding with a media sidecar
(`<NN>_<stem>.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,
)