bvandeusen
682beafbc5
Typed, loud failure mapping for the native Patreon ingester so a changed
API shape or expired auth never silently zero-downloads as "success".
- New ErrorType.API_DRIFT (free varchar error_type col → no migration):
distinct from auth so the operator knows the fix is updating the
ingester, not rotating cookies.
- patreon_client: PatreonAPIError carries status_code; new PatreonAuthError
for 401/403 + HTML-login/non-JSON bodies (reclassified from drift —
expired-session is auth, actionable as "rotate cookies").
- patreon_ingester._failure_result maps: PatreonAuthError→AUTH_ERROR,
PatreonDriftError→API_DRIFT ("Patreon API changed — ingester needs
update"), HTTP 429→RATE_LIMITED, 404→NOT_FOUND, other HTTP→HTTP_ERROR,
transport→NETWORK_ERROR. (429 thus drives the platform cooldown.)
- FailingSourcesCard: api_drift chip (red) + hint.
Contract test (new test_patreon_contract.py): the recorded /api/posts
fixture must parse end-to-end (no drift, 5 media across 4 posts) AND the
request params must still carry every field the parser depends on
(file_name, image_urls/download_url, images/attachments_media/media
includes, content/post_file/image post fields) — a trim of either trips a
red build. Plus client HTTP-status classification tests and ingester
error-type mapping tests.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
380 lines
16 KiB
Python
380 lines
16 KiB
Python
"""Native Patreon ingester — phase-2 orchestrator (build step 3).
|
|
|
|
Ties build steps 1 (`patreon_client`) and 2 (`patreon_downloader` +
|
|
`patreon_seen_media`) together into a single sync walk that REPLACES the
|
|
gallery-dl subprocess for Patreon. `download_service.download_source` calls
|
|
`PatreonIngester.run(...)` from phase 2 (in a thread, via run_in_executor, since
|
|
everything here is sync `requests`/`subprocess`) and gets back a
|
|
`DownloadResult` — the exact shape gallery-dl returns — so phase 1 (DB setup)
|
|
and phase 3 (import → pHash dedup → thumbnails → ML) are untouched and cannot
|
|
tell the difference.
|
|
|
|
Three modes (selected by `download_service` from `config_overrides` state):
|
|
- tick — newest→oldest, skip seen (tier-1 ledger + tier-2 disk), early-out
|
|
after N contiguous already-have-it items (the cheap native
|
|
equivalent of gallery-dl's `exit:20`, now free of per-file HEADs).
|
|
- backfill — full-history walk in a time-boxed chunk, resuming from the
|
|
pagination cursor checkpoint; reaches the bottom → "complete".
|
|
- recovery — like backfill but BYPASSES the tier-1 seen-ledger, so
|
|
deliberately-dropped-and-deleted near-dups get re-fetched and
|
|
re-evaluated under the current pHash threshold (tier-2 disk skip
|
|
still spares files we kept). Triggered by the same #693 backfill
|
|
state machine plus the `_backfill_bypass_seen` flag, so the whole
|
|
cursor/chunk/complete/stall lifecycle is reused verbatim.
|
|
|
|
Cursor contract: the ingester emits gallery-dl-style ``Cursor: <token>`` lines
|
|
into the returned `stdout`, one per page it walks. That is exactly what
|
|
`download_service`'s existing backfill lifecycle reads via
|
|
`parse_last_cursor(stdout, stderr)` — so checkpointing, the TIMEOUT→PARTIAL
|
|
reclassification, and completion detection all keep working unchanged.
|
|
|
|
The seen-ledger lives in Postgres (`patreon_seen_media`). The ingester opens a
|
|
SHORT-LIVED sync session per page batch (via an injected sessionmaker) — never
|
|
held across a network fetch — so the multi-minute walk can't strand a checked-out
|
|
connection for the server to reap ([[db-connection-held-across-subprocess]]).
|
|
|
|
FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
|
|
from ..models import PatreonSeenMedia
|
|
from .gallery_dl import DownloadResult, ErrorType
|
|
from .patreon_client import (
|
|
MediaItem,
|
|
PatreonAPIError,
|
|
PatreonAuthError,
|
|
PatreonClient,
|
|
PatreonDriftError,
|
|
)
|
|
from .patreon_downloader import PatreonDownloader
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# gallery-dl's `exit:20` default ported over: stop a tick after this many
|
|
# CONTIGUOUS already-have-it media (seen-ledger or on-disk). Native walks have
|
|
# zero per-file HEADs, so the only cost of a higher number is a few extra cheap
|
|
# ledger lookups — 20 is operator-set headroom against paywalled/undownloadable
|
|
# items interleaving with archived ones.
|
|
_TICK_SEEN_THRESHOLD = 20
|
|
|
|
# Ledger keys are stored in patreon_seen_media.filehash VARCHAR(128); bound any
|
|
# synthesized key so a pathologically long file_name can't overflow the column.
|
|
_LEDGER_KEY_MAX = 128
|
|
|
|
|
|
def _ledger_key(media: MediaItem) -> str:
|
|
"""Stable per-media identity for the cross-run seen-ledger.
|
|
|
|
A Patreon CDN URL carries a 32-char MD5 (`media.filehash`) — that is the
|
|
natural key. Some media have none: Mux/HLS video (`stream.mux.com`, no
|
|
content hash at discovery) and the odd inline-content `<img>` pointing at a
|
|
hashless URL. The plan calls the video case the ``video:<post_id>:<media_id>``
|
|
sentinel; `MediaItem` carries no media_id, so the post-scoped filename is the
|
|
stable proxy. Bounded to the column width.
|
|
"""
|
|
if media.filehash:
|
|
return media.filehash
|
|
return f"{media.post_id}:{media.filename}"[:_LEDGER_KEY_MAX]
|
|
|
|
|
|
class PatreonIngester:
|
|
"""Walk a Patreon campaign's posts, download unseen media, return a
|
|
`DownloadResult`.
|
|
|
|
Construct with the per-source `cookies_path` and a sync sessionmaker for the
|
|
seen-ledger. `client` / `downloader` are injectable seams so unit tests run
|
|
without network, subprocess, or a real CDN.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
images_root: Path,
|
|
cookies_path: str | None,
|
|
session_factory: Callable[[], object],
|
|
*,
|
|
validate: bool = True,
|
|
client: PatreonClient | None = None,
|
|
downloader: PatreonDownloader | None = None,
|
|
):
|
|
self.images_root = Path(images_root)
|
|
self.cookies_path = str(cookies_path) if cookies_path else None
|
|
self.session_factory = session_factory
|
|
self.client = client if client is not None else PatreonClient(cookies_path)
|
|
self.downloader = (
|
|
downloader
|
|
if downloader is not None
|
|
else PatreonDownloader(
|
|
self.images_root, cookies_path, validate=validate
|
|
)
|
|
)
|
|
|
|
# -- public ------------------------------------------------------------
|
|
|
|
def run(
|
|
self,
|
|
*,
|
|
source_id: int,
|
|
campaign_id: str,
|
|
artist_slug: str,
|
|
url: str,
|
|
mode: str,
|
|
resume_cursor: str | None = None,
|
|
time_budget_seconds: float = 870.0,
|
|
seen_threshold: int = _TICK_SEEN_THRESHOLD,
|
|
) -> DownloadResult:
|
|
"""Walk + download for one source, returning a gallery-dl-shaped result.
|
|
|
|
`mode` is "tick" | "backfill" | "recovery". Recovery bypasses the tier-1
|
|
seen-ledger (tier-2 disk still skips kept files). The walk stops on:
|
|
- budget exhaustion (time_budget_seconds) → TIMEOUT / PARTIAL
|
|
- tick early-out (seen_threshold contiguous seen) → success
|
|
- reaching the bottom of the feed → success (rc 0)
|
|
A client-level failure (drift / auth / network) fails the whole run loud.
|
|
"""
|
|
bypass_seen = mode == "recovery"
|
|
start = time.monotonic()
|
|
log_lines: list[str] = []
|
|
written: list[str] = []
|
|
downloaded = 0
|
|
errors = 0
|
|
consecutive_seen = 0
|
|
emitted_cursor: str | None = None
|
|
reached_bottom = False
|
|
budget_hit = False
|
|
early_out = False
|
|
|
|
def _result(
|
|
*, success: bool, return_code: int,
|
|
error_type: ErrorType | None, error_message: str | None,
|
|
) -> DownloadResult:
|
|
return DownloadResult(
|
|
success=success,
|
|
url=url,
|
|
artist_slug=artist_slug,
|
|
platform="patreon",
|
|
files_downloaded=downloaded,
|
|
files_quarantined=0,
|
|
quarantined_paths=[],
|
|
written_paths=written,
|
|
stdout="\n".join(log_lines),
|
|
stderr="",
|
|
return_code=return_code,
|
|
error_type=error_type,
|
|
error_message=error_message,
|
|
duration_seconds=time.monotonic() - start,
|
|
)
|
|
|
|
try:
|
|
for post, included, page_cursor in self.client.iter_posts(
|
|
campaign_id, cursor=resume_cursor
|
|
):
|
|
# Checkpoint: emit the cursor that FETCHED this page once, the
|
|
# moment we START it — so a chunk cut mid-page resumes the page,
|
|
# not the one after it (matches the gallery-dl cursor semantics
|
|
# download_service's lifecycle already depends on).
|
|
if page_cursor and page_cursor != emitted_cursor:
|
|
log_lines.append(f"Cursor: {page_cursor}")
|
|
emitted_cursor = page_cursor
|
|
|
|
# Time-box check at the post boundary (coarse, like a gallery-dl
|
|
# chunk). Backfill/recovery resume from emitted_cursor next chunk.
|
|
if time.monotonic() - start >= time_budget_seconds:
|
|
budget_hit = True
|
|
break
|
|
|
|
media = self.client.extract_media(post, included)
|
|
if not media:
|
|
continue
|
|
|
|
keys = [_ledger_key(m) for m in media]
|
|
seen = (
|
|
set()
|
|
if bypass_seen
|
|
else self._seen_keys(source_id, keys)
|
|
)
|
|
|
|
def _is_seen(m: MediaItem, _seen=seen) -> bool:
|
|
return _ledger_key(m) in _seen
|
|
|
|
outcomes = self.downloader.download_post(
|
|
post, media, artist_slug, is_seen=_is_seen
|
|
)
|
|
|
|
to_mark: list[tuple[str, str]] = []
|
|
for media_item, outcome in zip(media, outcomes, strict=False):
|
|
key = _ledger_key(media_item)
|
|
if outcome.status == "downloaded":
|
|
downloaded += 1
|
|
if outcome.path is not None:
|
|
written.append(str(outcome.path))
|
|
to_mark.append((key, media_item.post_id))
|
|
consecutive_seen = 0
|
|
elif outcome.status == "skipped_disk":
|
|
# Already on disk (a prior run). Reconcile the ledger so a
|
|
# later tick skips it at tier-1 without a disk stat, but
|
|
# do NOT re-feed it to phase 3 — attach_in_place would see
|
|
# the duplicate sha256 and unlink the on-disk copy.
|
|
to_mark.append((key, media_item.post_id))
|
|
consecutive_seen += 1
|
|
elif outcome.status == "skipped_seen":
|
|
consecutive_seen += 1
|
|
elif outcome.status == "error":
|
|
errors += 1
|
|
# An error neither advances nor resets the run-of-seen.
|
|
|
|
if mode == "tick" and consecutive_seen >= seen_threshold:
|
|
early_out = True
|
|
break
|
|
|
|
# Mark seen AFTER the network fetch, on its own short session.
|
|
if to_mark:
|
|
self._mark_seen(source_id, to_mark)
|
|
|
|
if early_out:
|
|
break
|
|
else:
|
|
reached_bottom = True
|
|
except PatreonAPIError as exc:
|
|
# Base of PatreonAuthError + PatreonDriftError — catches every
|
|
# client-level failure; _failure_result maps it to a typed error.
|
|
return self._failure_result(exc, _result)
|
|
|
|
if errors:
|
|
log_lines.append(f"{errors} media item(s) failed")
|
|
log_lines.append(
|
|
f"Patreon ingest ({mode}): {downloaded} downloaded, "
|
|
f"{errors} error(s)"
|
|
+ (", reached end" if reached_bottom else "")
|
|
+ (", time-boxed" if budget_hit else "")
|
|
)
|
|
|
|
if budget_hit:
|
|
# A chunk that hit its time-box but made forward progress is a
|
|
# NORMAL chunk boundary, not a failure (PARTIAL → status "ok"); the
|
|
# next chunk resumes from the emitted cursor. No progress → TIMEOUT,
|
|
# which feeds download_service's backfill stall-guard. rc<0 mirrors
|
|
# subprocess TimeoutExpired so completion detection stays false.
|
|
made_progress = downloaded > 0 or emitted_cursor != resume_cursor
|
|
if made_progress:
|
|
return _result(
|
|
success=False, return_code=-1,
|
|
error_type=ErrorType.PARTIAL,
|
|
error_message=(
|
|
f"Patreon backfill chunk: {downloaded} file(s) — continuing"
|
|
),
|
|
)
|
|
return _result(
|
|
success=False, return_code=-1,
|
|
error_type=ErrorType.TIMEOUT,
|
|
error_message="Patreon chunk timed out with no progress",
|
|
)
|
|
|
|
# Normal success: reached the bottom, or a tick that early-outed. rc 0 +
|
|
# error_type None is REQUIRED for a backfill/recovery walk that reached
|
|
# the bottom to be marked COMPLETE by
|
|
# download_service._apply_backfill_lifecycle — so we return None even
|
|
# when downloaded == 0 (a re-confirming walk that found nothing new still
|
|
# completed). NO_NEW_CONTENT would read as "not finished" there and trip
|
|
# the stall guard. success=True maps to status "ok" regardless. A tick
|
|
# that early-outed also returns here; ticks never set backfill state so
|
|
# the lifecycle is a no-op for them.
|
|
return _result(
|
|
success=True, return_code=0,
|
|
error_type=None, error_message=None,
|
|
)
|
|
|
|
# -- failure mapping ---------------------------------------------------
|
|
|
|
def _failure_result(self, exc: Exception, _result) -> DownloadResult:
|
|
"""Map a client-level exception to a loud, typed failed DownloadResult.
|
|
|
|
We NEVER return a silent zero-download "success" — the whole point of the
|
|
native ingester is to fail RED when Patreon's API shape or our auth
|
|
changes. The typed mapping lets FailingSourcesCard render the right chip
|
|
and tells the operator what to do:
|
|
- PatreonAuthError → AUTH_ERROR (rotate cookies)
|
|
- PatreonDriftError → API_DRIFT (ingester field-set/parser needs update)
|
|
- HTTP 429 / 404 → RATE_LIMITED / NOT_FOUND
|
|
- other HTTP status → HTTP_ERROR; transport failure → NETWORK_ERROR
|
|
|
|
PatreonAuthError and PatreonDriftError both subclass PatreonAPIError, so
|
|
they must be matched before the generic HTTP/transport fallthrough.
|
|
"""
|
|
message = str(exc)
|
|
if isinstance(exc, PatreonAuthError):
|
|
error_type = ErrorType.AUTH_ERROR
|
|
elif isinstance(exc, PatreonDriftError):
|
|
error_type = ErrorType.API_DRIFT
|
|
message = f"Patreon API changed — ingester needs update: {message}"
|
|
else: # generic PatreonAPIError: HTTP non-2xx (status_code set) or transport
|
|
status = getattr(exc, "status_code", None)
|
|
if status == 429:
|
|
error_type = ErrorType.RATE_LIMITED
|
|
elif status == 404:
|
|
error_type = ErrorType.NOT_FOUND
|
|
elif status is not None:
|
|
error_type = ErrorType.HTTP_ERROR
|
|
else:
|
|
error_type = ErrorType.NETWORK_ERROR
|
|
log.warning("Patreon ingest failed (%s): %s", error_type.value, message)
|
|
return _result(
|
|
success=False, return_code=1,
|
|
error_type=error_type, error_message=message,
|
|
)
|
|
|
|
# -- seen-ledger (short-lived sessions) --------------------------------
|
|
|
|
def _seen_keys(self, source_id: int, keys: list[str]) -> set[str]:
|
|
"""Which of `keys` are already in the ledger for this source.
|
|
|
|
One short SELECT on its own session — opened and closed without any
|
|
network in between (the GETs happen after, in download_post).
|
|
"""
|
|
if not keys:
|
|
return set()
|
|
with self.session_factory() as session:
|
|
rows = session.execute(
|
|
select(PatreonSeenMedia.filehash).where(
|
|
PatreonSeenMedia.source_id == source_id,
|
|
PatreonSeenMedia.filehash.in_(keys),
|
|
)
|
|
).scalars().all()
|
|
return set(rows)
|
|
|
|
def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None:
|
|
"""Idempotent upsert of (filehash, post_id) ledger rows for a page.
|
|
|
|
ON CONFLICT DO NOTHING against the (source_id, filehash) UNIQUE so a
|
|
re-sighting — or a concurrent walk — is a harmless no-op
|
|
([[scalar_one_or_none-duplicates]]: never check-then-insert without the
|
|
DB constraint backing it). De-dup the batch locally first so a single
|
|
page can't present the same key twice to one INSERT.
|
|
"""
|
|
seen_local: set[str] = set()
|
|
values = []
|
|
for key, post_id in items:
|
|
if key in seen_local:
|
|
continue
|
|
seen_local.add(key)
|
|
values.append(
|
|
{"source_id": source_id, "filehash": key, "post_id": post_id}
|
|
)
|
|
if not values:
|
|
return
|
|
with self.session_factory() as session:
|
|
stmt = pg_insert(PatreonSeenMedia).values(values)
|
|
stmt = stmt.on_conflict_do_nothing(
|
|
constraint="uq_patreon_seen_media_source_id"
|
|
)
|
|
session.execute(stmt)
|
|
session.commit()
|