bvandeusen/FabledCurator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 33 Wiki Activity

feat(patreon): native media downloader — ingester build step 2b (plan #697)
CI / lint (push) Successful in 2s
Details
CI / backend-lint-and-test (push) Successful in 11s
Details
CI / frontend-build (push) Successful in 27s
Details
CI / integration (push) Successful in 2m58s
Details

Browse Source 
PatreonDownloader.download_post: writes resolved MediaItems to gallery-dl's
exact on-disk layout (<slug>/patreon/<date>_<id>_<title40>/<NN>_<file>) +
a sidecar the importer's find_sidecar/parse_sidecar consume unchanged. Two-tier
skip (injected seen predicate, then disk). Streamed GET (.part→rename) +
file_validator quarantine; Mux/m3u8 video shells out to yt-dlp with Patreon
Referer/Origin. Pure (no DB) — ledger + orchestration land in step 3. Unit
tests stub the session + yt-dlp seams (no network/subprocess).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-06-05 19:39:12 -04:00
parent 6222928746
commit 2ec7d86a3b
2 changed files with 636 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/patreon_downloader.py
+384
View File
@@ -0,0 +1,384 @@
"""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
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
log = logging.getLogger(__name__)
_TITLE_MAX = 40
_TIMEOUT_SECONDS = 120.0
_CHUNK = 1 << 16
# Same Referer/Origin gallery-dl forwards to yt-dlp for Mux-hosted Patreon
# video (gallery_dl.py _get_default_config -> downloader.ytdl.raw-options).
# Mux's JWT playback policy checks Referer/Origin on every request, so yt-dlp
# must send Patreon's, not its own default.
_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,
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
# 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)
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."""
resp = self.session.get(url, stream=True, timeout=_TIMEOUT_SECONDS)
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
tests/test_patreon_downloader.py
+252
View File
@@ -0,0 +1,252 @@
"""Unit tests for PatreonDownloader (build step 2b).
No network, no real subprocess. The HTTP layer is stubbed via the `session`
seam (a fake session whose `.get` returns a fake streaming response) and yt-dlp
via monkeypatching `_run_ytdlp`. PURE module → no DB, so unmarked (fast lane).
"""
from __future__ import annotations
from pathlib import Path
import pytest
from backend.app.services.patreon_client import MediaItem
from backend.app.services.patreon_downloader import (
MediaOutcome,
PatreonDownloader,
_sanitize,
)
from backend.app.utils.sidecar import find_sidecar
# Minimal valid PNG so the file_validator passes on the happy path.
_PNG_HEAD = b"\x89PNG\r\n\x1a\n"
_PNG_TAIL = b"\x00\x00\x00\x00IEND\xaeB`\x82"
_PNG_BYTES = _PNG_HEAD + (b"\x00" * 16) + _PNG_TAIL
class _FakeResponse:
def __init__(self, payload: bytes):
self._payload = payload
def raise_for_status(self):
return None
def iter_content(self, chunk_size=65536):
for i in range(0, len(self._payload), chunk_size):
yield self._payload[i : i + chunk_size]
class _FakeSession:
"""Records GETs and returns canned bytes (per-URL, default _PNG_BYTES)."""
def __init__(self, payloads: dict[str, bytes] | None = None):
self.payloads = payloads or {}
self.calls: list[str] = []
def get(self, url, stream=False, timeout=None):
self.calls.append(url)
return _FakeResponse(self.payloads.get(url, _PNG_BYTES))
def _post(post_id="1001", title="My Post Title", published="2026-05-01T12:00:00Z"):
return {
"id": post_id,
"attributes": {
"title": title,
"content": "<p>body</p>",
"published_at": published,
"url": f"https://www.patreon.com/posts/{post_id}",
},
}
def _img(name, post_id="1001", url=None):
return MediaItem(
url=url or f"https://cdn.patreon.com/{name}",
filename=name,
kind="images",
filehash=None,
post_id=post_id,
)
def _downloader(tmp_path: Path, session=None, validate=True):
return PatreonDownloader(
images_root=tmp_path,
cookies_path=None,
validate=validate,
session=session or _FakeSession(),
)
def test_path_layout_two_images(tmp_path):
dl = _downloader(tmp_path)
items = [_img("media1.png"), _img("media2.png")]
outcomes = dl.download_post(_post(), items, "artist-x")
post_dir = tmp_path / "artist-x" / "patreon" / "2026-05-01_1001_My Post Title"
p1 = post_dir / "01_media1.png"
p2 = post_dir / "02_media2.png"
assert p1.is_file()
assert p2.is_file()
assert [o.status for o in outcomes] == ["downloaded", "downloaded"]
assert outcomes[0].path == p1
assert p1.read_bytes() == _PNG_BYTES
def test_dir_sanitized(tmp_path):
dl = _downloader(tmp_path)
post = _post(title='bad/name:with*chars?')
dl.download_post(post, [_img("a.png")], "artist-x")
patreon_dir = tmp_path / "artist-x" / "patreon"
children = list(patreon_dir.iterdir())
assert len(children) == 1
name = children[0].name
assert "/" not in name
assert not any(c in name for c in '<>:"\\|?*')
def test_no_date_prefix_when_unparseable(tmp_path):
dl = _downloader(tmp_path)
post = _post(published="not-a-date")
dl.download_post(post, [_img("a.png")], "artist-x")
patreon_dir = tmp_path / "artist-x" / "patreon"
children = [c.name for c in patreon_dir.iterdir()]
assert children == ["1001_My Post Title"]
def test_sidecar_written_and_findable(tmp_path):
dl = _downloader(tmp_path)
outcomes = dl.download_post(_post(), [_img("media1.png")], "artist-x")
media_path = outcomes[0].path
sidecar = find_sidecar(media_path)
assert sidecar is not None
assert sidecar.name == "01_media1.json"
import json
data = json.loads(sidecar.read_text())
assert data["category"] == "patreon"
assert data["id"] == "1001"
assert data["title"] == "My Post Title"
assert data["url"] == "https://www.patreon.com/posts/1001"
def test_skip_seen(tmp_path):
session = _FakeSession()
dl = _downloader(tmp_path, session=session)
outcomes = dl.download_post(
_post(), [_img("media1.png")], "artist-x", is_seen=lambda m: True
)
assert outcomes[0].status == "skipped_seen"
assert outcomes[0].path is None
assert session.calls == []
# No file written anywhere.
assert not (tmp_path / "artist-x").exists()
def test_skip_disk(tmp_path):
dl = _downloader(tmp_path)
post_dir = tmp_path / "artist-x" / "patreon" / "2026-05-01_1001_My Post Title"
post_dir.mkdir(parents=True)
existing = post_dir / "01_media1.png"
existing.write_bytes(b"already here")
outcomes = dl.download_post(_post(), [_img("media1.png")], "artist-x")
assert outcomes[0].status == "skipped_disk"
assert outcomes[0].path == existing
assert existing.read_bytes() == b"already here" # untouched
def test_video_routes_to_ytdlp(tmp_path, monkeypatch):
session = _FakeSession()
dl = _downloader(tmp_path, session=session)
captured = {}
def fake_ytdlp(url, dest, headers):
captured["url"] = url
captured["headers"] = headers
out = Path(dest).with_suffix(".mp4")
out.parent.mkdir(parents=True, exist_ok=True)
out.write_bytes(b"video-bytes")
return out
monkeypatch.setattr(dl, "_run_ytdlp", fake_ytdlp)
video = MediaItem(
url="https://stream.mux.com/abc123.m3u8",
filename="clip.mp4",
kind="postfile",
filehash=None,
post_id="1001",
)
outcomes = dl.download_post(_post(), [video], "artist-x")
assert captured["url"] == "https://stream.mux.com/abc123.m3u8"
assert captured["headers"]["Referer"] == "https://www.patreon.com/"
assert captured["headers"]["Origin"] == "https://www.patreon.com"
assert session.calls == [] # NOT the plain-GET path
assert outcomes[0].status == "downloaded"
assert outcomes[0].path.read_bytes() == b"video-bytes"
# Sidecar written next to the actual (remuxed) output.
assert find_sidecar(outcomes[0].path) is not None
def test_one_failure_isolated(tmp_path):
class _BoomSession(_FakeSession):
def get(self, url, stream=False, timeout=None):
self.calls.append(url)
if "bad" in url:
raise RuntimeError("network boom")
return _FakeResponse(_PNG_BYTES)
dl = _downloader(tmp_path, session=_BoomSession())
items = [
_img("good.png", url="https://cdn.patreon.com/good.png"),
_img("bad.png", url="https://cdn.patreon.com/bad.png"),
]
outcomes = dl.download_post(_post(), items, "artist-x")
assert outcomes[0].status == "downloaded"
assert outcomes[1].status == "error"
assert "boom" in outcomes[1].error
def test_validation_quarantines_corrupt(tmp_path):
# A .png whose bytes are not a valid PNG → validator fails → error +
# quarantine move; with validate=False it would pass.
bad = b"not a real png"
session = _FakeSession({"https://cdn.patreon.com/corrupt.png": bad})
dl = _downloader(tmp_path, session=session, validate=True)
item = _img("corrupt.png", url="https://cdn.patreon.com/corrupt.png")
outcomes = dl.download_post(_post(), [item], "artist-x")
assert outcomes[0].status == "error"
post_dir = tmp_path / "artist-x" / "patreon" / "2026-05-01_1001_My Post Title"
assert not (post_dir / "01_corrupt.png").exists()
quarantine = tmp_path / "_quarantine" / "artist-x" / "patreon"
assert quarantine.is_dir()
assert any(quarantine.iterdir())
def test_validation_off_keeps_bytes(tmp_path):
bad = b"not a real png"
session = _FakeSession({"https://cdn.patreon.com/x.png": bad})
dl = _downloader(tmp_path, session=session, validate=False)
item = _img("x.png", url="https://cdn.patreon.com/x.png")
outcomes = dl.download_post(_post(), [item], "artist-x")
assert outcomes[0].status == "downloaded"
assert outcomes[0].path.read_bytes() == bad
def test_sanitize_helper():
assert _sanitize('a/b') == "a_b"
assert _sanitize('x<>:"|?*y') == "x_______y"
assert _sanitize("trail. ") == "trail"
assert _sanitize("") == "_"
def test_media_outcome_shape():
o = MediaOutcome(media=None, status="downloaded", path=Path("/tmp/x"), error=None)
assert o.status == "downloaded"
assert o.path == Path("/tmp/x")