"""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, should_stop) -> [MediaOutcome]` (status in downloaded/ skipped_seen/skipped_disk/quarantined/error; `.path`/`.error`/`.post_id`). `should_stop()` is polled between media so the time-box is honoured mid-post. - 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 json 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, make_run_stats 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 # plan #709: throttle the live-progress write to the running DownloadEvent to one # every ~5s — a steady cadence for the Downloads view regardless of how big/slow a # page is (page boundaries can be minutes apart on image-dense backfills, so a # page-tied update would lurch). Trivial churn (~one single-row UPDATE / 5s). _LIVE_PROGRESS_INTERVAL = 5.0 # Post-body schema-drift canary (#862). Patreon's body lives in # content/content_json_string with NO post_type gate, so a field rename (as # content→content_json_string already was) zeroes EVERY body at once — across # every artist, every walk. If a native walk records at least this many posts # and extracts a body from NONE of them, treat it as that break (fail the run # API_DRIFT) rather than silently archiving empties. A *fraction* threshold would # false-positive on gallery/art creators who legitimately post images with no # caption, so the gate is "zero across a minimum sample": a real creator nearly # always has SOME text across this many posts, a broken parser has none. Set high # enough that a small tick (a few new posts) can't trip it — only a backfill / # recapture (the operator's schema-test flow) reaches the sample. _CANARY_MIN_SAMPLE = 30 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, posts_base: int = 0, event_id: int | None = None, ) -> DownloadResult: """Walk + download for one source, returning a gallery-dl-shaped result. `mode` is "tick" | "backfill" | "recovery" | "recapture". Recovery bypasses the tier-1 seen-ledger AND the dead-letter ledger (tier-2 disk still skips kept files). Recapture (#830) is the cheap "re-grab post text" walk: it bypasses the post-record gate (so EVERY post's body + external links are re-captured / detail-fetched) but KEEPS the media seen-ledger — on-disk media is NOT re-downloaded, only surfaced so its ImageRecord's source_filehash can be backfilled for inline-image localization. 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" recapture = mode == "recapture" # Both recovery and recapture re-capture EVERY post's body + links — they # bypass the post-record seen-gate (recovery via bypass_seen, recapture # explicitly). A plain backfill stays gated (capture once per post). recapture_records = bypass_seen or recapture # Only deep walks checkpoint their cursor mid-flight (plan #705 #6); a # tick has no resumable backfill state. checkpoint = mode in ("backfill", "recovery", "recapture") ledger_key = self._ledger_key # POST-FIRST CONTRACT (milestone #67): these two optional seams make a # platform "post-first" on the native core ingester — the post-record is # the single authoritative writer of the post body/links/metadata, and the # per-media sidecar carries image identity only (download_service flips # importer.post_first via uses_native_ingester, so the import side follows # automatically). A platform migrating off gallery-dl onto the native core # adopts post-first by implementing BOTH: # client.post_record_key(post) -> (ledger_key, post_id) | None (gate) # downloader.write_post_record(post, artist_slug) -> PostRecordOutcome # Absent on stub clients/downloaders (unit tests) and on not-yet-migrated # platforms → media-less posts are skipped as before and the body still # comes from the per-media sidecar (gallery-dl path). See [[post-first-ingest-contract]]. post_record_key = getattr(self.client, "post_record_key", None) write_post_record = getattr(self.downloader, "write_post_record", None) start = time.monotonic() last_live = start # plan #709: last live-progress write timestamp log_lines: list[str] = [] written: list[str] = [] post_records: list[str] = [] quarantined_paths: list[str] = [] # #830 recapture: (on-disk path, CDN source_url) pairs for already-present # media, so phase 3 can backfill the ImageRecord's source_filehash WITHOUT # re-downloading or unlinking the file. Empty outside recapture mode. relink: list[tuple[str, str]] = [] downloaded = 0 errors = 0 quarantined = 0 dead_lettered = 0 skipped_count = 0 posts_processed = 0 # Post-body schema-drift canary counters (#862, native ingester only — # gallery-dl walks never enter the post-record block below so these stay 0 # and the canary can't fire there). posts_recorded = post-records attempted # this walk; posts_with_body = how many yielded a non-empty body. posts_recorded = 0 posts_with_body = 0 # Net-new posts THIS chunk for the live progress badge (plan #704 #5); # excludes the re-walked resume page so _backfill_posts stays a monotonic # absolute across chunks instead of an inflating sum. posts_processed # stays the gross per-chunk count used for the run summary. chunk_new_posts = 0 consecutive_seen = 0 emitted_cursor: str | None = None reached_bottom = False budget_hit = False early_out = False stopped = False # plan #708 B4: operator hit Stop mid-walk cancel_armed = False # latched once we observe a live "running" state 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, post_record_paths=list(post_records), relink_source_paths=list(relink), 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=make_run_stats( exit_code=return_code, downloaded_count=downloaded, skipped_count=skipped_count, per_item_failures=errors, 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.) plan #704 #5: persist # the live posts count alongside it so the badge climbs DURING # the chunk, not only when it ends. if checkpoint: # plan #708 B4: an operator Stop pops `_backfill_state` — # bail at the page boundary (progress already checkpointed) # before more network work, so the live chunk halts # promptly instead of running to its time-box. LATCH on the # first observed "running" state, so a run invoked WITHOUT a # running state (a unit test, or a stale call) never # spuriously self-cancels. A short SELECT, never held. if self._still_running(source_id): cancel_armed = True elif cancel_armed: stopped = True break self._checkpoint_cursor(source_id, emitted_cursor) self._checkpoint_posts(source_id, posts_base + chunk_new_posts) # 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 # The resume page (its cursor == resume_cursor) was already # counted by the chunk that checkpointed it — don't re-count it # into the persisted badge (plan #704 #5). First chunk has # resume_cursor None, so everything counts. if not (resume_cursor and page_cursor == resume_cursor): chunk_new_posts += 1 # Capture the post body + external links ONCE per post (gated by # the synthetic post key in the seen-ledger), for EVERY post — # whether or not it has downloadable media. This is what makes a # backfill/recovery re-walk RECAPTURE bodies + links for posts # whose media is already on disk: re-downloading existing media # never fills links the system never had, so the body recapture # has to ride the walk itself. Detail-fetch (for an empty feed # body) happens at most once per post — the gate then spares it on # later walks. bypass_seen (recovery) re-captures unconditionally. if post_record_key and write_post_record: rk = post_record_key(post) if rk is not None: pkey, ppid = rk already = ( set() if recapture_records else self._seen_keys(source_id, [pkey]) ) if pkey not in already: rec = write_post_record(post, artist_slug) posts_recorded += 1 if rec.body_chars: posts_with_body += 1 if rec.path is not None: post_records.append(str(rec.path)) self._mark_seen(source_id, [(pkey, ppid)]) # Per-post handling line in the run stdout (the existing # "Raw stdout" panel) — the downloader already read the # post; we only format its outcome here. post_type beside # a 0-char body is the "why is this one empty" answer. log_lines.append( f" post {ppid} [{rec.post_type or '?'}] " f"body: {rec.body_chars} chars" + ("" if rec.body_chars else " — EMPTY") + (f" — {rec.title}" if rec.title else "") ) 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 # Honour the time-box DURING a media-dense post too, not only at # the per-post boundary below — else one heavy post can blow the # chunk budget out to the Celery soft limit (Pocketacer, 2026-06-07). outcomes = self.downloader.download_post( post, media, artist_slug, is_seen=_is_skip, should_stop=lambda: time.monotonic() - start >= time_budget_seconds, recapture=recapture, ) 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 # #830 recapture: surface (on-disk path, CDN url) so phase # 3 can backfill source_filehash for inline-image # localization — a SEPARATE non-deleting channel, never the # import list (which would unlink the file, per above). if recapture and outcome.path is not None: relink.append((str(outcome.path), media_item.url)) 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) # plan #709: time-throttled live progress to the running event so # the Downloads view ticks ~every 5s, independent of page size. now = time.monotonic() if event_id is not None and (now - last_live) >= _LIVE_PROGRESS_INTERVAL: last_live = now self._write_live_progress(event_id, { "downloaded": downloaded, "skipped": skipped_count, "errors": errors, "quarantined": quarantined, "posts": posts_processed, }) 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) # plan #708 B4: a Stop already popped the backfill state (incl. cursor + # posts), so don't re-write them — return PARTIAL (reads as "ok/progress", # the lifecycle no-ops since state is gone) instead of a false "complete". if stopped: return _result( success=False, return_code=-1, error_type=ErrorType.PARTIAL, error_message=f"Stopped by operator: {downloaded} file(s) this chunk", ) # Final authoritative posts count for the badge — captures the last page # after the last boundary write and the time-box break (plan #704 #5). if checkpoint: self._checkpoint_posts(source_id, posts_base + chunk_new_posts) 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), {len(post_records)} post-record(s), " f"{len(relink)} relinked" # Body-capture health (#862): even below the canary's red-alarm # threshold, surfacing the ratio makes a partial extraction regression # visible in the Raw stdout (e.g. "bodies 3/180" reads as off). + (f", bodies {posts_with_body}/{posts_recorded}" if posts_recorded else "") + (", 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", ) # Post-body schema-drift canary (#862): a native walk recorded a # meaningful sample of posts but extracted a body from NONE of them. Since # the body field has no post_type gate, that's the signature of Patreon # renaming/restructuring the body field (as content→content_json_string # already was) — fail RED (API_DRIFT: "fix is the field-set/parser, not # creds") so the breakage screams instead of silently archiving empties. # Only reached on an otherwise-clean walk (timeout/stop/error returned # above), so it never masks a more specific failure. if posts_recorded >= _CANARY_MIN_SAMPLE and posts_with_body == 0: msg = ( f"Post-body canary: extracted a body from 0 of {posts_recorded} " "posts — Patreon's body field shape likely changed; the ingester " "needs a field-set/parser update." ) log_lines.append(msg) log.error("%s (artist=%s)", msg, artist_slug) return _result( success=False, return_code=-1, error_type=ErrorType.API_DRIFT, error_message=msg, ) # 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, ) # -- preview (dry-run) ------------------------------------------------- def preview( self, source_id: int, campaign_id: str, *, page_limit: int = 3, sample_size: int = 10, ) -> dict: """Dry-run (plan #708 B4): walk up to `page_limit` pages and count media NOT already in the seen/dead ledgers, WITHOUT downloading anything. Read-only — only the seen/dead SELECTs touch the DB (short sessions). Lets an operator gauge "is this source worth a backfill?" cheaply. Returns: {total_new, posts_scanned, pages_scanned, has_more, sample: [{title, date, new}, ...]} # sample = posts with new media A client-level failure (auth/drift) propagates to the caller. """ total_new = 0 posts_scanned = 0 pages_scanned = 0 has_more = False sample: list[dict] = [] unset = object() last_page: object = unset for post, included, page_cursor in self.client.iter_posts( campaign_id, cursor=None ): if page_cursor != last_page: last_page = page_cursor pages_scanned += 1 if pages_scanned > page_limit: has_more = True pages_scanned = page_limit break posts_scanned += 1 media = self.client.extract_media(post, included) if not media: continue keys = [self._ledger_key(m) for m in media] skip = self._seen_keys(source_id, keys) | self._dead_keys(source_id, keys) new_count = sum(1 for m in media if self._ledger_key(m) not in skip) total_new += new_count if new_count > 0 and len(sample) < sample_size: meta = self.client.post_meta(post) sample.append( { "title": meta.get("title") or "(untitled)", "date": meta.get("date"), "new": new_count, } ) return { "total_new": total_new, "posts_scanned": posts_scanned, "pages_scanned": pages_scanned, "has_more": has_more, "sample": sample, } # -- 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 _write_live_progress(self, event_id: int, counts: dict) -> None: """Throttled mid-walk write of live counts to the RUNNING download_event (plan #709) so the Downloads view shows progress before the chunk finishes. A short session (never held across the walk); the `status = 'running'` guard avoids clobbering an event phase 3 already finalized. `metadata` is JSONB — jsonb_set sets just the `live` key, leaving the rest for phase 3 to overwrite with the final run_stats.""" with self.session_factory() as session: session.execute( text( "UPDATE download_event SET metadata = jsonb_set(" " coalesce(metadata, '{}'::jsonb), '{live}'," " cast(:live AS jsonb)) " "WHERE id = :eid AND status = 'running'" ), {"live": json.dumps(counts), "eid": event_id}, ) session.commit() def _still_running(self, source_id: int) -> bool: """True while the source is armed for a deep walk (plan #708 B4). An operator Stop (`source_service.stop_backfill`) pops `_backfill_state`, so a False here means "cancel this chunk now". One short SELECT on its own session — never held across the walk ([[db-connection-held-across-subprocess]]).""" with self.session_factory() as session: state = session.execute( text( "SELECT config_overrides::jsonb ->> '_backfill_state' " "FROM source WHERE id = :sid" ), {"sid": source_id}, ).scalar_one_or_none() return state == "running" def _checkpoint_posts(self, source_id: int, posts: int) -> None: """Persist the live backfill posts-processed count mid-walk (plan #704 #5). Same atomic single-key jsonb_set dance as _checkpoint_cursor, on the `_backfill_posts` key (cast to a JSON number) — so the progress badge climbs DURING a chunk without clobbering operator config or the cursor. The ingester OWNS this key now; download_service no longer accumulates it post-chunk (which lagged a whole chunk and over-counted the resume page). """ with self.session_factory() as session: session.execute( text( "UPDATE source SET config_overrides = jsonb_set(" " coalesce(config_overrides::jsonb, '{}'::jsonb)," " '{_backfill_posts}', to_jsonb(cast(:posts AS int))" ")::json WHERE id = :sid" ), {"posts": posts, "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()