diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index b33e5b4..7139134 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -33,7 +33,6 @@ from .gallery_dl import ( ErrorType, GalleryDLService, SourceConfig, - parse_last_cursor, ) from .importer import Importer from .patreon_ingester import PatreonIngester @@ -155,7 +154,7 @@ class DownloadService: # timed out with NO progress stays TIMEOUT and feeds phase 3's # stall-guard. Leave RATE_LIMITED alone so the platform-cooldown fires. if in_backfill and dl_result.error_type == ErrorType.TIMEOUT: - new_cursor = parse_last_cursor(dl_result.stdout, dl_result.stderr) + new_cursor = dl_result.cursor # plan #704: structured, not scraped advanced = bool( (new_cursor and new_cursor != overrides.get("_backfill_cursor")) or dl_result.files_downloaded > 0 @@ -439,9 +438,14 @@ class DownloadService: select(DownloadEvent).where(DownloadEvent.id == event_id) )).scalar_one() - run_stats = self.gdl._compute_run_stats( - dl_result.return_code, dl_result.stdout, dl_result.stderr - ) + # plan #704: the native ingester returns structured run_stats; only the + # gallery-dl path needs the regex-over-stdout reconstruction. + if dl_result.run_stats is not None: + run_stats = dict(dl_result.run_stats) + else: + run_stats = self.gdl._compute_run_stats( + dl_result.return_code, dl_result.stdout, dl_result.stderr + ) run_stats["quarantined_count"] = dl_result.files_quarantined stderr_summary = self.gdl._extract_errors_warnings(dl_result.stderr) @@ -528,12 +532,12 @@ class DownloadService: src.backfill_runs_remaining = 0 return - # Did not finish. Patreon checkpoints + resumes via cursor; other - # platforms have no resumable cursor (every chunk re-walks from the - # top), so they advance only by the download archive growing. + # Did not finish. The native ingester checkpoints + resumes via cursor + # (carried structurally on the result, plan #704); gallery-dl platforms + # have no resumable cursor (every chunk re-walks from the top), so they + # advance only by the download archive growing. new_cursor = ( - parse_last_cursor(dl_result.stdout, dl_result.stderr) - if uses_native_ingester(ctx["platform"]) else None + dl_result.cursor if uses_native_ingester(ctx["platform"]) else None ) advanced = bool( (new_cursor and new_cursor != old_cursor) diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py index bab1d63..f4c9967 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -13,7 +13,6 @@ from __future__ import annotations import asyncio import json import logging -import re import subprocess import sys import tempfile @@ -156,6 +155,14 @@ class DownloadResult: duration_seconds: float = 0.0 started_at: str | None = None completed_at: str | None = None + # Plan #704 — structured fields the NATIVE ingester populates directly (None + # on the gallery-dl path, which keeps the regex-over-stdout route). When set, + # phase 3 reads these instead of scraping the text it would otherwise have to + # reconstruct: `run_stats` mirrors _compute_run_stats' shape; `cursor` is the + # backfill checkpoint the ingester knows exactly (no parse_last_cursor). + run_stats: dict | None = None + cursor: str | None = None + posts_processed: int = 0 def _summarize_validation_failures(failures: list[dict]) -> str: @@ -172,20 +179,9 @@ def _summarize_validation_failures(failures: list[dict]) -> str: return f"{n} files quarantined ({top_count}× {top_reason}, mixed)" -# The native Patreon ingester emits its pagination cursor as `Cursor: ` -# — one line per fetched page — into the DownloadResult stdout (mirroring the -# convention gallery-dl used before the #697 cutover, so the backfill lifecycle -# stayed unchanged). The LAST such line is the furthest-progressed page, i.e. -# the resume point. We scan both streams (be defensive) and take the final -# match; download_service checkpoints it as the next chunk's resume_cursor. -# Survives a time-boxed chunk too: the ingester emits the cursor for a page when -# it STARTS it, so an interrupted walk still yields its last cursor. -_CURSOR_RE = re.compile(r"Cursor:\s*(\S+)") - - -def parse_last_cursor(stdout: str, stderr: str) -> str | None: - matches = _CURSOR_RE.findall(f"{stdout or ''}\n{stderr or ''}") - return matches[-1] if matches else None +# (parse_last_cursor was removed in plan #704: the native ingester now carries +# its checkpoint cursor as a structured DownloadResult.cursor field, so there is +# no log text to scrape — and gallery-dl platforms never had a cursor.) class GalleryDLService: diff --git a/backend/app/services/patreon_downloader.py b/backend/app/services/patreon_downloader.py index ed46d25..9dda0f4 100644 --- a/backend/app/services/patreon_downloader.py +++ b/backend/app/services/patreon_downloader.py @@ -124,11 +124,12 @@ def _post_dir_name(post: dict) -> str: class MediaOutcome: """Per-media result of a download_post pass. - status is one of: "downloaded", "skipped_seen", "skipped_disk", "error". - `path` is the final on-disk path for "downloaded" (the actual yt-dlp output - for video), or the path that already existed for "skipped_disk"; None for - "skipped_seen" and (usually) "error". `error` carries the failure reason for - "error", else None. + status is one of: "downloaded", "skipped_seen", "skipped_disk", + "quarantined", "error". `path` is the final on-disk path for "downloaded" + (the actual yt-dlp output for video), the path that already existed for + "skipped_disk", or the _quarantine destination for "quarantined"; None for + "skipped_seen" and (usually) "error". `error` carries the failure/validation + reason for "error"/"quarantined", else None. """ media: object # MediaItem (avoid importing the name for a bare annotation) @@ -253,10 +254,13 @@ class PatreonDownloader: out_path = Path(out_path) else: out_path = self._fetch_get(media.url, media_path) - invalid = self._validate_path(out_path, artist_slug) - if invalid is not None: + reason, quarantine_dest = self._validate_path(out_path, artist_slug) + if reason is not None: + # Quarantined (corrupt/invalid) — distinct from a download error + # so the run can report a real files_quarantined count + paths. return MediaOutcome( - media=media, status="error", path=None, error=invalid + media=media, status="quarantined", + path=quarantine_dest, error=reason, ) self._write_sidecar(post, out_path) @@ -358,23 +362,29 @@ class PatreonDownloader: # -- validation -------------------------------------------------------- - def _validate_path(self, path: Path, artist_slug: str) -> str | None: - """Validate a freshly-written file; quarantine + return reason if bad. + def _validate_path( + self, path: Path, artist_slug: str + ) -> tuple[str | None, Path | None]: + """Validate a freshly-written file; quarantine if bad. Mirrors gallery_dl._validate_and_quarantine: fail-open for unknown - formats, move the corrupt file to _quarantine//patreon, and return - the failure reason string (None when ok / not validatable / disabled). + formats, move the corrupt file to _quarantine//patreon. Returns + `(reason, quarantine_dest)` when quarantined (dest is the original path if + the move itself failed), else `(None, None)` (ok / not validatable / + disabled). plan #704: the dest is surfaced so the run reports a real + quarantined-paths list instead of a blank count. """ if not self._validate or not is_validatable(path): - return None + return None, None try: result = validate_file(path) except Exception as exc: log.warning("Validator raised on %s: %s", path, exc) - return None + return None, None if result.ok: - return None + return None, None quarantine_root = self.images_root / "_quarantine" / artist_slug / "patreon" + dest = path try: quarantine_root.mkdir(parents=True, exist_ok=True) dest = quarantine_root / path.name @@ -385,7 +395,8 @@ class PatreonDownloader: shutil.move(str(path), str(dest)) except OSError as exc: log.error("Failed to quarantine %s: %s. File left in place.", path, exc) - return result.reason or "validation failed" + dest = path + return (result.reason or "validation failed"), dest # -- sidecar ----------------------------------------------------------- diff --git a/backend/app/services/patreon_ingester.py b/backend/app/services/patreon_ingester.py index 0a40fb6..07186b7 100644 --- a/backend/app/services/patreon_ingester.py +++ b/backend/app/services/patreon_ingester.py @@ -156,8 +156,12 @@ class PatreonIngester: start = time.monotonic() log_lines: list[str] = [] written: list[str] = [] + quarantined_paths: list[str] = [] downloaded = 0 errors = 0 + quarantined = 0 + skipped_count = 0 + posts_processed = 0 consecutive_seen = 0 emitted_cursor: str | None = None reached_bottom = False @@ -168,14 +172,17 @@ class PatreonIngester: *, 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=0, - quarantined_paths=[], + files_quarantined=quarantined, + quarantined_paths=list(quarantined_paths), written_paths=written, stdout="\n".join(log_lines), stderr="", @@ -183,18 +190,28 @@ class PatreonIngester: 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, + }, ) 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). + # 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: - log_lines.append(f"Cursor: {page_cursor}") emitted_cursor = page_cursor # Time-box check at the post boundary (coarse, like a gallery-dl @@ -203,6 +220,7 @@ class PatreonIngester: budget_hit = True break + posts_processed += 1 media = self.client.extract_media(post, included) if not media: continue @@ -236,9 +254,20 @@ class PatreonIngester: # 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)) + 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. + quarantined += 1 + if outcome.path is not None: + quarantined_paths.append(str(outcome.path)) + consecutive_seen = 0 elif outcome.status == "error": errors += 1 # An error neither advances nor resets the run-of-seen. @@ -262,9 +291,12 @@ class PatreonIngester: if errors: log_lines.append(f"{errors} media item(s) failed") + if quarantined: + log_lines.append(f"{quarantined} media item(s) quarantined (invalid)") log_lines.append( f"Patreon ingest ({mode}): {downloaded} downloaded, " - f"{errors} error(s)" + f"{skipped_count} skipped, {quarantined} quarantined, " + f"{errors} error(s), {posts_processed} post(s)" + (", reached end" if reached_bottom else "") + (", time-boxed" if budget_hit else "") ) diff --git a/tests/test_download_service.py b/tests/test_download_service.py index 0cd6cbe..4d59fbb 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -59,7 +59,7 @@ def _make_jpg(path: Path, split: str = "h"): def _make_fake_dl_result( *, success=True, written_paths=None, quarantined_paths=None, files_downloaded=0, error_type=None, error_message=None, - stdout="", stderr="", + stdout="", stderr="", cursor=None, run_stats=None, posts_processed=0, ): return SimpleNamespace( success=success, @@ -78,6 +78,11 @@ def _make_fake_dl_result( duration_seconds=1.23, started_at="2026-05-20T14:00:00+00:00", completed_at="2026-05-20T14:01:00+00:00", + # plan #704: native ingester now returns the cursor + run_stats + # structurally (no more stderr `Cursor:` scraping). + cursor=cursor, + run_stats=run_stats, + posts_processed=posts_processed, ) @@ -505,7 +510,7 @@ async def test_backfill_chunk_progress_advances_cursor( svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result( success=False, written_paths=[], files_downloaded=0, - stderr="[patreon][debug] Cursor: 03:PAGE2:xyz\n", + cursor="03:PAGE2:xyz", )) await svc.download_source(source.id) @@ -534,7 +539,7 @@ async def test_backfill_state_running_selects_backfill_mode_and_resumes( svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result( success=False, written_paths=[], - stderr="[patreon][debug] Cursor: 03:RESUME2:next\n", + cursor="03:RESUME2:next", )) await svc.download_source(source.id) @@ -587,7 +592,7 @@ async def test_backfill_cap_exhaustion_stalls( svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result( success=False, written_paths=[], - stderr="[patreon][debug] Cursor: 03:MORE:left\n", + cursor="03:MORE:left", )) await svc.download_source(source.id) @@ -623,7 +628,7 @@ async def test_backfill_stuck_guard_stalls_after_two_no_advance( "config_overrides": {"_backfill_state": "running", "_backfill_cursor": "stuck"}, "backfill_runs_remaining": 5, } - stuck = _make_fake_dl_result(success=False, stderr="Cursor: stuck\n") + stuck = _make_fake_dl_result(success=False, cursor="stuck") await svc._apply_backfill_lifecycle(ctx, stuck) await db.commit() @@ -661,7 +666,7 @@ async def test_backfill_timeout_chunk_reclassified_to_ok( success=False, files_downloaded=3, error_type=ErrorType.TIMEOUT, error_message="Download timed out", - stderr="[patreon][debug] Cursor: 03:NEXT:page\n", + cursor="03:NEXT:page", )) event_id = await svc.download_source(source.id) diff --git a/tests/test_gallery_dl_service.py b/tests/test_gallery_dl_service.py index be06108..76f0997 100644 --- a/tests/test_gallery_dl_service.py +++ b/tests/test_gallery_dl_service.py @@ -8,7 +8,6 @@ from backend.app.services.gallery_dl import ( ErrorType, GalleryDLService, SourceConfig, - parse_last_cursor, ) @@ -316,28 +315,6 @@ def test_categorize_tier_limited_wins_over_partial(gdl): return_code=1, stdout=stdout, stderr=stderr, ) assert etype == ErrorType.TIER_LIMITED - - -# --- Cursor parsing (plan #689 / native ingester #697) --------------------- -# parse_last_cursor is retained post-cutover: the native Patreon ingester emits -# `Cursor: ` per page into DownloadResult.stdout and download_service -# checkpoints the last one. (The gallery-dl Patreon `files`/`cursor` config and -# the Mux yt-dlp Referer block were removed at the #697 cutover, along with the -# tests that pinned them.) - - -def test_parse_last_cursor_returns_last_match_across_streams(): - # One `Cursor: ` line per page; the last is the furthest-progressed - # resume point. - stderr = ( - "[patreon][debug] Cursor: 03:AAAA:bbb\n" - "[urllib3] GET ...\n" - "[patreon][debug] Cursor: 03:CCCC:ddd\n" - ) - assert parse_last_cursor("", stderr) == "03:CCCC:ddd" - - -def test_parse_last_cursor_scans_stdout_too_and_none_when_absent(): - assert parse_last_cursor("Cursor: 99:ZZZ", "") == "99:ZZZ" - assert parse_last_cursor("no marker here", "still nothing") is None - assert parse_last_cursor("", "") is None +# (The cursor-parsing tests were removed in plan #704: parse_last_cursor is gone +# — the native ingester now carries its checkpoint cursor as a structured +# DownloadResult.cursor field instead of a `Cursor:` log line to scrape.) diff --git a/tests/test_patreon_downloader.py b/tests/test_patreon_downloader.py index f96a6fb..8b5d4ff 100644 --- a/tests/test_patreon_downloader.py +++ b/tests/test_patreon_downloader.py @@ -165,6 +165,22 @@ def test_skip_seen(tmp_path): assert not (tmp_path / "artist-x").exists() +def test_invalid_file_is_quarantined(tmp_path): + """plan #704 (#4): a downloaded file that fails validation is moved to + _quarantine and reported as a 'quarantined' outcome carrying that path — + distinct from a download 'error'.""" + bad_url = "https://cdn.patreon.com/bad.png" + session = _FakeSession({bad_url: b"this is not a valid PNG"}) + dl = _downloader(tmp_path, session=session, validate=True) + outcomes = dl.download_post( + _post(), [_img("bad.png", url=bad_url)], "artist-x" + ) + assert outcomes[0].status == "quarantined" + assert outcomes[0].path is not None + assert "_quarantine" in str(outcomes[0].path) + assert outcomes[0].path.is_file() + + def test_skip_disk(tmp_path): dl = _downloader(tmp_path) post_dir = tmp_path / "artist-x" / "patreon" / "2026-05-01_1001_My Post Title" diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 15b8824..7f40c0c 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -63,11 +63,13 @@ class _FakeClient: class _FakeDownloader: """Stub PatreonDownloader. Honors the injected is_seen (tier-1); media in - `on_disk` report skipped_disk (tier-2); everything else downloads.""" + `on_disk` report skipped_disk (tier-2); media in `quarantine` report + quarantined; everything else downloads.""" - def __init__(self, tmp_path, on_disk=None): + def __init__(self, tmp_path, on_disk=None, quarantine=None): self.tmp_path = tmp_path self.on_disk = set(on_disk or ()) + self.quarantine = set(quarantine or ()) self.download_calls = 0 def download_post(self, post, media_items, artist_slug, *, is_seen): @@ -78,6 +80,10 @@ class _FakeDownloader: elif _ledger_key(m) in self.on_disk: p = self.tmp_path / f"{m.post_id}_{m.filename}" outcomes.append(MediaOutcome(media=m, status="skipped_disk", path=p, error=None)) + elif _ledger_key(m) in self.quarantine: + self.download_calls += 1 + q = self.tmp_path / "_quarantine" / f"{m.post_id}_{m.filename}" + outcomes.append(MediaOutcome(media=m, status="quarantined", path=q, error="bad image")) else: self.download_calls += 1 p = self.tmp_path / f"{m.post_id}_{m.filename}" @@ -137,9 +143,36 @@ async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_ assert result.return_code == 0 assert result.files_downloaded == 2 assert len(result.written_paths) == 2 + # plan #704: structured run_stats carry the real counts. + assert result.run_stats["downloaded_count"] == 2 + assert result.posts_processed == 1 assert _count_ledger(sync_engine, source_id) == 2 +@pytest.mark.asyncio +async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_path): + """plan #704 (#4): a media that fails validation is reported as quarantined — + a real files_quarantined count + paths + run_stats — not a blank 0, and is + NOT written or marked seen.""" + m1, m2 = _media("p1", 1), _media("p1", 2) + client = _FakeClient([(None, [("p1", [m1, m2])])]) + downloader = _FakeDownloader(tmp_path, quarantine={_ledger_key(m2)}) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", + ) + assert result.files_downloaded == 1 # only m1 + assert result.files_quarantined == 1 # m2 + assert len(result.quarantined_paths) == 1 + assert result.run_stats["quarantined_count"] == 1 + assert result.run_stats["downloaded_count"] == 1 + assert len(result.written_paths) == 1 # quarantined NOT written + # Quarantined media is NOT marked seen (a fixed file may be re-fetched). + assert _count_ledger(sync_engine, source_id) == 1 + + @pytest.mark.asyncio async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path): m1 = _media("p1", 1) @@ -206,8 +239,10 @@ async def test_backfill_reaches_bottom_signals_complete(source_id, sync_engine, assert result.return_code == 0 assert result.error_type is None assert result.files_downloaded == 2 - # The page-2 cursor was emitted for checkpointing. - assert "Cursor: CUR2" in result.stdout + # The page-2 cursor is carried structurally for checkpointing (plan #704). + assert result.cursor == "CUR2" + assert result.posts_processed == 2 + assert result.run_stats["downloaded_count"] == 2 @pytest.mark.asyncio @@ -245,7 +280,7 @@ async def test_backfill_budget_cut_returns_partial_with_progress( assert result.return_code == -1 assert result.files_downloaded == 1 # post1 downloaded before the cut assert downloader.download_calls == 1 # post2's body never ran - assert "Cursor: CUR1" in result.stdout + assert result.cursor == "CUR1" # structured checkpoint (plan #704) # --- recovery -------------------------------------------------------------