feat(ingest): Recapture mode — re-grab post bodies/links + localize on-disk inline images (#830)
A plain backfill gates post-body capture on the seen-ledger, so a post whose media is already on disk AND whose post key is already seen never gets its body recaptured (operator-flagged: Industrial Lust description missing). Recovery recaptures unconditionally but re-downloads the whole source. New 'recapture' walk mode (4th beside tick/backfill/recovery): bypasses the post-record gate so EVERY post's body + external links are re-captured (detail-fetching empty bodies) WITHOUT re-downloading on-disk media; and surfaces already-present media via a separate non-deleting relink channel so the importer backfills ImageRecord.source_filehash for inline-image localization. - ingest_core: recapture mode + recapture_records gate bypass + relink collect - patreon_downloader: recapture surfaces seen-on-disk as skipped_disk(path), never refetches seen-missing media, still downloads genuinely-new - importer.relink_source_filehash: NULL-only sha256 backfill, never unlinks - download_service: mode derivation + phase-3 relink loop + lifecycle clear - source_service/api: start_recapture + backfill_recapture field + action - frontend: Recapture kebab action + 'Recapturing' badge across SourceActions/ Row/Card/SubscriptionsTab + sources store - tests across ingester/downloader/importer/source_service/api/download_service Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+15
-11
@@ -138,14 +138,16 @@ async def delete_source(source_id: int):
|
||||
|
||||
@sources_bp.route("/<int:source_id>/backfill", methods=["POST"])
|
||||
async def set_backfill(source_id: int):
|
||||
"""Plan #693/#697: start/stop a run-until-done backfill, or start a recovery.
|
||||
Body: `{"action": "start" | "stop" | "recover"}` (default "start"). 'start'
|
||||
walks the full post history in time-boxed chunks until it reaches the bottom
|
||||
(then the source shows 'complete'); 'recover' is the same walk but bypasses
|
||||
the Patreon seen-ledger to re-fetch dropped-and-deleted near-dups under the
|
||||
current pHash threshold; 'stop' cancels either back to tick mode. Returns the
|
||||
updated source dict (incl. backfill_state / backfill_chunks /
|
||||
backfill_bypass_seen)."""
|
||||
"""Plan #693/#697 + #830: start/stop a backfill, or start a recovery /
|
||||
recapture. Body: `{"action": "start" | "stop" | "recover" | "recapture"}`
|
||||
(default "start"). 'start' walks the full post history in time-boxed chunks
|
||||
until it reaches the bottom (then the source shows 'complete'); 'recover' is
|
||||
the same walk but bypasses the Patreon seen-ledger to re-fetch
|
||||
dropped-and-deleted near-dups under the current pHash threshold; 'recapture'
|
||||
re-grabs EVERY post's body + external links and localizes on-disk inline
|
||||
images WITHOUT re-downloading media; 'stop' cancels any back to tick mode.
|
||||
Returns the updated source dict (incl. backfill_state / backfill_chunks /
|
||||
backfill_bypass_seen / backfill_recapture)."""
|
||||
from pathlib import Path
|
||||
|
||||
from ..services.credential_service import CredentialService
|
||||
@@ -157,10 +159,10 @@ async def set_backfill(source_id: int):
|
||||
|
||||
payload = await request.get_json(silent=True) or {}
|
||||
action = payload.get("action", "start")
|
||||
if action not in ("start", "stop", "recover"):
|
||||
if action not in ("start", "stop", "recover", "recapture"):
|
||||
return _bad(
|
||||
"invalid_action",
|
||||
detail="action must be 'start', 'stop', or 'recover'",
|
||||
detail="action must be 'start', 'stop', 'recover', or 'recapture'",
|
||||
)
|
||||
|
||||
# Pre-flight (plan #703 #2): before arming a deep walk on a native-ingester
|
||||
@@ -170,7 +172,7 @@ async def set_backfill(source_id: int):
|
||||
# platforms: gallery-dl verify is a slow --simulate subprocess, too heavy for
|
||||
# an arm action. The credential read happens in a session that's CLOSED
|
||||
# before the verify network call (don't hold a DB conn across the request).
|
||||
if action in ("start", "recover"):
|
||||
if action in ("start", "recover", "recapture"):
|
||||
async with get_session() as session:
|
||||
rec = await SourceService(session).get(source_id)
|
||||
if rec is None:
|
||||
@@ -200,6 +202,8 @@ async def set_backfill(source_id: int):
|
||||
record = await svc.start_backfill(source_id)
|
||||
elif action == "recover":
|
||||
record = await svc.start_recovery(source_id)
|
||||
elif action == "recapture":
|
||||
record = await svc.start_recapture(source_id)
|
||||
else:
|
||||
record = await svc.stop_backfill(source_id)
|
||||
except LookupError:
|
||||
|
||||
@@ -111,6 +111,10 @@ class DownloadService:
|
||||
# `_backfill_bypass_seen` flag rides alongside the running backfill state;
|
||||
# download mode is "recovery" when both are set.
|
||||
bypass_seen = bool(overrides.get("_backfill_bypass_seen"))
|
||||
# #830 recapture: re-grab post bodies/links + localize on-disk inline
|
||||
# images, WITHOUT re-downloading media. Rides alongside the backfill
|
||||
# state like _backfill_bypass_seen; mutually exclusive with it.
|
||||
recapture = bool(overrides.get("_backfill_recapture"))
|
||||
if in_backfill:
|
||||
skip_value: bool | str = BACKFILL_SKIP_VALUE
|
||||
source_config.timeout = BACKFILL_CHUNK_SECONDS
|
||||
@@ -130,6 +134,8 @@ class DownloadService:
|
||||
if uses_native_ingester(ctx["platform"]):
|
||||
if in_backfill and bypass_seen:
|
||||
mode = "recovery"
|
||||
elif in_backfill and recapture:
|
||||
mode = "recapture"
|
||||
elif in_backfill:
|
||||
mode = "backfill"
|
||||
else:
|
||||
@@ -396,6 +402,20 @@ class DownloadService:
|
||||
|
||||
await loop.run_in_executor(None, _upsert)
|
||||
|
||||
# #830 recapture: backfill source_filehash on EXISTING on-disk images so
|
||||
# their post-body inline <img src=CDN> remaps to the local copy. A
|
||||
# SEPARATE non-deleting channel (NOT the import list — that would unlink
|
||||
# the duplicate file). Empty outside recapture mode.
|
||||
for rel_str, rel_url in getattr(dl_result, "relink_source_paths", None) or []:
|
||||
rel_path = Path(rel_str)
|
||||
if not rel_path.exists(): # noqa: ASYNC240
|
||||
continue
|
||||
|
||||
def _relink(p=rel_path, u=rel_url):
|
||||
return self.importer.relink_source_filehash(p, u, artist=artist)
|
||||
|
||||
await loop.run_in_executor(None, _relink)
|
||||
|
||||
# Kick the off-platform file-host downloader for any links this run
|
||||
# recorded (mega/gdrive/…). Global + idempotent (only claims pending/
|
||||
# retryable rows); the beat sweep is the backstop. Lazy import dodges a
|
||||
@@ -508,6 +528,8 @@ class DownloadService:
|
||||
# plan #697: a recovery walk shares this lifecycle; clear its bypass
|
||||
# flag on completion so the next routine tick honors the seen-ledger.
|
||||
new_overrides.pop("_backfill_bypass_seen", None)
|
||||
# #830: same for the recapture flag.
|
||||
new_overrides.pop("_backfill_recapture", None)
|
||||
src.config_overrides = new_overrides
|
||||
src.backfill_runs_remaining = 0
|
||||
return
|
||||
|
||||
@@ -153,6 +153,11 @@ class DownloadResult:
|
||||
# Importer.upsert_post_record (keyed on external_post_id → updates, never
|
||||
# doubles, the same Post a media import would create).
|
||||
post_record_paths: list[str] = field(default_factory=list)
|
||||
# Native ingester, recapture mode only (#830): (on-disk path, CDN source_url)
|
||||
# pairs for already-present media whose ImageRecord.source_filehash should be
|
||||
# backfilled (inline-image localization) WITHOUT re-download or unlink. Empty
|
||||
# on the gallery-dl path and outside recapture.
|
||||
relink_source_paths: list[tuple[str, str]] = field(default_factory=list)
|
||||
stdout: str = ""
|
||||
stderr: str = ""
|
||||
return_code: int = 0
|
||||
|
||||
@@ -1005,6 +1005,35 @@ class Importer:
|
||||
return sv.strip()
|
||||
return None
|
||||
|
||||
def relink_source_filehash(
|
||||
self, path: Path, source_url: str, *, artist: Artist | None = None,
|
||||
) -> bool:
|
||||
"""#830 recapture: backfill source_url + source_filehash on an EXISTING
|
||||
on-disk image's ImageRecord (matched by sha256) so its post-body inline
|
||||
`<img src=CDN>` remaps to the local copy at render time — WITHOUT
|
||||
re-downloading and WITHOUT unlinking the file (unlike the import path's
|
||||
duplicate-hash branch). NULL-only, mirroring _apply_sidecar: never
|
||||
clobbers an already-set filehash. Returns True only when a row was
|
||||
updated. No-op when the file is gone, the url has no filehash, no record
|
||||
matches the bytes, or the record already carries a filehash.
|
||||
"""
|
||||
fh = filehash_from_url(source_url)
|
||||
if not fh:
|
||||
return False
|
||||
try:
|
||||
sha = _sha256_of(path)
|
||||
except OSError:
|
||||
return False
|
||||
record = self.session.execute(
|
||||
select(ImageRecord).where(ImageRecord.sha256 == sha)
|
||||
).scalar_one_or_none()
|
||||
if record is None or record.source_filehash is not None:
|
||||
return False
|
||||
record.source_url = source_url
|
||||
record.source_filehash = fh
|
||||
self.session.commit()
|
||||
return True
|
||||
|
||||
def _apply_sidecar(
|
||||
self,
|
||||
record: ImageRecord,
|
||||
|
||||
@@ -105,18 +105,28 @@ class Ingester:
|
||||
) -> 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:
|
||||
`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")
|
||||
checkpoint = mode in ("backfill", "recovery", "recapture")
|
||||
ledger_key = self._ledger_key
|
||||
# Optional seams (Patreon native ingester): capture pure-text posts that
|
||||
# have NO downloadable media so the artist archive is complete. Absent on
|
||||
@@ -129,6 +139,10 @@ class Ingester:
|
||||
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
|
||||
@@ -165,6 +179,7 @@ class Ingester:
|
||||
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,
|
||||
@@ -241,7 +256,7 @@ class Ingester:
|
||||
if rk is not None:
|
||||
pkey, ppid = rk
|
||||
already = (
|
||||
set() if bypass_seen
|
||||
set() if recapture_records
|
||||
else self._seen_keys(source_id, [pkey])
|
||||
)
|
||||
if pkey not in already:
|
||||
@@ -275,6 +290,7 @@ class Ingester:
|
||||
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]] = []
|
||||
@@ -300,6 +316,12 @@ class Ingester:
|
||||
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
|
||||
|
||||
@@ -198,6 +198,7 @@ class PatreonDownloader:
|
||||
*,
|
||||
is_seen: Callable[[object], bool] = lambda m: False,
|
||||
should_stop: Callable[[], bool] = lambda: False,
|
||||
recapture: bool = False,
|
||||
) -> list[MediaOutcome]:
|
||||
"""Download every media item of one post; return per-item outcomes.
|
||||
|
||||
@@ -212,6 +213,11 @@ class PatreonDownloader:
|
||||
re-checks the budget between posts), so we honour the deadline mid-post
|
||||
and return the items done so far — the rest re-fetch next chunk (they
|
||||
were never marked seen). Bounds chunk overrun to one media download.
|
||||
|
||||
`recapture` (#830): don't re-download already-present media, but DO surface
|
||||
on-disk media as `skipped_disk` (with its path) even when the seen-ledger
|
||||
would tier-1 skip it — so the engine can backfill source_filehash for
|
||||
inline-image localization. Genuinely-missing seen media is NOT refetched.
|
||||
"""
|
||||
post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post)
|
||||
outcomes: list[MediaOutcome] = []
|
||||
@@ -221,7 +227,10 @@ class PatreonDownloader:
|
||||
break
|
||||
try:
|
||||
outcomes.append(
|
||||
self._download_one(post, media, post_dir, artist_slug, i, is_seen)
|
||||
self._download_one(
|
||||
post, media, post_dir, artist_slug, i, is_seen,
|
||||
recapture=recapture,
|
||||
)
|
||||
)
|
||||
except Exception as exc: # resilient: isolate one item's failure
|
||||
log.warning(
|
||||
@@ -243,9 +252,15 @@ class PatreonDownloader:
|
||||
artist_slug: str,
|
||||
index: int,
|
||||
is_seen: Callable[[object], bool],
|
||||
*,
|
||||
recapture: bool = False,
|
||||
) -> MediaOutcome:
|
||||
# tier-1: seen ledger (injected; no DB here).
|
||||
if is_seen(media):
|
||||
# tier-1: seen ledger (injected; no DB here). In recapture mode we DON'T
|
||||
# short-circuit here — we fall through to the disk check so an on-disk
|
||||
# seen file is surfaced as skipped_disk (with its path) for source_filehash
|
||||
# backfill; a seen file that's NOT on disk is left alone (not refetched).
|
||||
seen = is_seen(media)
|
||||
if seen and not recapture:
|
||||
return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
|
||||
|
||||
nn = f"{index:02d}"
|
||||
@@ -265,6 +280,12 @@ class PatreonDownloader:
|
||||
media=media, status="skipped_disk", path=existing, error=None
|
||||
)
|
||||
|
||||
# recapture: a seen item that isn't on disk is NOT re-downloaded (that's
|
||||
# recovery's job) — recapture only re-grabs post text + localizes existing
|
||||
# files. Returns skipped_seen so the run-of-seen / counts stay consistent.
|
||||
if seen:
|
||||
return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
|
||||
|
||||
post_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Pace real downloads only (the skips above already returned). plan #703.
|
||||
|
||||
@@ -70,6 +70,10 @@ class SourceRecord:
|
||||
# plan #697: a running deep-walk that bypasses the Patreon seen-ledger
|
||||
# (recovery) vs. a normal backfill. Lets the badge label it "Recovering".
|
||||
backfill_bypass_seen: bool
|
||||
# #830: a running deep-walk in RECAPTURE mode (re-grab post bodies/links +
|
||||
# localize on-disk inline images, no media re-download). Lets the badge label
|
||||
# it "Recapturing".
|
||||
backfill_recapture: bool
|
||||
# plan #704: cumulative posts processed across the walk's chunks — live
|
||||
# progress for the badge.
|
||||
backfill_posts: int
|
||||
@@ -94,6 +98,7 @@ class SourceRecord:
|
||||
"backfill_state": self.backfill_state,
|
||||
"backfill_chunks": self.backfill_chunks,
|
||||
"backfill_bypass_seen": self.backfill_bypass_seen,
|
||||
"backfill_recapture": self.backfill_recapture,
|
||||
"backfill_posts": self.backfill_posts,
|
||||
}
|
||||
|
||||
@@ -171,6 +176,7 @@ class SourceService:
|
||||
backfill_state=co.get("_backfill_state"),
|
||||
backfill_chunks=int(co.get("_backfill_chunks", 0)),
|
||||
backfill_bypass_seen=bool(co.get("_backfill_bypass_seen")),
|
||||
backfill_recapture=bool(co.get("_backfill_recapture")),
|
||||
backfill_posts=int(co.get("_backfill_posts", 0)),
|
||||
)
|
||||
|
||||
@@ -359,6 +365,34 @@ class SourceService:
|
||||
await self.session.commit()
|
||||
return await self._row_to_record(source)
|
||||
|
||||
async def start_recapture(self, source_id: int) -> SourceRecord:
|
||||
"""#830: arm a RECAPTURE walk — a backfill that re-grabs EVERY post's body
|
||||
+ external links (detail-fetching empty bodies) and localizes already-on-
|
||||
disk inline images, WITHOUT re-downloading media. Reuses the entire #693
|
||||
backfill state machine plus a `_backfill_recapture` flag that flips
|
||||
download mode to recapture. Distinct from recovery (which re-downloads the
|
||||
whole source); the two flags are mutually exclusive, so arming recapture
|
||||
clears bypass_seen. Clears prior cursor/chunk/stall state so it walks
|
||||
fresh from the top. The flag is cleared on completion (download_service)
|
||||
and on stop. Recapture is Patreon-only (the native ingester's post-record
|
||||
capture); inert elsewhere. The UI gates the action to Patreon sources."""
|
||||
source = (await self.session.execute(
|
||||
select(Source).where(Source.id == source_id)
|
||||
)).scalar_one_or_none()
|
||||
if source is None:
|
||||
raise LookupError(f"source id={source_id} not found")
|
||||
co = dict(source.config_overrides or {})
|
||||
co["_backfill_state"] = "running"
|
||||
co["_backfill_recapture"] = True
|
||||
co.pop("_backfill_bypass_seen", None) # mutually exclusive with recovery
|
||||
for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks",
|
||||
"_backfill_posts"):
|
||||
co.pop(k, None)
|
||||
source.config_overrides = co
|
||||
source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
|
||||
await self.session.commit()
|
||||
return await self._row_to_record(source)
|
||||
|
||||
async def stop_backfill(self, source_id: int) -> SourceRecord:
|
||||
"""Plan #693: cancel an in-progress backfill — back to idle/tick mode.
|
||||
Clears the running state + cursor/chunk/stall bookkeeping."""
|
||||
@@ -369,7 +403,8 @@ class SourceService:
|
||||
raise LookupError(f"source id={source_id} not found")
|
||||
co = dict(source.config_overrides or {})
|
||||
for k in ("_backfill_state", "_backfill_cursor", "_backfill_cursor_stalls",
|
||||
"_backfill_chunks", "_backfill_bypass_seen", "_backfill_posts"):
|
||||
"_backfill_chunks", "_backfill_bypass_seen", "_backfill_recapture",
|
||||
"_backfill_posts"):
|
||||
co.pop(k, None)
|
||||
source.config_overrides = co
|
||||
source.backfill_runs_remaining = 0
|
||||
|
||||
Reference in New Issue
Block a user