diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py new file mode 100644 index 0000000..099f634 --- /dev/null +++ b/backend/app/services/ingest_core.py @@ -0,0 +1,459 @@ +"""Platform-agnostic native-ingest core (plan #706, build on #697/#703/#704/#705). + +The orchestration that drives a native subscription walk — page a feed → +extract media → tiered skip (seen-ledger / on-disk / dead-letter) → download → +mark-seen / record-failures / checkpoint-cursor → return a gallery-dl-shaped +`DownloadResult`, across tick/backfill/recovery modes — is identical for every +platform. Only four things are platform-specific, and they're INJECTED at +construction by a thin adapter (e.g. `PatreonIngester`): + + - `client` — `.iter_posts(feed_id, cursor)` yielding `(post, included, + page_cursor)` + `.extract_media(post, included) -> [media]`. + - `downloader`— `.download_post(post, media, artist_slug, is_seen) -> + [MediaOutcome]` (status in downloaded/skipped_seen/skipped_disk + /quarantined/error; `.path`/`.error`/`.post_id`). + - ledger — `seen_model` + `failed_model` SQLAlchemy models (+ their + on-conflict UNIQUE constraint names) and a `ledger_key(media)`. + - failure map — the adapter overrides `_failure_result` (platform exception + → DownloadResult.error_type) and supplies `error_base` (the + exception type the walk catches) + `platform` (result label). + +Everything DB touches a SHORT-LIVED sync session from the injected sessionmaker — +never held across a network fetch ([[db-connection-held-across-subprocess]]). +Plain-HTTP homelab: no secure-context Web API. +""" + +from __future__ import annotations + +import logging +import time +from collections.abc import Callable + +from sqlalchemy import delete, func, select, text +from sqlalchemy.dialects.postgresql import insert as pg_insert + +from .gallery_dl import DownloadResult, ErrorType + +log = logging.getLogger(__name__) + +# Stop a tick after this many CONTIGUOUS already-have-it media (seen-ledger or +# on-disk) — the cheap native equivalent of gallery-dl's `exit:20`, now free of +# per-file HEADs. Headroom against paywalled/undownloadable items interleaving. +_TICK_SEEN_THRESHOLD = 20 + +# plan #705 #7: after this many failed download/validate attempts a media is +# "dead-lettered" and skipped on routine tick/backfill walks (recovery still +# re-attempts it). Stops a permanently-broken media re-erroring forever. +DEAD_LETTER_THRESHOLD = 3 +# last_error is Text but bound it so a giant traceback doesn't bloat the row. +_ERROR_MAX = 1000 + + +class Ingester: + """Generic native-ingest orchestration. Subclass with a platform adapter + (see the module docstring) — or construct directly with the keyword seams.""" + + def __init__( + self, + *, + client, + downloader, + session_factory: Callable[[], object], + seen_model, + failed_model, + seen_constraint: str, + failed_constraint: str, + ledger_key: Callable[[object], str], + platform: str, + error_base: type[Exception], + ): + self.client = client + self.downloader = downloader + self.session_factory = session_factory + self._seen_model = seen_model + self._failed_model = failed_model + self._seen_constraint = seen_constraint + self._failed_constraint = failed_constraint + self._ledger_key = ledger_key + self._platform = platform + self._error_base = error_base + + # -- 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 AND the dead-letter 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" + # Only deep walks checkpoint their cursor mid-flight (plan #705 #6); a + # tick has no resumable backfill state. + checkpoint = mode in ("backfill", "recovery") + ledger_key = self._ledger_key + start = time.monotonic() + log_lines: list[str] = [] + written: list[str] = [] + quarantined_paths: list[str] = [] + downloaded = 0 + errors = 0 + quarantined = 0 + dead_lettered = 0 + skipped_count = 0 + posts_processed = 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: + # plan #704: return STRUCTURED data — phase 3 reads run_stats/cursor + # directly instead of regex-scraping a reconstructed stdout. stdout + # stays a human-readable summary (no fake `Cursor:` lines). + return DownloadResult( + success=success, + url=url, + artist_slug=artist_slug, + platform=self._platform, + files_downloaded=downloaded, + files_quarantined=quarantined, + quarantined_paths=list(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, + cursor=emitted_cursor, + posts_processed=posts_processed, + run_stats={ + "exit_code": return_code, + "downloaded_count": downloaded, + "skipped_count": skipped_count, + "per_item_failures": errors, + "warning_count": 0, + "tier_gated_count": 0, + "quarantined_count": quarantined, + "dead_lettered_count": dead_lettered, + }, + ) + + try: + for post, included, page_cursor in self.client.iter_posts( + campaign_id, cursor=resume_cursor + ): + # Checkpoint the cursor that FETCHED this page the moment we + # START it — so a chunk cut mid-page resumes the page, not the one + # after it. Carried as DownloadResult.cursor (plan #704). + if page_cursor and page_cursor != emitted_cursor: + emitted_cursor = page_cursor + # plan #705 #6: persist the cursor at each page boundary so a + # worker SIGKILL mid-chunk resumes near the crash, not the + # chunk start. (phase 3 still writes the final cursor — same + # value; this is the crash-safety net.) + if checkpoint: + self._checkpoint_cursor(source_id, emitted_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 + + posts_processed += 1 + media = self.client.extract_media(post, included) + if not media: + continue + + keys = [ledger_key(m) for m in media] + # Recovery bypasses BOTH the seen-ledger AND the dead-letter + # ledger (the operator's "try everything again"); routine walks + # skip seen + dead media (tier-1 + tier-1.5, plan #705 #7). + dead = set() if bypass_seen else self._dead_keys(source_id, keys) + seen = ( + set() + if bypass_seen + else self._seen_keys(source_id, keys) + ) + skip = seen | dead + + def _is_skip(m, _skip=skip) -> bool: + return ledger_key(m) in _skip + + outcomes = self.downloader.download_post( + post, media, artist_slug, is_seen=_is_skip + ) + + to_mark: list[tuple[str, str]] = [] + to_clear: list[str] = [] # recovered → drop any dead-letter row + to_fail: list[tuple[str, str, str]] = [] # (key, post_id, error) + for media_item, outcome in zip(media, outcomes, strict=False): + key = ledger_key(media_item) + if key in dead: + dead_lettered += 1 # skipped because previously dead + 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)) + to_clear.append(key) + 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)) + to_clear.append(key) + skipped_count += 1 + consecutive_seen += 1 + elif outcome.status == "skipped_seen": + skipped_count += 1 + consecutive_seen += 1 + elif outcome.status == "quarantined": + # New content that failed validation (corrupt) — counted + # distinctly so the run surfaces a real quarantined total. + # Not marked seen (a later walk may re-fetch a fixed file); + # it IS new content, so it breaks the run-of-seen. Counts + # toward the dead-letter ledger (plan #705 #7). + quarantined += 1 + if outcome.path is not None: + quarantined_paths.append(str(outcome.path)) + to_fail.append((key, media_item.post_id, outcome.error or "quarantined")) + consecutive_seen = 0 + elif outcome.status == "error": + errors += 1 + to_fail.append((key, media_item.post_id, outcome.error or "error")) + # An error neither advances nor resets the run-of-seen. + + if mode == "tick" and consecutive_seen >= seen_threshold: + early_out = True + break + + # Persist ledger changes AFTER the network fetch, on short + # sessions: mark downloaded/on-disk seen, clear any dead-letter + # for recovered media, and record failures (plan #705 #7). + if to_mark: + self._mark_seen(source_id, to_mark) + if to_clear: + self._clear_failures(source_id, to_clear) + if to_fail: + self._record_failures(source_id, to_fail) + + if early_out: + break + else: + reached_bottom = True + except self._error_base as exc: + # The platform's client-error base — _failure_result (adapter) + # maps it to a typed error. + return self._failure_result(exc, _result) + + if errors: + log_lines.append(f"{errors} media item(s) failed") + if quarantined: + log_lines.append(f"{quarantined} media item(s) quarantined (invalid)") + if dead_lettered: + log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)") + log_lines.append( + f"{self._platform} ingest ({mode}): {downloaded} downloaded, " + f"{skipped_count} skipped, {quarantined} quarantined, " + f"{dead_lettered} dead-lettered, {errors} error(s), " + f"{posts_processed} post(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"Backfill chunk: {downloaded} file(s) — continuing" + ), + ) + return _result( + success=False, return_code=-1, + error_type=ErrorType.TIMEOUT, + error_message="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). 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 (adapter overrides) ------------------------------- + + def _failure_result(self, exc: Exception, _result) -> DownloadResult: + """Map a platform client-error to a typed failed DownloadResult. The base + gives a safe default; adapters override with their exception taxonomy.""" + log.warning("%s ingest failed: %s", self._platform, exc) + return _result( + success=False, return_code=1, + error_type=ErrorType.UNKNOWN_ERROR, error_message=str(exc), + ) + + # -- seen-ledger (short-lived sessions) -------------------------------- + + def _seen_keys(self, source_id: int, keys: list[str]) -> set[str]: + """Which of `keys` are already in the seen-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(self._seen_model.filehash).where( + self._seen_model.source_id == source_id, + self._seen_model.filehash.in_(keys), + ) + ).scalars().all() + return set(rows) + + def _checkpoint_cursor(self, source_id: int, cursor: str) -> None: + """Persist the in-progress backfill cursor mid-walk (plan #705 #6). + + ATOMIC, single-key UPDATE: cast the JSON column to jsonb, set just + `_backfill_cursor`, cast back — so it never clobbers operator config or + the other backfill keys (no read-modify-write race). The in-flight guard + means only this source's one download runs at a time; a concurrent + operator stop is benign (a stray cursor with no `_backfill_state` is + ignored by tick mode and cleared on the next start). + """ + with self.session_factory() as session: + session.execute( + text( + "UPDATE source SET config_overrides = jsonb_set(" + " coalesce(config_overrides::jsonb, '{}'::jsonb)," + " '{_backfill_cursor}', to_jsonb(cast(:cur AS text))" + ")::json WHERE id = :sid" + ), + {"cur": cursor, "sid": source_id}, + ) + session.commit() + + def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None: + """Idempotent upsert of (filehash, post_id) seen-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(self._seen_model).values(values) + stmt = stmt.on_conflict_do_nothing(constraint=self._seen_constraint) + session.execute(stmt) + session.commit() + + # -- dead-letter ledger (plan #705 #7) --------------------------------- + + def _dead_keys(self, source_id: int, keys: list[str]) -> set[str]: + """Which of `keys` have failed >= DEAD_LETTER_THRESHOLD times (dead). + One short SELECT; recovery never calls this (it re-attempts dead media).""" + if not keys: + return set() + with self.session_factory() as session: + rows = session.execute( + select(self._failed_model.filehash).where( + self._failed_model.source_id == source_id, + self._failed_model.filehash.in_(keys), + self._failed_model.attempts >= DEAD_LETTER_THRESHOLD, + ) + ).scalars().all() + return set(rows) + + def _record_failures( + self, source_id: int, items: list[tuple[str, str, str]] + ) -> None: + """Upsert-increment the dead-letter ledger for failed media. On conflict + bump `attempts` and refresh last_error/last_failed_at (UNIQUE backs the + upsert — no check-then-insert). De-dup the batch (one row/key, last error + wins).""" + by_key: dict[str, str] = {} + for key, _post_id, err in items: + by_key[key] = (err or "")[:_ERROR_MAX] + if not by_key: + return + values = [ + {"source_id": source_id, "filehash": k, "attempts": 1, "last_error": e} + for k, e in by_key.items() + ] + with self.session_factory() as session: + stmt = pg_insert(self._failed_model).values(values) + stmt = stmt.on_conflict_do_update( + constraint=self._failed_constraint, + set_={ + "attempts": self._failed_model.attempts + 1, + "last_error": stmt.excluded.last_error, + "last_failed_at": func.now(), + }, + ) + session.execute(stmt) + session.commit() + + def _clear_failures(self, source_id: int, keys: list[str]) -> None: + """Drop dead-letter rows for media that just downloaded cleanly — they + recovered. A no-op DELETE for keys that were never failing.""" + unique = list(dict.fromkeys(keys)) + if not unique: + return + with self.session_factory() as session: + session.execute( + delete(self._failed_model).where( + self._failed_model.source_id == source_id, + self._failed_model.filehash.in_(unique), + ) + ) + session.commit() diff --git a/backend/app/services/patreon_ingester.py b/backend/app/services/patreon_ingester.py index 206b555..3862088 100644 --- a/backend/app/services/patreon_ingester.py +++ b/backend/app/services/patreon_ingester.py @@ -1,13 +1,14 @@ -"""Native Patreon ingester — phase-2 orchestrator (build step 3). +"""Native Patreon ingester — the Patreon ADAPTER over the platform-agnostic core. -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. +The orchestration that drives a native subscription walk (page a feed → extract +media → tiered skip → download → mark-seen / record-failures / checkpoint-cursor +→ return a gallery-dl-shaped `DownloadResult`, across tick/backfill/recovery) +now lives in `ingest_core.Ingester` — it's identical for every platform. This +module is the thin Patreon adapter: it wires the Patreon `client`/`downloader`/ +ledger models/constraints/key into the core and supplies the Patreon-specific +failure mapping. `download_service.download_source` calls `PatreonIngester.run` +exactly as before; the public surface (this class, `_ledger_key`, +`DEAD_LETTER_THRESHOLD`, `verify_patreon_credential`) is unchanged. Three modes (selected by `download_service` from `config_overrides` state): - tick — newest→oldest, skip seen (tier-1 ledger + tier-2 disk), early-out @@ -15,24 +16,14 @@ Three modes (selected by `download_service` from `config_overrides` state): 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: `` 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]]). + - recovery — like backfill but BYPASSES the tier-1 seen-ledger AND the + dead-letter 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). +The seen/dead-letter ledgers live in Postgres (`patreon_seen_media` / +`patreon_failed_media`); the core opens SHORT-LIVED sync sessions per page batch +— never held across a network fetch ([[db-connection-held-across-subprocess]]). FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. """ @@ -40,15 +31,12 @@ from __future__ import annotations import asyncio import logging -import time from collections.abc import Callable from pathlib import Path -from sqlalchemy import delete, func, select, text -from sqlalchemy.dialects.postgresql import insert as pg_insert - from ..models import PatreonFailedMedia, PatreonSeenMedia from .gallery_dl import DownloadResult, ErrorType +from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester from .patreon_client import ( MediaItem, PatreonAPIError, @@ -59,27 +47,19 @@ from .patreon_client import ( from .patreon_downloader import PatreonDownloader from .patreon_resolver import resolve_campaign_id_for_source -log = logging.getLogger(__name__) +__all__ = [ + "DEAD_LETTER_THRESHOLD", + "PatreonIngester", + "_ledger_key", + "verify_patreon_credential", +] -# 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 +log = logging.getLogger(__name__) # 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 -# plan #705 #7: after this many failed download/validate attempts a media is -# "dead-lettered" and skipped on routine tick/backfill walks (recovery still -# re-attempts it). Stops a permanently-broken media (404'd CDN, deleted post) -# re-erroring forever and re-burning chunks. -DEAD_LETTER_THRESHOLD = 3 -# last_error is Text but bound it so a giant traceback doesn't bloat the row. -_ERROR_MAX = 1000 - def _ledger_key(media: MediaItem) -> str: """Stable per-media identity for the cross-run seen-ledger. @@ -96,12 +76,12 @@ def _ledger_key(media: MediaItem) -> str: return f"{media.post_id}:{media.filename}"[:_LEDGER_KEY_MAX] -class PatreonIngester: +class PatreonIngester(Ingester): """Walk a Patreon campaign's posts, download unseen media, return a - `DownloadResult`. + `DownloadResult`. A thin adapter over `ingest_core.Ingester`. Construct with the per-source `cookies_path` and a sync sessionmaker for the - seen-ledger. `client` / `downloader` are injectable seams so unit tests run + ledgers. `client` / `downloader` are injectable seams so unit tests run without network, subprocess, or a real CDN. """ @@ -119,266 +99,36 @@ class PatreonIngester: ): self.images_root = Path(images_root) self.cookies_path = str(cookies_path) if cookies_path else None - self.session_factory = session_factory # Pacing (plan #703): request_sleep paces the API page fetches, # rate_limit paces the media downloads. Injected client/downloader (in # tests) already carry their own pacing, so these only apply to the # default-constructed ones. - self.client = ( + resolved_client = ( client if client is not None else PatreonClient(cookies_path, request_sleep=request_sleep) ) - self.downloader = ( + resolved_downloader = ( downloader if downloader is not None else PatreonDownloader( self.images_root, cookies_path, validate=validate, rate_limit=rate_limit ) ) - - # -- 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" - # Only deep walks checkpoint their cursor mid-flight (plan #705 #6); a - # tick has no resumable backfill state. - checkpoint = mode in ("backfill", "recovery") - start = time.monotonic() - log_lines: list[str] = [] - written: list[str] = [] - quarantined_paths: list[str] = [] - downloaded = 0 - errors = 0 - quarantined = 0 - dead_lettered = 0 - skipped_count = 0 - posts_processed = 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: - # plan #704: return STRUCTURED data — phase 3 reads run_stats/cursor - # directly instead of regex-scraping a reconstructed stdout. stdout - # stays a human-readable summary (no fake `Cursor:` lines). - return DownloadResult( - success=success, - url=url, - artist_slug=artist_slug, - platform="patreon", - files_downloaded=downloaded, - files_quarantined=quarantined, - quarantined_paths=list(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, - cursor=emitted_cursor, - posts_processed=posts_processed, - run_stats={ - "exit_code": return_code, - "downloaded_count": downloaded, - "skipped_count": skipped_count, - "per_item_failures": errors, - "warning_count": 0, - "tier_gated_count": 0, - "quarantined_count": quarantined, - "dead_lettered_count": dead_lettered, - }, - ) - - try: - for post, included, page_cursor in self.client.iter_posts( - campaign_id, cursor=resume_cursor - ): - # Checkpoint the cursor that FETCHED this page the moment we - # START it — so a chunk cut mid-page resumes the page, not the one - # after it. Carried as DownloadResult.cursor (plan #704); no fake - # `Cursor:` stdout line to regex back out. - if page_cursor and page_cursor != emitted_cursor: - emitted_cursor = page_cursor - # plan #705 #6: persist the cursor at each page boundary so a - # worker SIGKILL mid-chunk resumes near the crash, not the - # chunk start. (phase 3 still writes the final cursor — same - # value; this is the crash-safety net.) - if checkpoint: - self._checkpoint_cursor(source_id, emitted_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 - - posts_processed += 1 - media = self.client.extract_media(post, included) - if not media: - continue - - keys = [_ledger_key(m) for m in media] - # Recovery bypasses BOTH the seen-ledger AND the dead-letter - # ledger (the operator's "try everything again"); routine walks - # skip seen + dead media (tier-1 + tier-1.5, plan #705 #7). - dead = set() if bypass_seen else self._dead_keys(source_id, keys) - seen = ( - set() - if bypass_seen - else self._seen_keys(source_id, keys) - ) - skip = seen | dead - - def _is_skip(m: MediaItem, _skip=skip) -> bool: - return _ledger_key(m) in _skip - - outcomes = self.downloader.download_post( - post, media, artist_slug, is_seen=_is_skip - ) - - to_mark: list[tuple[str, str]] = [] - to_clear: list[str] = [] # recovered → drop any dead-letter row - to_fail: list[tuple[str, str, str]] = [] # (key, post_id, error) - for media_item, outcome in zip(media, outcomes, strict=False): - key = _ledger_key(media_item) - if key in dead: - dead_lettered += 1 # skipped because previously dead - 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)) - to_clear.append(key) - 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)) - to_clear.append(key) - skipped_count += 1 - consecutive_seen += 1 - elif outcome.status == "skipped_seen": - skipped_count += 1 - consecutive_seen += 1 - elif outcome.status == "quarantined": - # New content that failed validation (corrupt) — counted - # distinctly so the run surfaces a real quarantined total. - # Not marked seen (a later walk may re-fetch a fixed file); - # it IS new content, so it breaks the run-of-seen. Counts - # toward the dead-letter ledger (plan #705 #7). - quarantined += 1 - if outcome.path is not None: - quarantined_paths.append(str(outcome.path)) - to_fail.append((key, media_item.post_id, outcome.error or "quarantined")) - consecutive_seen = 0 - elif outcome.status == "error": - errors += 1 - to_fail.append((key, media_item.post_id, outcome.error or "error")) - # An error neither advances nor resets the run-of-seen. - - if mode == "tick" and consecutive_seen >= seen_threshold: - early_out = True - break - - # Persist ledger changes AFTER the network fetch, on short - # sessions: mark downloaded/on-disk seen, clear any dead-letter - # for recovered media, and record failures (plan #705 #7). - if to_mark: - self._mark_seen(source_id, to_mark) - if to_clear: - self._clear_failures(source_id, to_clear) - if to_fail: - self._record_failures(source_id, to_fail) - - 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") - if quarantined: - log_lines.append(f"{quarantined} media item(s) quarantined (invalid)") - if dead_lettered: - log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)") - log_lines.append( - f"Patreon ingest ({mode}): {downloaded} downloaded, " - f"{skipped_count} skipped, {quarantined} quarantined, " - f"{dead_lettered} dead-lettered, {errors} error(s), " - f"{posts_processed} post(s)" - + (", reached end" if reached_bottom else "") - + (", time-boxed" if budget_hit else "") + super().__init__( + client=resolved_client, + downloader=resolved_downloader, + session_factory=session_factory, + seen_model=PatreonSeenMedia, + failed_model=PatreonFailedMedia, + seen_constraint="uq_patreon_seen_media_source_id", + failed_constraint="uq_patreon_failed_media_source_id", + ledger_key=_ledger_key, + platform="patreon", + error_base=PatreonAPIError, ) - 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 --------------------------------------------------- + # -- failure mapping (Patreon exception taxonomy) ---------------------- def _failure_result(self, exc: Exception, _result) -> DownloadResult: """Map a client-level exception to a loud, typed failed DownloadResult. @@ -417,136 +167,6 @@ class PatreonIngester: 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 _checkpoint_cursor(self, source_id: int, cursor: str) -> None: - """Persist the in-progress backfill cursor mid-walk (plan #705 #6). - - ATOMIC, single-key UPDATE: cast the JSON column to jsonb, set just - `_backfill_cursor`, cast back — so it never clobbers operator config or - the other backfill keys (no read-modify-write race). The in-flight guard - means only this source's one download runs at a time; a concurrent - operator stop is benign (a stray cursor with no `_backfill_state` is - ignored by tick mode and cleared on the next start). - """ - with self.session_factory() as session: - session.execute( - text( - "UPDATE source SET config_overrides = jsonb_set(" - " coalesce(config_overrides::jsonb, '{}'::jsonb)," - " '{_backfill_cursor}', to_jsonb(cast(:cur AS text))" - ")::json WHERE id = :sid" - ), - {"cur": cursor, "sid": source_id}, - ) - session.commit() - - 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() - - # -- dead-letter ledger (plan #705 #7) --------------------------------- - - def _dead_keys(self, source_id: int, keys: list[str]) -> set[str]: - """Which of `keys` have failed >= DEAD_LETTER_THRESHOLD times (dead). - One short SELECT; recovery never calls this (it re-attempts dead media).""" - if not keys: - return set() - with self.session_factory() as session: - rows = session.execute( - select(PatreonFailedMedia.filehash).where( - PatreonFailedMedia.source_id == source_id, - PatreonFailedMedia.filehash.in_(keys), - PatreonFailedMedia.attempts >= DEAD_LETTER_THRESHOLD, - ) - ).scalars().all() - return set(rows) - - def _record_failures( - self, source_id: int, items: list[tuple[str, str, str]] - ) -> None: - """Upsert-increment the dead-letter ledger for failed media. On conflict - bump `attempts` and refresh last_error/last_failed_at (UNIQUE backs the - upsert — no check-then-insert, [[scalar_one_or_none-duplicates]]). De-dup - the batch locally (one row per key, last error wins).""" - by_key: dict[str, str] = {} - for key, _post_id, err in items: - by_key[key] = (err or "")[:_ERROR_MAX] - if not by_key: - return - values = [ - {"source_id": source_id, "filehash": k, "attempts": 1, "last_error": e} - for k, e in by_key.items() - ] - with self.session_factory() as session: - stmt = pg_insert(PatreonFailedMedia).values(values) - stmt = stmt.on_conflict_do_update( - constraint="uq_patreon_failed_media_source_id", - set_={ - "attempts": PatreonFailedMedia.attempts + 1, - "last_error": stmt.excluded.last_error, - "last_failed_at": func.now(), - }, - ) - session.execute(stmt) - session.commit() - - def _clear_failures(self, source_id: int, keys: list[str]) -> None: - """Drop dead-letter rows for media that just downloaded cleanly — they - recovered. A no-op DELETE for keys that were never failing.""" - unique = list(dict.fromkeys(keys)) - if not unique: - return - with self.session_factory() as session: - session.execute( - delete(PatreonFailedMedia).where( - PatreonFailedMedia.source_id == source_id, - PatreonFailedMedia.filehash.in_(unique), - ) - ) - session.commit() - async def verify_patreon_credential( url: str, diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 0b3df18..32db179 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -300,7 +300,9 @@ async def test_backfill_budget_cut_returns_partial_with_progress( source_id, sync_engine, tmp_path, monkeypatch, ): # Deterministic clock: start=0, post1 check=10 (ok), post2 check=200 (>budget). - import backend.app.services.patreon_ingester as mod + # run() lives in ingest_core now, so patch the clock there (the same global + # time module object, but we reference it through the module that uses it). + import backend.app.services.ingest_core as core ticks = iter([0.0, 10.0, 200.0, 250.0]) last = [0.0] @@ -311,7 +313,7 @@ async def test_backfill_budget_cut_returns_partial_with_progress( pass return last[0] - monkeypatch.setattr(mod.time, "monotonic", fake_monotonic) + monkeypatch.setattr(core.time, "monotonic", fake_monotonic) pages = [ ("CUR1", [("p1", [_media("p1", 1)])]),