diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index d463b1e..8dcc923 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -4,7 +4,7 @@ Queues: import — filesystem import path (FC-2) ml — WD14 + SigLIP inference (FC-2; runs in the ml-worker image) thumbnail — image and video thumbnail generation (FC-2) - download — gallery-dl tasks (FC-3) + download — source walks: the native ingesters and gallery-dl (FC-3) scan — periodic source checks (FC-3) — kept separate so long imports don't starve the scheduler maintenance — recovery sweeps, pHash backfill, GPU-queue coordination, etc. diff --git a/backend/app/services/db_helpers.py b/backend/app/services/db_helpers.py index 7325ea2..7d7848a 100644 --- a/backend/app/services/db_helpers.py +++ b/backend/app/services/db_helpers.py @@ -21,7 +21,7 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..models import Source -from .gallery_dl import ErrorType +from .download_result import ErrorType async def get_or_create[T]( diff --git a/backend/app/services/discord_client.py b/backend/app/services/discord_client.py index 19ccf09..d54415f 100644 --- a/backend/app/services/discord_client.py +++ b/backend/app/services/discord_client.py @@ -49,13 +49,13 @@ import logging import re import time from collections.abc import Iterator -from dataclasses import dataclass from datetime import date from urllib.parse import unquote import requests from .native_ingest_common import ( + MediaItem, NativeAuthError, NativeDriftError, NativeIngestError, @@ -256,29 +256,6 @@ def rank_posters(server_name: str | None, owner_id, posters: list[dict]) -> list return ranked -@dataclass -class MediaItem: - """One file of a Discord message. `filename`/`extension` are gallery-dl's - split of the URL; `num` is its 1-based position across the message's files, - which is what names it on disk. - - `media_id` is what the seen-ledger keys on, and it is deliberately NOT - `num`: an edit that removes a file renumbers the ones after it, and a - positional key would then call a different file seen. It is the - attachment's id, or for an embed (which has none) a hash of its URL path — - the query string is a signature that changes on every fetch. `filehash` is - always None; nothing in a signed CDN URL is a content hash.""" - - url: str - filename: str - extension: str - kind: str - post_id: str - num: int - media_id: str - filehash: str | None = None - - class DiscordClient: """Synchronous Discord API v10 read client for one user token.""" diff --git a/backend/app/services/discord_downloader.py b/backend/app/services/discord_downloader.py index c9f542a..cab6db0 100644 --- a/backend/app/services/discord_downloader.py +++ b/backend/app/services/discord_downloader.py @@ -30,10 +30,7 @@ PURE: no DB; the seen-skip is an injected predicate. from __future__ import annotations import json -import logging import re -import time -from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path @@ -42,12 +39,11 @@ import requests from .discord_client import firefox_user_agent, message_text from .native_ingest_common import ( BaseNativeDownloader, - MediaOutcome, + MediaItem, PostRecordOutcome, make_session, ) -log = logging.getLogger(__name__) PLATFORM = "discord" _CONTROL = re.compile("[\x00-\x1f\x7f]") @@ -111,67 +107,19 @@ class DiscordDownloader(BaseNativeDownloader): }), ) - def download_post( - self, - post: dict, - media_items: list, - artist_slug: str, - *, - is_seen: Callable[[object], bool] = lambda m: False, - should_stop: Callable[[], bool] = lambda: False, - recapture: bool = False, - ) -> list[MediaOutcome]: - """Every file of one message; per-file outcomes, one failure isolated.""" - folder = channel_dir(self.images_root, artist_slug, post) - outcomes: list[MediaOutcome] = [] - for media in media_items: - if should_stop(): - break - try: - outcomes.append(self._download_one( - post, media, folder, artist_slug, is_seen, recapture=recapture, - )) - except Exception as exc: # resilient: isolate one item's failure - log.warning( - "Discord media failed (message %s, file %d): %s", - post.get("id"), media.num, exc, - ) - outcomes.append( - MediaOutcome(media=media, status="error", path=None, error=str(exc)) - ) - return outcomes + # The download loop is BaseNativeDownloader's; these seams give it + # gallery-dl's Discord layout. - def _download_one( - self, - post: dict, - media, - folder: Path, - artist_slug: str, - is_seen: Callable[[object], bool], - *, - recapture: bool = False, - ) -> MediaOutcome: - seen = is_seen(media) - if seen and not recapture: - return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) - stem = media_stem(post, media) - path = folder / f"{stem}.{media.extension or _NO_EXTENSION}" - if path.exists(): # tier-2: gallery-dl (or an earlier walk) wrote it - return MediaOutcome(media=media, status="skipped_disk", path=path, error=None) - if seen: # recapture never re-fetches a seen file that is gone - return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) + def _post_folder(self, post: dict, artist_slug: str) -> Path: + return channel_dir(self.images_root, artist_slug, post) - folder.mkdir(parents=True, exist_ok=True) - if self._rate_limit > 0: - time.sleep(self._rate_limit) - out = self._fetch_get(media.url, path) - reason, quarantined = self._validate_path(out, artist_slug, media.url) - if reason is not None: - return MediaOutcome(media=media, status="quarantined", path=quarantined, error=reason) - sidecar = {"category": PLATFORM, "message_id": str(post.get("id") or "")} - sidecar["source_url"] = media.url - (folder / f"{stem}.json").write_text(json.dumps(sidecar, indent=2)) - return MediaOutcome(media=media, status="downloaded", path=out, error=None) + def _media_name(self, post: dict, media: MediaItem, index: int) -> str: + return f"{media_stem(post, media)}.{media.extension or _NO_EXTENSION}" + + def _media_sidecar(self, post: dict, media: MediaItem) -> dict: + # `message_id`, never `id`: `id` outranks it in the post-id chain. + return {"category": PLATFORM, "message_id": str(post.get("id") or ""), + "source_url": media.url} def write_post_record( self, post: dict, artist_slug: str, *, revisit: bool = False, diff --git a/backend/app/services/discord_ingester.py b/backend/app/services/discord_ingester.py index 0e2dff2..c4c105e 100644 --- a/backend/app/services/discord_ingester.py +++ b/backend/app/services/discord_ingester.py @@ -32,23 +32,18 @@ from pathlib import Path from sqlalchemy import select, update from ..models import DiscordFailedMedia, DiscordSeenMedia, Source -from .discord_client import DiscordAPIError, DiscordClient, MediaItem +from .discord_client import DiscordAPIError, DiscordClient from .discord_downloader import DiscordDownloader from .ingest_core import Ingester +from .native_ingest_common import ledger_key log = logging.getLogger(__name__) -_LEDGER_KEY_MAX = 128 # `config_overrides` key: the people whose messages this source takes (user # ids, usernames or display names). Absent or empty takes everyone's. AUTHORS_KEY = "discord_authors" -def _ledger_key(media: MediaItem) -> str: - """`:` — stable across edits (see MediaItem).""" - return f"{media.post_id}:{media.media_id}"[:_LEDGER_KEY_MAX] - - class DiscordIngester(Ingester): """Walk a Discord source's channels, download unseen files, return a `DownloadResult`. `client` / `downloader` are injectable for tests.""" @@ -80,7 +75,7 @@ class DiscordIngester(Ingester): failed_model=DiscordFailedMedia, seen_constraint="uq_discord_seen_media_source_id", failed_constraint="uq_discord_failed_media_source_id", - ledger_key=_ledger_key, + ledger_key=ledger_key, platform="discord", error_base=DiscordAPIError, drift_label="Discord API", diff --git a/backend/app/services/download_backends.py b/backend/app/services/download_backends.py index 74fc88b..ab9c42b 100644 --- a/backend/app/services/download_backends.py +++ b/backend/app/services/download_backends.py @@ -24,7 +24,7 @@ import asyncio from pathlib import Path from .discord_ingester import DiscordIngester -from .gallery_dl import DownloadResult, ErrorType +from .download_result import DownloadResult, ErrorType from .ingest_core import DEFAULT_REVISIT_DAYS from .patreon_ingester import PatreonIngester from .patreon_resolver import extract_vanity, resolve_campaign_id_for_source diff --git a/backend/app/services/download_result.py b/backend/app/services/download_result.py new file mode 100644 index 0000000..252f1eb --- /dev/null +++ b/backend/app/services/download_result.py @@ -0,0 +1,222 @@ +"""What a download walk reports, whichever backend ran it. + +The native ingesters (Patreon, SubscribeStar, Discord) and the gallery-dl +subprocess (HentaiFoundry) both return a `DownloadResult`, classify failures +with the same `ErrorType`s and build `run_stats` through `make_run_stats`, so +phase 3, the source's status chip and the Logs UI read one shape from either. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, field +from enum import StrEnum + + +class ErrorType(StrEnum): + AUTH_ERROR = "auth_error" + RATE_LIMITED = "rate_limited" + NOT_FOUND = "not_found" + ACCESS_DENIED = "access_denied" + NETWORK_ERROR = "network_error" + NO_NEW_CONTENT = "no_new_content" + TIER_LIMITED = "tier_limited" + TIMEOUT = "timeout" + HTTP_ERROR = "http_error" + UNSUPPORTED_URL = "unsupported_url" + VALIDATION_FAILED = "validation_failed" + # Native Patreon ingester only (plan #697): a response parsed as JSON but + # didn't match the JSON:API shape the ingester depends on (missing `data`, + # media lacking `file_name`/`url`, etc.). Distinct from AUTH_ERROR — the fix + # is updating the ingester's field-set/parser, not rotating credentials. The + # contract test guards against silently shipping this. + API_DRIFT = "api_drift" + # Run made real progress (downloaded ≥1 file) but did not finish in the + # subprocess budget. Distinct from UNKNOWN_ERROR — the downstream status + # mapping classifies this as "ok" because the next tick continues. + PARTIAL = "partial" + UNKNOWN_ERROR = "unknown_error" + + +@dataclass +class DownloadResult: + success: bool + url: str + artist_slug: str + platform: str + files_downloaded: int = 0 + files_quarantined: int = 0 + quarantined_paths: list[str] = field(default_factory=list) + written_paths: list[str] = field(default_factory=list) + # Native ingester only: post-only sidecar paths for posts that have NO + # downloadable media (pure-text posts), so the importer can still upsert the + # Post + its body. Empty on the gallery-dl path. Phase 3 imports these via + # 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, str]] = field(default_factory=list) + # Native ingester only: one entry per `written_paths` entry, the backfill + # cursor of the page that file came from. Phase 3 resumes the next chunk + # there when it leaves files unimported at its deadline. Empty on gallery-dl. + written_cursors: list[str | None] = field(default_factory=list) + stdout: str = "" + stderr: str = "" + return_code: int = 0 + error_type: ErrorType | None = None + error_message: str | None = None + 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 + # Plan #708 B1 — the server's Retry-After seconds on a RATE_LIMITED result, so + # the platform cooldown matches the hint instead of a flat default. None when + # unknown (no header, or not a rate-limit failure). + retry_after_seconds: float | None = None + # Native ingester only: marks this walk's fetched media seen in its ledger. + # Phase 3 calls it AFTER the import loop, never before — a file marked seen + # but not yet imported is invisible to every later walk, so a run killed in + # between orphaned it for good (TamadaHeijun's 12PCG post lost 7 of 13 + # images to a stranded run, 2026-09-24). Unmarked, the next walk finds the + # file on disk with no ImageRecord and imports it. Takes `skip_paths`: the + # written paths phase 3 did not import, which stay unmarked for the same + # reason. None on gallery-dl. + mark_seen_after_import: Callable[..., None] | None = None + + +def extract_errors_warnings(stderr: str) -> str: + """Keep only the `[error]`/`[warning]` lines from a captured stderr blob. + + A generic download-log helper (module-level so the native-ingester result + path can shape its logs without reaching through a GalleryDLService instance). + """ + if not stderr: + return "" + kept = [ + line for line in stderr.splitlines() + if "][error]" in line.lower() or "][warning]" in line.lower() + ] + return "\n".join(kept) + + +def truncate_log(text: str, max_bytes: int = 500_000) -> str: + """Cap a captured log to `max_bytes`, eliding the middle (head + tail kept).""" + if not text: + return text + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text + half = max_bytes // 2 + head = encoded[:half].decode("utf-8", errors="ignore") + tail = encoded[-half:].decode("utf-8", errors="ignore") + head_lines = head.count("\n") + tail_lines = tail.count("\n") + total_lines = text.count("\n") + elided = max(0, total_lines - head_lines - tail_lines) + marker = f"\n\n... [{elided} lines elided, {len(encoded) - max_bytes} bytes] ...\n\n" + return head + marker + tail + + +def make_run_stats( + *, + exit_code: int = 0, + downloaded_count: int = 0, + skipped_count: int = 0, + per_item_failures: int = 0, + warning_count: int = 0, + tier_gated_count: int = 0, + quarantined_count: int = 0, + dead_lettered_count: int = 0, +) -> dict: + """The canonical `run_stats` dict shape, in ONE place. + + Both result producers — gallery-dl's `_compute_run_stats` (log scrape) and the + native ingester's per-outcome tally (ingest_core) — build through this so the + key set can't drift between backends. Phase 3 + the Logs UI read these keys. + """ + return { + "exit_code": exit_code, + "downloaded_count": downloaded_count, + "skipped_count": skipped_count, + "per_item_failures": per_item_failures, + "warning_count": warning_count, + "tier_gated_count": tier_gated_count, + "quarantined_count": quarantined_count, + "dead_lettered_count": dead_lettered_count, + } + + +# --- tier-gated classification, shared by BOTH backends --------------------- +# +# These three live together because the native ingester and the gallery-dl +# subprocess must reach the same verdict from the same number. They did not: +# gallery-dl classified TIER_LIMITED while ingest_core counted gated posts and +# threw the count away, so the platforms FC owns reported a paywalled creator as +# a silent one (#874 follow-up). One predicate, spread into both, rather than +# the condition re-derived per backend. + + +def classify_tier_gated(tier_gated_count: int) -> ErrorType | None: + """TIER_LIMITED when a walk saw tier-gated posts and nothing else failed. + + Deliberately NOT conditioned on `downloaded == 0`. A creator whose top-tier + posts we cannot see is tier-limited even in a week we did get their cheaper + ones — the fact the operator needs ("there is content here you are not + paying for") is true either way. gallery-dl has classified it this way since + the paywall-as-"needs attention" complaint (see `_categorize_error`), and the + native path now matches rather than inventing a stricter rule. + + Callers must apply this only AFTER the real error categories (auth, rate + limit, drift, …) have had their turn; tier-gating is the weakest signal and + must never mask a genuine failure. + """ + return ErrorType.TIER_LIMITED if tier_gated_count else None + + +def tier_gated_message(count: int) -> str: + """The one wording for the tier-gated verdict, so the two backends can't + describe the same state differently in the Logs UI.""" + return ( + f"Subscription tier does not grant access to " + f"{count} post{'s' if count != 1 else ''}" + ) + + +# `Source.error_type` doubles as the failure-class chip, and a status of "ok" +# CLEARS it (alembic 0032). TIER_LIMITED breaks that assumption: it rides an +# otherwise-successful run, so without an exemption the chip is wiped the moment +# it is set and `FailingSourcesCard`'s `tier_limited` palette entry can never +# render. Informational classes are the exemption — they describe the source, +# not a failure of the run. +INFORMATIONAL_ERROR_TYPES = frozenset({ErrorType.TIER_LIMITED.value}) + + +def is_informational(error_type) -> bool: + """True for a class that reports a state rather than a failure. Accepts an + ErrorType or the plain string persisted on Source.error_type.""" + return error_type is not None and str(error_type) in INFORMATIONAL_ERROR_TYPES + + +def walk_completed(result: DownloadResult) -> bool: + """Did this walk reach the bottom cleanly? + + The backfill lifecycle's completion test. An informational error_type still + counts as complete: a fully-paywalled creator's backfill DID finish, and + treating it as unfinished re-walks the same wall until the stall counter + trips — the creator we can see least becoming the one we fetch most. + """ + return ( + result.success + and result.return_code == 0 + and (result.error_type is None or is_informational(result.error_type)) + ) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 1b0e948..e5517d0 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -3,8 +3,9 @@ Three-phase pipeline: Phase 1 — DB setup (brief session): load source, mark event running, fetch credentials + settings. - Phase 2 — Execute (no DB connection): call GalleryDLService; on - Patreon campaign-ID failure, resolve vanity and retry once. + Phase 2 — Execute (no DB connection): run the source's backend (a native + ingester, or gallery-dl for HentaiFoundry); on a Patreon + campaign-ID failure, resolve the vanity and retry once. Phase 3 — Import + persist (fresh session): attach_in_place each written file; clean up duplicates; persist rich metadata. """ @@ -26,19 +27,21 @@ from sqlalchemy.orm import joinedload from ..models import Artist, DownloadEvent, Source from .credential_service import CredentialService from .download_backends import run_download, uses_native_ingester -from .gallery_dl import ( - BACKFILL_CHUNK_SECONDS, - BACKFILL_SKIP_VALUE, - TICK_SKIP_VALUE, +from .download_result import ( DownloadResult, ErrorType, - GalleryDLService, - SourceConfig, extract_errors_warnings, is_informational, truncate_log, walk_completed, ) +from .gallery_dl import ( + BACKFILL_CHUNK_SECONDS, + BACKFILL_SKIP_VALUE, + TICK_SKIP_VALUE, + GalleryDLService, + SourceConfig, +) from .importer import Importer from .ingest_core import DEFAULT_REVISIT_DAYS, PHASE3_DEADLINE_SECONDS from .platforms import auth_type_for @@ -47,6 +50,16 @@ from .scheduler_service import set_platform_cooldown log = logging.getLogger(__name__) +def credential_proven(ctx: dict, dl_result) -> bool: + """Did this walk prove the platform's stored credential still works? + + A clean walk made with a credential is the evidence the Verify button looks + for. A walk with any error type proves nothing, tier-limited included: + content hidden behind a tier is also what an expired session looks like.""" + used = bool(ctx.get("cookies_path") or ctx.get("auth_token")) + return used and bool(dl_result.success) and dl_result.error_type is None + + class DownloadService: """Async orchestrator. The Celery task runs `asyncio.run(svc.download_source(N))`. @@ -536,6 +549,10 @@ class DownloadService: ) await self._apply_backfill_lifecycle(ctx, dl_result) await self.async_session.commit() + if credential_proven(ctx, dl_result): + # The credentials page then shows when the account last worked, + # not only when someone last pressed Verify. + await self.cred_service.mark_verified(ctx["platform"]) return event_id @staticmethod diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py index 98ce034..ddf5d0e 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -6,6 +6,9 @@ backend/app/services/gallery_dl.py). FC adaptations: - Validation uses backend/app/services/file_validator.py - Config dir under `/.gallery-dl/`, not a separate config_path - PLATFORM_DEFAULTS scoped to the GS-6 platforms registered in FC-3b + +The result it returns (`DownloadResult`, `ErrorType`) is shared with the native +ingesters and lives in download_result. """ from __future__ import annotations @@ -17,42 +20,22 @@ import subprocess import sys import tempfile import time -from collections.abc import Callable from dataclasses import dataclass, field from datetime import UTC, datetime -from enum import StrEnum from pathlib import Path +from .download_result import ( + DownloadResult, + ErrorType, + classify_tier_gated, + make_run_stats, + tier_gated_message, +) from .file_validator import is_validatable, quarantine_file, validate_file log = logging.getLogger(__name__) -class ErrorType(StrEnum): - AUTH_ERROR = "auth_error" - RATE_LIMITED = "rate_limited" - NOT_FOUND = "not_found" - ACCESS_DENIED = "access_denied" - NETWORK_ERROR = "network_error" - NO_NEW_CONTENT = "no_new_content" - TIER_LIMITED = "tier_limited" - TIMEOUT = "timeout" - HTTP_ERROR = "http_error" - UNSUPPORTED_URL = "unsupported_url" - VALIDATION_FAILED = "validation_failed" - # Native Patreon ingester only (plan #697): a response parsed as JSON but - # didn't match the JSON:API shape the ingester depends on (missing `data`, - # media lacking `file_name`/`url`, etc.). Distinct from AUTH_ERROR — the fix - # is updating the ingester's field-set/parser, not rotating credentials. The - # contract test guards against silently shipping this. - API_DRIFT = "api_drift" - # Run made real progress (downloaded ≥1 file) but did not finish in the - # subprocess budget. Distinct from UNKNOWN_ERROR — the downstream status - # mapping classifies this as "ok" because the next tick continues. - PARTIAL = "partial" - UNKNOWN_ERROR = "unknown_error" - - # Tick mode (routine cron polls): skip ≤20 contiguous already-archived # items, then exit gallery-dl. Established subscription with zero new # content exits in ~30s of HEAD requests instead of walking to the bottom @@ -148,95 +131,6 @@ class SourceConfig: ) -@dataclass -class DownloadResult: - success: bool - url: str - artist_slug: str - platform: str - files_downloaded: int = 0 - files_quarantined: int = 0 - quarantined_paths: list[str] = field(default_factory=list) - written_paths: list[str] = field(default_factory=list) - # Native ingester only: post-only sidecar paths for posts that have NO - # downloadable media (pure-text posts), so the importer can still upsert the - # Post + its body. Empty on the gallery-dl path. Phase 3 imports these via - # 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, str]] = field(default_factory=list) - # Native ingester only: one entry per `written_paths` entry, the backfill - # cursor of the page that file came from. Phase 3 resumes the next chunk - # there when it leaves files unimported at its deadline. Empty on gallery-dl. - written_cursors: list[str | None] = field(default_factory=list) - stdout: str = "" - stderr: str = "" - return_code: int = 0 - error_type: ErrorType | None = None - error_message: str | None = None - 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 - # Plan #708 B1 — the server's Retry-After seconds on a RATE_LIMITED result, so - # the platform cooldown matches the hint instead of a flat default. None when - # unknown (no header, or not a rate-limit failure). - retry_after_seconds: float | None = None - # Native ingester only: marks this walk's fetched media seen in its ledger. - # Phase 3 calls it AFTER the import loop, never before — a file marked seen - # but not yet imported is invisible to every later walk, so a run killed in - # between orphaned it for good (TamadaHeijun's 12PCG post lost 7 of 13 - # images to a stranded run, 2026-09-24). Unmarked, the next walk finds the - # file on disk with no ImageRecord and imports it. Takes `skip_paths`: the - # written paths phase 3 did not import, which stay unmarked for the same - # reason. None on gallery-dl. - mark_seen_after_import: Callable[..., None] | None = None - - -def extract_errors_warnings(stderr: str) -> str: - """Keep only the `[error]`/`[warning]` lines from a captured stderr blob. - - A generic download-log helper (module-level so the native-ingester result - path can shape its logs without reaching through a GalleryDLService instance). - """ - if not stderr: - return "" - kept = [ - line for line in stderr.splitlines() - if "][error]" in line.lower() or "][warning]" in line.lower() - ] - return "\n".join(kept) - - -def truncate_log(text: str, max_bytes: int = 500_000) -> str: - """Cap a captured log to `max_bytes`, eliding the middle (head + tail kept).""" - if not text: - return text - encoded = text.encode("utf-8") - if len(encoded) <= max_bytes: - return text - half = max_bytes // 2 - head = encoded[:half].decode("utf-8", errors="ignore") - tail = encoded[-half:].decode("utf-8", errors="ignore") - head_lines = head.count("\n") - tail_lines = tail.count("\n") - total_lines = text.count("\n") - elided = max(0, total_lines - head_lines - tail_lines) - marker = f"\n\n... [{elided} lines elided, {len(encoded) - max_bytes} bytes] ...\n\n" - return head + marker + tail - - def _summarize_validation_failures(failures: list[dict]) -> str: if not failures: return "Validation failed" @@ -251,106 +145,6 @@ def _summarize_validation_failures(failures: list[dict]) -> str: return f"{n} files quarantined ({top_count}× {top_reason}, mixed)" -# (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.) - - -def make_run_stats( - *, - exit_code: int = 0, - downloaded_count: int = 0, - skipped_count: int = 0, - per_item_failures: int = 0, - warning_count: int = 0, - tier_gated_count: int = 0, - quarantined_count: int = 0, - dead_lettered_count: int = 0, -) -> dict: - """The canonical `run_stats` dict shape, in ONE place. - - Both result producers — gallery-dl's `_compute_run_stats` (log scrape) and the - native ingester's per-outcome tally (ingest_core) — build through this so the - key set can't drift between backends. Phase 3 + the Logs UI read these keys. - """ - return { - "exit_code": exit_code, - "downloaded_count": downloaded_count, - "skipped_count": skipped_count, - "per_item_failures": per_item_failures, - "warning_count": warning_count, - "tier_gated_count": tier_gated_count, - "quarantined_count": quarantined_count, - "dead_lettered_count": dead_lettered_count, - } - - -# --- tier-gated classification, shared by BOTH backends --------------------- -# -# These three live together because the native ingester and the gallery-dl -# subprocess must reach the same verdict from the same number. They did not: -# gallery-dl classified TIER_LIMITED while ingest_core counted gated posts and -# threw the count away, so the platforms FC owns reported a paywalled creator as -# a silent one (#874 follow-up). One predicate, spread into both, rather than -# the condition re-derived per backend. - - -def classify_tier_gated(tier_gated_count: int) -> ErrorType | None: - """TIER_LIMITED when a walk saw tier-gated posts and nothing else failed. - - Deliberately NOT conditioned on `downloaded == 0`. A creator whose top-tier - posts we cannot see is tier-limited even in a week we did get their cheaper - ones — the fact the operator needs ("there is content here you are not - paying for") is true either way. gallery-dl has classified it this way since - the paywall-as-"needs attention" complaint (see `_categorize_error`), and the - native path now matches rather than inventing a stricter rule. - - Callers must apply this only AFTER the real error categories (auth, rate - limit, drift, …) have had their turn; tier-gating is the weakest signal and - must never mask a genuine failure. - """ - return ErrorType.TIER_LIMITED if tier_gated_count else None - - -def tier_gated_message(count: int) -> str: - """The one wording for the tier-gated verdict, so the two backends can't - describe the same state differently in the Logs UI.""" - return ( - f"Subscription tier does not grant access to " - f"{count} post{'s' if count != 1 else ''}" - ) - - -# `Source.error_type` doubles as the failure-class chip, and a status of "ok" -# CLEARS it (alembic 0032). TIER_LIMITED breaks that assumption: it rides an -# otherwise-successful run, so without an exemption the chip is wiped the moment -# it is set and `FailingSourcesCard`'s `tier_limited` palette entry can never -# render. Informational classes are the exemption — they describe the source, -# not a failure of the run. -INFORMATIONAL_ERROR_TYPES = frozenset({ErrorType.TIER_LIMITED.value}) - - -def is_informational(error_type) -> bool: - """True for a class that reports a state rather than a failure. Accepts an - ErrorType or the plain string persisted on Source.error_type.""" - return error_type is not None and str(error_type) in INFORMATIONAL_ERROR_TYPES - - -def walk_completed(result: DownloadResult) -> bool: - """Did this walk reach the bottom cleanly? - - The backfill lifecycle's completion test. An informational error_type still - counts as complete: a fully-paywalled creator's backfill DID finish, and - treating it as unfinished re-walks the same wall until the stall counter - trips — the creator we can see least becoming the one we fetch most. - """ - return ( - result.success - and result.return_code == 0 - and (result.error_type is None or is_informational(result.error_type)) - ) - - class GalleryDLService: """Service for executing gallery-dl downloads.""" @@ -728,16 +522,6 @@ class GalleryDLService: tier_gated_count=tier_gated_count, ) - @staticmethod - def _extract_errors_warnings(stderr: str) -> str: - # Thin delegator to the module-level helper (kept for existing callers). - return extract_errors_warnings(stderr) - - @staticmethod - def _truncate_log(text: str, max_bytes: int = 500_000) -> str: - # Thin delegator to the module-level helper (kept for existing callers). - return truncate_log(text, max_bytes) - async def download( self, url: str, diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 2ee9154..4ce96f4 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -37,7 +37,7 @@ from sqlalchemy import delete, func, select, text from sqlalchemy.dialects.postgresql import insert as pg_insert from ..models import ImageRecord -from .gallery_dl import ( +from .download_result import ( DownloadResult, ErrorType, classify_tier_gated, diff --git a/backend/app/services/native_ingest_common.py b/backend/app/services/native_ingest_common.py index 29226b0..b3da1fa 100644 --- a/backend/app/services/native_ingest_common.py +++ b/backend/app/services/native_ingest_common.py @@ -19,6 +19,7 @@ import json import logging import os import time +from collections.abc import Callable from dataclasses import dataclass from datetime import datetime from pathlib import Path @@ -184,6 +185,56 @@ def post_dir_name(post: dict) -> str: return sanitize_segment(raw) +# -- one downloadable file, and its seen-ledger identity ------------------- + +@dataclass +class MediaItem: + """One resolved downloadable file belonging to a post, as every platform + client returns it. + + - `url`: what to fetch. `filename`: the media's own name when the platform + gives one, else the URL basename (never a network call). + - `kind`: the content type, using gallery-dl's `files` names ("images", + "attachments", …) so per-source content_types filter the same way. + - `post_id`: the owning post, so a flattened list stays traceable. + - `filehash`: the 32-hex content hash in a content-addressed CDN URL + (Patreon), else None. + - `media_id`: the platform's stable per-file id where the URL has no hash: + SubscribeStar's gallery id, Discord's attachment id (or, for an embed, a + hash of its URL path; the query string is a signature that changes on + every fetch). Empty when the platform has none. + - `extension` / `num`: Discord only. gallery-dl splits the name from its + extension and numbers files 1-based across the message; both name the + file on disk. + """ + + url: str + filename: str + kind: str + post_id: str + filehash: str | None = None + media_id: str = "" + extension: str = "" + num: int = 0 + + +# The seen-ledger tables' key column width. +_LEDGER_KEY_MAX = 128 + + +def ledger_key(media: MediaItem) -> str: + """A file's identity in the cross-run seen-ledger, stable across walks. + + The content hash when the URL carries one. Otherwise the file is keyed + within its post: by the platform's per-file id, or by filename when there is + none (a Patreon Mux video or a hashless inline image). A positional key is + never used, because an edit that removes a file renumbers the ones after it + and would make a different file read as seen.""" + if media.filehash: + return media.filehash + return f"{media.post_id}:{media.media_id or media.filename}"[:_LEDGER_KEY_MAX] + + # -- per-item download outcomes (shared dataclasses) ----------------------- @dataclass @@ -267,11 +318,16 @@ class Membership: # -- base downloader (shared fetch/validate plumbing) ---------------------- class BaseNativeDownloader: - """Shared download plumbing for native-platform downloaders: the streaming - GET (transient-retry + Range-resume) and file validation/quarantine. Platform - downloaders subclass this and implement `download_post` / `write_post_record` - / the per-media sidecar (and any platform-specific fetch, e.g. Patreon's - Mux/yt-dlp video branch). PURE: no DB; the seen-skip is an injected predicate. + """Shared download plumbing for native-platform downloaders: one post's + files (`download_post`, with the seen/disk skips and the time-box), the + streaming GET (transient-retry + Range-resume), and file + validation/quarantine. + + A platform downloader subclasses this, implements `write_post_record`, and + overrides only the seams where its layout differs: `_post_folder`, + `_media_name`, `_media_sidecar`, and for Patreon's Mux/yt-dlp video, + `_existing_output` and `_fetch_media`. PURE: no DB; the seen-skip is an + injected predicate. """ def __init__( @@ -383,18 +439,118 @@ class BaseNativeDownloader: ) return reason, (dest or path) - # -- sidecar (per-media, minimal) -------------------------------------- + # -- one post's files --------------------------------------------------- - def _write_minimal_sidecar( - self, post: dict, media_path: Path, *, source_url: str | None = None - ) -> Path: - """Post-first per-media sidecar (#856): image identity ONLY - (category/id/source_url). The post body/links live solely in _post.json.""" - data: dict = {"category": self.platform, "id": str(post.get("id") or "")} - if source_url: - data["source_url"] = source_url + def download_post( + self, + post: dict, + media_items: list[MediaItem], + artist_slug: str, + *, + is_seen: Callable[[MediaItem], bool] = lambda m: False, + should_stop: Callable[[], bool] = lambda: False, + recapture: bool = False, + ) -> list[MediaOutcome]: + """Download every file of one post and return one outcome per file. + + - Two-tier skip: the injected seen-ledger predicate first, then the disk. + - `should_stop()` is polled before each file, so a file-dense post + cannot run a time-boxed chunk far past its deadline. The files not + reached are not marked seen and are fetched by the next chunk. + - One file's failure is an "error" outcome for that file; the rest + proceed. + - `recapture` re-reads what is on disk without re-downloading: a seen + file that is on disk comes back `skipped_disk` with its path (so the + engine can backfill its source filehash), and a seen file that is gone + is left alone, since refetching is recovery's job. + """ + folder = self._post_folder(post, artist_slug) + outcomes: list[MediaOutcome] = [] + for index, media in enumerate(media_items, start=1): + if should_stop(): + break + try: + outcomes.append(self._download_one( + post, media, folder, artist_slug, index, is_seen, + recapture=recapture, + )) + except Exception as exc: # resilient: isolate one item's failure + log.warning( + "%s media failed (post %s, item %d): %s", + self.platform, post.get("id"), index, exc, + ) + outcomes.append( + MediaOutcome(media=media, status="error", path=None, error=str(exc)) + ) + return outcomes + + def _download_one( + self, + post: dict, + media: MediaItem, + folder: Path, + artist_slug: str, + index: int, + is_seen: Callable[[MediaItem], bool], + *, + recapture: bool = False, + ) -> MediaOutcome: + seen = is_seen(media) + if seen and not recapture: + return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) + path = folder / self._media_name(post, media, index) + existing = self._existing_output(path, media) + if existing is not None: + return MediaOutcome(media=media, status="skipped_disk", path=existing, error=None) + if seen: + return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) + + folder.mkdir(parents=True, exist_ok=True) + # Real downloads are paced; the skips above have already returned. + if self._rate_limit > 0: + time.sleep(self._rate_limit) + return self._fetch_media(post, media, path, artist_slug) + + # -- platform seams ---------------------------------------------------- + # Where a platform lays its files out. Each matches the layout gallery-dl + # wrote, so files from before a platform moved to its native downloader + # are found on disk instead of fetched again. + + def _post_folder(self, post: dict, artist_slug: str) -> Path: + return self.images_root / artist_slug / self.platform / post_dir_name(post) + + def _media_name(self, post: dict, media: MediaItem, index: int) -> str: + return sanitize_segment(f"{index:02d}_{media.filename}") + + def _existing_output(self, path: Path, media: MediaItem) -> Path | None: + """The file already on disk for this media, or None.""" + return path if path.exists() else None + + def _fetch_media( + self, post: dict, media: MediaItem, path: Path, artist_slug: str, + ) -> MediaOutcome: + """Fetch one file, validate it, and write its sidecar.""" + out = self._fetch_get(media.url, path) + reason, quarantined = self._validate_path(out, artist_slug, media.url) + if reason is not None: + # Distinct from an error, so the run reports a real + # files_quarantined count and the paths. + return MediaOutcome(media=media, status="quarantined", path=quarantined, error=reason) + self._write_media_sidecar(post, out, media) + return MediaOutcome(media=media, status="downloaded", path=out, error=None) + + def _media_sidecar(self, post: dict, media: MediaItem) -> dict: + """The per-file sidecar's content: the file's identity only. The post's + body, title and links live in its post record alone, so they are never + copied beside every file.""" + return {"category": self.platform, "id": str(post.get("id") or ""), + "source_url": media.url} + + def _write_media_sidecar(self, post: dict, media_path: Path, media: MediaItem) -> Path: + """Write the per-file sidecar beside the file, named like it minus the + extension, which is what `find_sidecar` pairs first.""" sidecar_path = media_path.with_suffix(".json") - sidecar_path.write_text(json.dumps(data, indent=2)) + sidecar_path.write_text(json.dumps(self._media_sidecar(post, media), indent=2)) return sidecar_path # --- membership status vocabulary (#387) ------------------------------------ diff --git a/backend/app/services/patreon_client.py b/backend/app/services/patreon_client.py index e732e90..7e95378 100644 --- a/backend/app/services/patreon_client.py +++ b/backend/app/services/patreon_client.py @@ -44,7 +44,6 @@ import logging import re import time from collections.abc import Iterator -from dataclasses import dataclass from html import unescape from pathlib import Path from urllib.parse import parse_qs, unquote, urlsplit @@ -55,6 +54,7 @@ from ..utils.paths import filehash_from_url from ..utils.prosemirror import post_body_html from .native_ingest_common import ( _MAX_429_RETRIES, + MediaItem, Membership, NativeAuthError, NativeDriftError, @@ -142,31 +142,6 @@ class PatreonDriftError(PatreonAPIError, NativeDriftError): """ -@dataclass -class MediaItem: - """One resolved downloadable item belonging to a post. - - Fields: - url — the CDN/download URL to fetch. - filename — media `file_name` when present; otherwise the URL basename - (NEVER a network call). Bounded/sane extension. - kind — one of: "images", "image_large", "attachments", "postfile", - "content". Mirrors gallery-dl's `files` content-type names so - the later step can honor the same per-source content_types. - filehash — the 32-char hex (MD5) segment from the CDN URL, or None if - the URL carries no such segment. Used for in-post dedup and - the cross-run seen-ledger. - post_id — the owning post's id (so a flattened media list stays - traceable to its post). - """ - - url: str - filename: str - kind: str - filehash: str | None - post_id: str - - # Patreon's CDN URLs carry a base64 JSON segment of render options, such as # eyJ3Ijo2MjB9 for {"w":620}. A locked preview is rendered blurred and says so # with a "b" (blur) option. A post the account can view can still embed media diff --git a/backend/app/services/patreon_downloader.py b/backend/app/services/patreon_downloader.py index c588b3d..213922f 100644 --- a/backend/app/services/patreon_downloader.py +++ b/backend/app/services/patreon_downloader.py @@ -43,10 +43,10 @@ from .native_ingest_common import ( _BACKOFF_CAP_SECONDS, _MAX_MEDIA_RETRIES, BaseNativeDownloader, + MediaItem, MediaOutcome, PostRecordOutcome, post_dir_name, - sanitize_segment, ) log = logging.getLogger(__name__) @@ -102,139 +102,33 @@ class PatreonDownloader(BaseNativeDownloader): # None in unit tests / when enrichment isn't wanted. self._content_fetcher = content_fetcher - # -- public ------------------------------------------------------------ + # The download loop, layout and per-file sidecar are BaseNativeDownloader's. + # Patreon adds one branch: a Mux/HLS video is fetched with yt-dlp, which may + # remux it to a container of its own choosing. - def download_post( - self, - post: dict, - media_items: list, - artist_slug: str, - *, - 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. + def _existing_output(self, path: Path, media: MediaItem) -> Path | None: + if path.exists(): + return path + if _is_video_url(media.url): + # A remux lands at a different extension; honour one already there. + return self._existing_video_output(path) + return None - Builds the post directory, iterates media (1-based NN), applies the - two-tier skip (injected is_seen, then disk), downloads (plain GET or - yt-dlp for video), writes the sidecar for each freshly-downloaded item, - validates, and returns outcomes. Resilient: one media's failure yields - an "error" outcome for that item; the rest proceed. - - `should_stop()` is polled BEFORE each media item: a media-dense post can - otherwise run a backfill chunk far past its time-box (the engine only - 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] = [] - - for i, media in enumerate(media_items, start=1): - if should_stop(): - break - try: - outcomes.append( - 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( - "Patreon media failed (post %s, item %d): %s", - post.get("id"), i, exc, - ) - outcomes.append( - MediaOutcome(media=media, status="error", path=None, error=str(exc)) - ) - return outcomes - - # -- per-item ---------------------------------------------------------- - - def _download_one( - self, - post: dict, - media, - post_dir: Path, - artist_slug: str, - index: int, - is_seen: Callable[[object], bool], - *, - recapture: bool = False, + def _fetch_media( + self, post: dict, media: MediaItem, path: Path, artist_slug: str, ) -> MediaOutcome: - # 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}" - final_name = sanitize_segment(f"{nn}_{media.filename}") - media_path = post_dir / final_name - - # tier-2: already on disk. - if media_path.exists(): + if not _is_video_url(media.url): + return super()._fetch_media(post, media, path, artist_slug) + out_path = self._run_ytdlp(media.url, path, _VIDEO_HEADERS) + if out_path is None or not Path(out_path).exists(): return MediaOutcome( - media=media, status="skipped_disk", path=media_path, error=None + media=media, status="error", path=None, error="yt-dlp produced no output", ) - # Video may land at a different extension; honor a pre-existing remux. - if _is_video_url(media.url): - existing = self._existing_video_output(media_path) - if existing is not None: - return MediaOutcome( - 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. - if self._rate_limit > 0: - time.sleep(self._rate_limit) - - if _is_video_url(media.url): - out_path = self._run_ytdlp(media.url, media_path, _VIDEO_HEADERS) - if out_path is None or not Path(out_path).exists(): - return MediaOutcome( - media=media, - status="error", - path=None, - error="yt-dlp produced no output", - ) - out_path = Path(out_path) - else: - out_path = self._fetch_get(media.url, media_path) - reason, quarantine_dest = self._validate_path( - out_path, artist_slug, media.url - ) - 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="quarantined", - path=quarantine_dest, error=reason, - ) - - self._write_sidecar(post, out_path, source_url=media.url) + out_path = Path(out_path) + self._write_media_sidecar(post, out_path, media) return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) # -- video (Mux/HLS via yt-dlp) ---------------------------------------- - # The plain-GET streaming path (_fetch_get / _fetch_to_file) and - # _validate_path are inherited from BaseNativeDownloader. def _run_ytdlp(self, url: str, dest: Path, headers: dict) -> Path | None: """Invoke yt-dlp to fetch a Mux/HLS stream to (around) `dest`. @@ -316,43 +210,12 @@ class PatreonDownloader(BaseNativeDownloader): return cand return None - # -- sidecar ----------------------------------------------------------- - - def _write_sidecar( - self, post: dict, media_path: Path, *, source_url: str | None = None + def _write_post_record_file( + self, post: dict, sidecar_path: Path, *, detail_fetch: bool = True, ) -> Path: - """Write the per-media sidecar next to `media_path` — post-first (#856). - - On the native ingester the POST-RECORD (`write_post_record` → `_post.json`) - is the sole writer of the post body/links/metadata, captured once per post - BEFORE its media in the walk. So the per-media sidecar carries ONLY - image-specific identity: `category` (platform) + `id` (external_post_id, to - link provenance to the right Post) + this file's `source_url` (its CDN URL, - #830 Phase 2 — the importer persists its filehash so the body's inline - `` remaps to the local copy at render time). No body: writing it - next to every image duplicated the post body N+1× and risked divergence - (milestone #67). The importer skips post fields for these (post_first). - """ - return self._write_sidecar_data( - post, media_path.with_suffix(".json"), source_url=source_url, - minimal=True, - ) - - def _write_sidecar_data( - self, post: dict, sidecar_path: Path, *, source_url: str | None = None, - minimal: bool = False, detail_fetch: bool = True, - ) -> Path: - """Serialize the post's metadata to `sidecar_path`. The post-only record - (`write_post_record`) writes the FULL post (body/title/date/url); the - per-media sidecar (`_write_sidecar`, minimal=True) writes only image - identity (category/id/source_url) — post-first (#856). `source_url` is set - only for the per-media sidecar — a media-less post has no source file.""" - if minimal: - data = {"category": "patreon", "id": str(post.get("id") or "")} - if source_url: - data["source_url"] = source_url - sidecar_path.write_text(json.dumps(data, indent=2)) - return sidecar_path + """Serialize the full post record (body, title, date, url) to + `sidecar_path`. The per-file sidecars carry only the file's identity + (BaseNativeDownloader._media_sidecar).""" attrs = post.get("attributes") or {} title = attrs.get("title") # Resolve the body HTML from the feed attrs: legacy flat `content`, else @@ -387,8 +250,6 @@ class PatreonDownloader(BaseNativeDownloader): "published_at": published if isinstance(published, str) else None, "url": url if isinstance(url, str) else None, } - if source_url: - data["source_url"] = source_url sidecar_path.write_text(json.dumps(data, indent=2)) return sidecar_path @@ -434,10 +295,10 @@ class PatreonDownloader(BaseNativeDownloader): ) post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post) post_dir.mkdir(parents=True, exist_ok=True) - path = self._write_sidecar_data( + path = self._write_post_record_file( post, post_dir / "_post.json", detail_fetch=not revisit, ) - # _write_sidecar_data has by now memoized any detail-fetched body onto + # _write_post_record_file has by now memoized any detail-fetched body onto # post["attributes"]["content"], so re-read it for the FINAL char count. body = (post.get("attributes") or {}).get("content") body_chars = len(body) if isinstance(body, str) else 0 diff --git a/backend/app/services/patreon_ingester.py b/backend/app/services/patreon_ingester.py index 2c1e756..8c36088 100644 --- a/backend/app/services/patreon_ingester.py +++ b/backend/app/services/patreon_ingester.py @@ -6,9 +6,8 @@ media → tiered skip → download → mark-seen / record-failures / checkpoint- 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. +failure mapping. `download_service.download_source` calls `PatreonIngester.run`; +the seen-ledger key is the shared `native_ingest_common.ledger_key`. Three modes (selected by `download_service` from `config_overrides` state): - tick — newest→oldest, skip seen (tier-1 ledger + tier-2 disk), early-out @@ -36,38 +35,19 @@ from pathlib import Path from ..models import PatreonFailedMedia, PatreonSeenMedia from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester -from .patreon_client import MediaItem, PatreonAPIError, PatreonClient +from .native_ingest_common import ledger_key +from .patreon_client import PatreonAPIError, PatreonClient from .patreon_downloader import PatreonDownloader from .patreon_resolver import extract_vanity, resolve_campaign_id_for_source __all__ = [ "DEAD_LETTER_THRESHOLD", "PatreonIngester", - "_ledger_key", "verify_patreon_credential", ] 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 - - -def _ledger_key(media: MediaItem) -> str: - """Stable per-media identity for the cross-run seen-ledger. - - A Patreon CDN URL carries a 32-char MD5 (`media.filehash`) — that is the - natural key. Some media have none: Mux/HLS video (`stream.mux.com`, no - content hash at discovery) and the odd inline-content `` pointing at a - hashless URL. The plan calls the video case the ``video::`` - sentinel; `MediaItem` carries no media_id, so the post-scoped filename is the - stable proxy. Bounded to the column width. - """ - if media.filehash: - return media.filehash - return f"{media.post_id}:{media.filename}"[:_LEDGER_KEY_MAX] - class PatreonIngester(Ingester): """Walk a Patreon campaign's posts, download unseen media, return a @@ -124,7 +104,7 @@ class PatreonIngester(Ingester): failed_model=PatreonFailedMedia, seen_constraint="uq_patreon_seen_media_source_id", failed_constraint="uq_patreon_failed_media_source_id", - ledger_key=_ledger_key, + ledger_key=ledger_key, platform="patreon", error_base=PatreonAPIError, # API_DRIFT message phrasing; the base Ingester._failure_result owns diff --git a/backend/app/services/platform_lock.py b/backend/app/services/platform_lock.py index 89f36ca..2094759 100644 --- a/backend/app/services/platform_lock.py +++ b/backend/app/services/platform_lock.py @@ -17,16 +17,16 @@ import logging import redis from ..config import get_config +from .download_backends import NATIVE_INGESTER_PLATFORMS log = logging.getLogger(__name__) -# Platforms walked one-at-a-time. gallery-dl platforms are intentionally NOT -# here: each runs as a self-pacing subprocess and they're lower-volume. The -# native-ingester platforms are serialized (one paced scrape/API walk at a time). -# Add a platform here to cap it to a single concurrent walk. Discord most of -# all: every source walks on the operator's ONE user token, and parallel walks +# Platforms walked one-at-a-time: every native-ingester platform, one paced +# scrape/API walk at a time. gallery-dl platforms are not: each runs as a +# self-pacing subprocess and they are lower-volume. Discord most of all needs +# it: every source walks on the operator's ONE user token, and parallel walks # on a user account are both how its rate limit trips and what gets it flagged. -SERIALIZED_PLATFORMS = frozenset({"patreon", "subscribestar", "discord"}) +SERIALIZED_PLATFORMS = NATIVE_INGESTER_PLATFORMS _LOCK_PREFIX = "fc:download_lock:" diff --git a/backend/app/services/source_service.py b/backend/app/services/source_service.py index 4b5130a..510bc60 100644 --- a/backend/app/services/source_service.py +++ b/backend/app/services/source_service.py @@ -19,7 +19,7 @@ from ..models import ( ) from .db_helpers import failing_sources_clause from .download_backends import uses_native_ingester -from .gallery_dl import ErrorType +from .download_result import ErrorType from .membership_reconcile import KEPT_KEY, STOPPED_KEY from .membership_roster import gated_reasons_for_sources from .platforms import known_platform_keys diff --git a/backend/app/services/subscribestar_client.py b/backend/app/services/subscribestar_client.py index 2589e79..d508d7c 100644 --- a/backend/app/services/subscribestar_client.py +++ b/backend/app/services/subscribestar_client.py @@ -32,7 +32,6 @@ import logging import re import time from collections.abc import Iterator -from dataclasses import dataclass from datetime import datetime from html import unescape from pathlib import Path @@ -43,6 +42,7 @@ import requests from ..utils.paths import filehash_from_url from .native_ingest_common import ( _MAX_429_RETRIES, + MediaItem, Membership, NativeAuthError, NativeDriftError, @@ -163,24 +163,6 @@ class SubscribeStarDriftError(SubscribeStarAPIError, NativeDriftError): silently importing nothing.""" -@dataclass -class MediaItem: - """One resolved downloadable item belonging to a SubscribeStar post. - - Fields mirror patreon_client.MediaItem (so the downloader is structurally the - same) plus `media_id` — the stable per-upload gallery id. SubscribeStar's - full-res URL is an opaque `/post_uploads?payload=...` (not content-addressed), - so `filehash` is usually None and the ledger keys on `:`. - """ - - url: str - filename: str - kind: str - filehash: str | None - post_id: str - media_id: str - - def _extr(text: str, start: str, end: str) -> str: """Substring between the first `start` and the next `end` after it (gallery- dl's `text.extr`); '' when either marker is absent.""" diff --git a/backend/app/services/subscribestar_downloader.py b/backend/app/services/subscribestar_downloader.py index fadb89e..a0d2354 100644 --- a/backend/app/services/subscribestar_downloader.py +++ b/backend/app/services/subscribestar_downloader.py @@ -23,30 +23,23 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. from __future__ import annotations import json -import logging -import time -from collections.abc import Callable from pathlib import Path import requests from .native_ingest_common import ( BaseNativeDownloader, - MediaOutcome, PostRecordOutcome, post_dir_name, - sanitize_segment, ) -log = logging.getLogger(__name__) - class SubscribeStarDownloader(BaseNativeDownloader): """Download resolved SubscribeStar media to gallery-dl's on-disk layout. - Subclasses BaseNativeDownloader for the shared streaming GET (transient-retry + - Range-resume) and validation/quarantine. No video branch (SubscribeStar serves - files directly via /post_uploads) and no detail-fetch (the body is already in - the feed HTML). PURE: no DB.""" + The download loop, per-file layout and sidecars are BaseNativeDownloader's + defaults, unchanged: there is no video branch (SubscribeStar serves files + directly via /post_uploads) and no detail-fetch (the body is already in the + feed HTML). This class adds only the post record. PURE: no DB.""" def __init__( self, @@ -62,110 +55,8 @@ class SubscribeStarDownloader(BaseNativeDownloader): validate=validate, rate_limit=rate_limit, session=session, ) - # -- public ------------------------------------------------------------ - - def download_post( - self, - post: dict, - media_items: list, - artist_slug: str, - *, - 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. - Mirrors PatreonDownloader.download_post (two-tier skip, mid-post time-box, - recapture surfacing) minus the video branch.""" - post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post) - outcomes: list[MediaOutcome] = [] - for i, media in enumerate(media_items, start=1): - if should_stop(): - break - try: - outcomes.append( - 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( - "SubscribeStar media failed (post %s, item %d): %s", - post.get("id"), i, exc, - ) - outcomes.append( - MediaOutcome(media=media, status="error", path=None, error=str(exc)) - ) - return outcomes - - # -- per-item ---------------------------------------------------------- - - def _download_one( - self, - post: dict, - media, - post_dir: Path, - artist_slug: str, - index: int, - is_seen: Callable[[object], bool], - *, - recapture: bool = False, - ) -> MediaOutcome: - seen = is_seen(media) - if seen and not recapture: - return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) - - nn = f"{index:02d}" - media_path = post_dir / sanitize_segment(f"{nn}_{media.filename}") - - if media_path.exists(): # tier-2: already on disk - return MediaOutcome( - media=media, status="skipped_disk", path=media_path, error=None - ) - # recapture: a seen item not on disk is NOT re-downloaded (recovery's job). - if seen: - return MediaOutcome(media=media, status="skipped_seen", path=None, error=None) - - post_dir.mkdir(parents=True, exist_ok=True) - if self._rate_limit > 0: - time.sleep(self._rate_limit) - - out_path = self._fetch_get(media.url, media_path) - reason, quarantine_dest = self._validate_path(out_path, artist_slug, media.url) - if reason is not None: - return MediaOutcome( - media=media, status="quarantined", path=quarantine_dest, error=reason, - ) - self._write_sidecar(post, out_path, source_url=media.url) - return MediaOutcome(media=media, status="downloaded", path=out_path, error=None) - - # The plain-GET streaming path (_fetch_get / _fetch_to_file) and - # _validate_path are inherited from BaseNativeDownloader. - - # -- sidecar ----------------------------------------------------------- - - def _write_sidecar( - self, post: dict, media_path: Path, *, source_url: str | None = None - ) -> Path: - """Per-media sidecar — post-first (#856): image identity ONLY - (category/id/source_url). The post body/links live solely in _post.json.""" - return self._write_sidecar_data( - post, media_path.with_suffix(".json"), source_url=source_url, minimal=True, - ) - - def _write_sidecar_data( - self, post: dict, sidecar_path: Path, *, source_url: str | None = None, - minimal: bool = False, - ) -> Path: - """Serialize the post's metadata. minimal=True → per-media sidecar (image - identity only); else the full post record (body/title/date/url).""" - if minimal: - data = {"category": "subscribestar", "id": str(post.get("id") or "")} - if source_url: - data["source_url"] = source_url - sidecar_path.write_text(json.dumps(data, indent=2)) - return sidecar_path + def _write_post_record_file(self, post: dict, sidecar_path: Path) -> Path: + """Serialize the full post record: body, title and date.""" attrs = post.get("attributes") or {} content = attrs.get("content") # SubscribeStar synthesizes the post permalink from the id (matches the @@ -179,8 +70,6 @@ class SubscribeStarDownloader(BaseNativeDownloader): "content": content if isinstance(content, str) else "", "published_at": attrs.get("published_at"), } - if source_url: - data["source_url"] = source_url sidecar_path.write_text(json.dumps(data, indent=2)) return sidecar_path @@ -214,7 +103,7 @@ class SubscribeStarDownloader(BaseNativeDownloader): ) post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post) post_dir.mkdir(parents=True, exist_ok=True) - path = self._write_sidecar_data(post, post_dir / "_post.json") + path = self._write_post_record_file(post, post_dir / "_post.json") body = attrs.get("content") body_chars = len(body) if isinstance(body, str) else 0 return PostRecordOutcome( diff --git a/backend/app/services/subscribestar_ingester.py b/backend/app/services/subscribestar_ingester.py index 6049362..8864ae3 100644 --- a/backend/app/services/subscribestar_ingester.py +++ b/backend/app/services/subscribestar_ingester.py @@ -22,30 +22,18 @@ from pathlib import Path from ..models import SubscribeStarFailedMedia, SubscribeStarSeenMedia from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester -from .subscribestar_client import MediaItem, SubscribeStarAPIError, SubscribeStarClient +from .native_ingest_common import ledger_key +from .subscribestar_client import SubscribeStarAPIError, SubscribeStarClient from .subscribestar_downloader import SubscribeStarDownloader __all__ = [ "DEAD_LETTER_THRESHOLD", "SubscribeStarIngester", - "_ledger_key", "verify_subscribestar_credential", ] log = logging.getLogger(__name__) -_LEDGER_KEY_MAX = 128 - - -def _ledger_key(media: MediaItem) -> str: - """Stable per-media identity for the cross-run seen-ledger. SubscribeStar's - full-res URL is an opaque `/post_uploads?payload=...` (no content hash), so - `media.filehash` is normally None and the stable proxy is the gallery item id - scoped to its post: `:`. Bounded to the column width.""" - if media.filehash: - return media.filehash - return f"{media.post_id}:{media.media_id}"[:_LEDGER_KEY_MAX] - class SubscribeStarIngester(Ingester): """Walk a SubscribeStar creator's posts, download unseen media, return a @@ -91,7 +79,7 @@ class SubscribeStarIngester(Ingester): failed_model=SubscribeStarFailedMedia, seen_constraint="uq_subscribestar_seen_media_source_id", failed_constraint="uq_subscribestar_failed_media_source_id", - ledger_key=_ledger_key, + ledger_key=ledger_key, platform="subscribestar", error_base=SubscribeStarAPIError, # API_DRIFT message phrasing; the base Ingester._failure_result owns diff --git a/extension/background/background.js b/extension/background/background.js index 74b62d0..3d4a669 100644 --- a/extension/background/background.js +++ b/extension/background/background.js @@ -1,11 +1,6 @@ /** * Background script — message router + Discord token capture (webRequest). * Direct port of GS background.js; api.js client points at FC instead of GS. - * - * pixiv's PKCE OAuth flow lived here until FC retired pixiv (milestone #406). - * It was removed together with pixiv's host permissions rather than left - * behind: a webRequest listener on a host the manifest no longer grants is at - * best dead and at worst a startup failure for the whole background script. */ let discordToken = null; @@ -17,19 +12,9 @@ async function ensureInitialized() { if (initialized) return; await api.init(); await loadDiscordToken(); - await forgetRetiredPixivToken(); initialized = true; } -// A browser that authenticated pixiv before the retirement still holds a live -// OAuth refresh token in extension storage. Nothing reads it any more, and a -// credential for a service FC no longer uses is a liability with no benefit -// (the same reasoning as the server-side cleanup, issue #3980). Removing keys -// that are absent is a no-op, so this is safe on every startup. -async function forgetRetiredPixivToken() { - await browser.storage.local.remove(['pixivRefreshToken', 'pixivTokenCapturedAt']); -} - browser.runtime.onInstalled.addListener(() => ensureInitialized()); browser.runtime.onStartup.addListener(() => ensureInitialized()); ensureInitialized().catch(e => console.error('init failed:', e)); diff --git a/tests/test_discord_client.py b/tests/test_discord_client.py index 3a4db98..56307d6 100644 --- a/tests/test_discord_client.py +++ b/tests/test_discord_client.py @@ -414,7 +414,7 @@ def test_the_request_profile_is_gallery_dls(): # -- the adapter --------------------------------------------------------------- def test_the_adapter_authenticates_with_the_token_and_keys_by_identity(tmp_path): - from backend.app.services.discord_ingester import DiscordIngester, _ledger_key + from backend.app.services.discord_ingester import DiscordIngester ing = DiscordIngester(tmp_path, None, session_factory=None, auth_token="tok") assert ing.client._session.headers["Authorization"] == "tok" @@ -423,7 +423,7 @@ def test_the_adapter_authenticates_with_the_token_and_keys_by_identity(tmp_path) [media] = DiscordClient.extract_media( _msg(9, attachments=[{"id": "300", "url": "https://cdn/a/1.png"}]) ) - assert _ledger_key(media) == "9:300" + assert ing._ledger_key(media) == "9:300" # -- the poster picker (#4491) ------------------------------------------------- diff --git a/tests/test_download_backends.py b/tests/test_download_backends.py index 932a2f6..fa55aef 100644 --- a/tests/test_download_backends.py +++ b/tests/test_download_backends.py @@ -13,7 +13,7 @@ from backend.app.services.download_backends import ( uses_native_ingester, verify_source_credential, ) -from backend.app.services.gallery_dl import ErrorType +from backend.app.services.download_result import ErrorType def test_native_platforms(): diff --git a/tests/test_download_service.py b/tests/test_download_service.py index 8c12f88..f9c725a 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -101,8 +101,6 @@ def _fake_gdl_with_result(result): "skipped_count": 0, "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0, } - fake._extract_errors_warnings = lambda *a, **k: "" - fake._truncate_log = lambda x, **k: x return fake @@ -416,7 +414,8 @@ async def test_run_download_native_unresolvable_fails_loud( """A campaign id we can't resolve is a loud NOT_FOUND failure, never a silent empty success.""" from backend.app.services import download_backends as db_mod - from backend.app.services.gallery_dl import ErrorType, SourceConfig + from backend.app.services.download_result import ErrorType + from backend.app.services.gallery_dl import SourceConfig _artist, source = seed_artist_and_source @@ -733,7 +732,7 @@ async def test_backfill_completes_when_every_post_was_tier_gated( it 'stalled', and the creator we can see least becomes the one we re-walk most. `walk_completed` admits informational classes for exactly this. """ - from backend.app.services.gallery_dl import ErrorType + from backend.app.services.download_result import ErrorType _artist, source = seed_artist_and_source source.config_overrides = { @@ -955,7 +954,7 @@ async def test_backfill_timeout_chunk_reclassified_to_ok( """A backfill chunk that hit its time-box (TIMEOUT) but advanced is reclassified to PARTIAL → event status 'ok' (progress, not error), and consecutive_failures is not bumped.""" - from backend.app.services.gallery_dl import ErrorType + from backend.app.services.download_result import ErrorType _artist, source = seed_artist_and_source source.config_overrides = {"_backfill_state": "running"} @@ -1059,7 +1058,7 @@ async def test_partial_error_type_maps_to_ok_status( (real files were downloaded before the run was cut short) is treated as status=ok by the orchestrator — next tick continues.""" from backend.app.services.download_service import DownloadService - from backend.app.services.gallery_dl import ErrorType + from backend.app.services.download_result import ErrorType from backend.app.services.importer import Importer _artist, source = seed_artist_and_source @@ -1244,3 +1243,54 @@ async def test_async_task_engine_uses_nullpool(): assert isinstance(engine.sync_engine.pool, NullPool) finally: await engine.dispose() + + +@pytest.mark.asyncio +async def test_a_clean_walk_marks_the_credential_verified_and_a_tier_limited_one_does_not( + db, db_sync, tmp_path, seed_artist_and_source, +): + """The credentials page shows when the account last worked. A tier-limited + walk is not evidence: an expired session hides paid posts the same way.""" + from backend.app.models import Credential + from backend.app.services.download_result import ErrorType + from backend.app.services.download_service import DownloadService + from backend.app.services.importer import Importer + + _artist, source = seed_artist_and_source + images_root = tmp_path / "images" + crypto = CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True) + cred_service = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies") + await cred_service.upsert( + platform="patreon", credential_type="cookies", + data=".patreon.com\tTRUE\t/\tTRUE\t0\tsession_id\tabc", + ) + settings = db_sync.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + importer = Importer( + session=db_sync, images_root=images_root, import_root=images_root, + thumbnailer=Thumbnailer(images_root=images_root), settings=settings, + ) + + async def last_verified(): + row = (await db.execute( + select(Credential).where(Credential.platform == "patreon") + )).scalar_one() + await db.refresh(row) + return row.last_verified + + gated = _make_fake_dl_result( + success=True, error_type=ErrorType.TIER_LIMITED, error_message="gated", + ) + svc = DownloadService( + async_session=db, sync_session=db_sync, + gdl=_fake_gdl_with_result(gated), importer=importer, cred_service=cred_service, + ) + _stub_patreon_ingester(svc, gated) + await svc.download_source(source.id) + assert await last_verified() is None + + clean = _make_fake_dl_result(success=True) + _stub_patreon_ingester(svc, clean) + await svc.download_source(source.id) + assert await last_verified() is not None diff --git a/tests/test_gallery_dl_service.py b/tests/test_gallery_dl_service.py index 84454f2..2ee652f 100644 --- a/tests/test_gallery_dl_service.py +++ b/tests/test_gallery_dl_service.py @@ -4,11 +4,12 @@ from types import SimpleNamespace import pytest -from backend.app.services.gallery_dl import ( +from backend.app.services.download_result import ( ErrorType, - GalleryDLService, - SourceConfig, + extract_errors_warnings, + truncate_log, ) +from backend.app.services.gallery_dl import GalleryDLService, SourceConfig @pytest.fixture @@ -228,27 +229,27 @@ def test_compute_run_stats(gdl): assert stats["per_item_failures"] == 1 -def test_extract_errors_warnings(gdl): +def test_extract_errors_warnings(): stderr = ( "[patreon][debug] foo\n[patreon][error] bar\n" "[urllib3][debug] baz\n[patreon][warning] qux\n" ) - result = gdl._extract_errors_warnings(stderr) + result = extract_errors_warnings(stderr) assert "[error]" in result assert "[warning]" in result assert "[debug]" not in result -def test_truncate_log_caps_large_text(gdl): +def test_truncate_log_caps_large_text(): huge = "line\n" * 200_000 - result = gdl._truncate_log(huge, max_bytes=10_000) + result = truncate_log(huge, max_bytes=10_000) assert len(result.encode()) < 11_000 assert "lines elided" in result -def test_truncate_log_passes_short_text(gdl): +def test_truncate_log_passes_short_text(): text = "small\n" - assert gdl._truncate_log(text) == text + assert truncate_log(text) == text # --- Plan #544: tick/backfill skip emission + PARTIAL classifier ------------ diff --git a/tests/test_gated_reason.py b/tests/test_gated_reason.py index 8242a1b..d4bf510 100644 --- a/tests/test_gated_reason.py +++ b/tests/test_gated_reason.py @@ -24,7 +24,7 @@ from pathlib import Path import pytest from backend.app.models import Artist, Source -from backend.app.services.gallery_dl import ErrorType +from backend.app.services.download_result import ErrorType from backend.app.services.membership_roster import ( GATED_FREE, GATED_LAPSED, diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index c857069..80cbe30 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -21,8 +21,9 @@ from backend.app.models import ( PatreonSeenMedia, Source, ) -from backend.app.services.gallery_dl import ErrorType +from backend.app.services.download_result import ErrorType from backend.app.services.ingest_core import _CANARY_MIN_SAMPLE +from backend.app.services.native_ingest_common import ledger_key from backend.app.services.patreon_client import ( MediaItem, PatreonAPIError, @@ -30,7 +31,7 @@ from backend.app.services.patreon_client import ( PatreonDriftError, ) from backend.app.services.patreon_downloader import MediaOutcome, PostRecordOutcome -from backend.app.services.patreon_ingester import PatreonIngester, _ledger_key +from backend.app.services.patreon_ingester import PatreonIngester pytestmark = pytest.mark.integration @@ -141,15 +142,15 @@ class _FakeDownloader: # disk is left alone (skipped_seen, never refetched). if seen and not recapture: outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None)) - elif _ledger_key(m) in self.on_disk: + 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 seen: outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None)) - elif _ledger_key(m) in self.error: + elif ledger_key(m) in self.error: self.download_calls += 1 outcomes.append(MediaOutcome(media=m, status="error", path=None, error="boom")) - elif _ledger_key(m) in self.quarantine: + 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")) @@ -255,7 +256,7 @@ async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_ 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)}) + downloader = _FakeDownloader(tmp_path, quarantine={ledger_key(m2)}) ing = _ingester(sync_engine, tmp_path, client, downloader) result = ing.run( @@ -280,7 +281,7 @@ async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path): # Pre-seed the ledger with m1's key. factory = sessionmaker(sync_engine, expire_on_commit=False) with factory() as s: - s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1")) + s.add(PatreonSeenMedia(source_id=source_id, filehash=ledger_key(m1), post_id="p1")) s.commit() client = _FakeClient([(None, [("p1", [m1])])]) @@ -302,7 +303,7 @@ async def test_tick_early_out_after_threshold(source_id, sync_engine, tmp_path): factory = sessionmaker(sync_engine, expire_on_commit=False) with factory() as s: for m in seen_media: - s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m), post_id=m.post_id)) + s.add(PatreonSeenMedia(source_id=source_id, filehash=ledger_key(m), post_id=m.post_id)) s.commit() pages = [(None, [(m.post_id, [m]) for m in seen_media])] @@ -348,7 +349,7 @@ def _seed_seen_media(sync_engine, source_id, media): factory = sessionmaker(sync_engine, expire_on_commit=False) with factory() as s: for m in media: - s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m), post_id=m.post_id)) + s.add(PatreonSeenMedia(source_id=source_id, filehash=ledger_key(m), post_id=m.post_id)) s.commit() @@ -615,7 +616,7 @@ async def test_recovery_bypasses_seen_ledger(source_id, sync_engine, tmp_path): m1 = _media("p1", 1) factory = sessionmaker(sync_engine, expire_on_commit=False) with factory() as s: - s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1")) + s.add(PatreonSeenMedia(source_id=source_id, filehash=ledger_key(m1), post_id="p1")) s.commit() client = _FakeClient([(None, [("p1", [m1])])]) @@ -638,7 +639,7 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path) client = _FakeClient([(None, [("p1", [m1])])]) # File still on disk (a kept image) → tier-2 spares it even under recovery. _seed_record(sync_engine, tmp_path / "p1_1.jpg") - downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)}) + downloader = _FakeDownloader(tmp_path, on_disk={ledger_key(m1)}) ing = _ingester(sync_engine, tmp_path, client, downloader) result = ing.run( @@ -665,7 +666,7 @@ async def test_backfill_recaptures_body_for_already_downloaded_post( itself, and a normal backfill is the backfill.""" m1 = _media("p1", 1) client = _FakeClient([(None, [("p1", [m1])])]) - downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)}) + downloader = _FakeDownloader(tmp_path, on_disk={ledger_key(m1)}) ing = _ingester(sync_engine, tmp_path, client, downloader) result = ing.run( @@ -704,13 +705,13 @@ async def test_backfill_skips_already_captured_post_but_recapture_forces_it( re-downloading the on-disk media.""" m1 = _media("p1", 1) # Pre-seed BOTH the media key and the synthetic post key as already seen. - _seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1") + _seed_seen(sync_engine, source_id, ledger_key(m1), post_id="p1") _seed_seen(sync_engine, source_id, "post:p1", post_id="p1") _seed_record(sync_engine, tmp_path / "p1_1.jpg") # 1) Plain backfill: post key is seen → gate skips body recapture. client = _FakeClient([(None, [("p1", [m1])])]) - downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)}) + downloader = _FakeDownloader(tmp_path, on_disk={ledger_key(m1)}) ing = _ingester(sync_engine, tmp_path, client, downloader) res_backfill = ing.run( source_id=source_id, campaign_id="c1", artist_slug="ingest", @@ -722,7 +723,7 @@ async def test_backfill_skips_already_captured_post_but_recapture_forces_it( # 2) Recapture: gate bypassed → body recaptured; media NOT re-downloaded; # the on-disk file is surfaced for source_filehash relink. client2 = _FakeClient([(None, [("p1", [m1])])]) - downloader2 = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)}) + downloader2 = _FakeDownloader(tmp_path, on_disk={ledger_key(m1)}) ing2 = _ingester(sync_engine, tmp_path, client2, downloader2) res_recap = ing2.run( source_id=source_id, campaign_id="c1", artist_slug="ingest", @@ -892,7 +893,7 @@ async def test_recapture_does_not_refetch_seen_media_missing_from_disk( """Recapture must NOT re-download a seen item that's absent from disk (that's recovery's job) — it returns skipped_seen and contributes no relink pair.""" m1 = _media("p1", 1) - _seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1") + _seed_seen(sync_engine, source_id, ledger_key(m1), post_id="p1") client = _FakeClient([(None, [("p1", [m1])])]) # NOT in on_disk → seen + missing. downloader = _FakeDownloader(tmp_path) @@ -952,7 +953,7 @@ async def test_a_run_that_dies_before_import_leaves_its_media_unmarked( # The next walk finds the file on disk with no record, and imports it. ing2 = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), - _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})) + _FakeDownloader(tmp_path, on_disk={ledger_key(m1)})) result = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", url="https://patreon.com/ingest", mode="tick") assert result.written_paths == [str(tmp_path / "p1_1.jpg")] @@ -971,9 +972,9 @@ async def test_recapture_imports_a_seen_file_nothing_imported( a walk that looks past it — recapture — reaches them.""" m1, m2 = _media("p1", 1), _media("p1", 2) for m in (m1, m2): - _seed_seen(sync_engine, source_id, _ledger_key(m), post_id="p1") + _seed_seen(sync_engine, source_id, ledger_key(m), post_id="p1") _seed_record(sync_engine, tmp_path / "p1_2.jpg") # m2 made it in; m1 did not - downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1), _ledger_key(m2)}) + downloader = _FakeDownloader(tmp_path, on_disk={ledger_key(m1), ledger_key(m2)}) ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1, m2])])]), downloader) result = ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", @@ -1002,7 +1003,7 @@ async def test_files_phase_3_leaves_unimported_stay_unmarked( assert _count_ledger(sync_engine, source_id) == 3 # m1 and both post records ing2 = _ingester(sync_engine, tmp_path, _FakeClient(pages), - _FakeDownloader(tmp_path, on_disk={_ledger_key(m1), _ledger_key(m2)})) + _FakeDownloader(tmp_path, on_disk={ledger_key(m1), ledger_key(m2)})) again = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", url="https://patreon.com/ingest", mode="backfill") assert again.written_paths == [str(tmp_path / "p2_1.jpg")] @@ -1041,7 +1042,7 @@ async def test_repeated_failures_dead_letter_then_skip(source_id, sync_engine, t ing = _ingester( sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), - _FakeDownloader(tmp_path, error={_ledger_key(m1)}), + _FakeDownloader(tmp_path, error={ledger_key(m1)}), ) result = ing.run( source_id=source_id, campaign_id="c1", artist_slug="ingest", @@ -1070,7 +1071,7 @@ async def test_recovery_reattempts_dead_lettered_and_clears_on_success( from backend.app.services.patreon_ingester import DEAD_LETTER_THRESHOLD m1 = _media("p1", 1) - _seed_failed(sync_engine, source_id, _ledger_key(m1), DEAD_LETTER_THRESHOLD) + _seed_failed(sync_engine, source_id, ledger_key(m1), DEAD_LETTER_THRESHOLD) downloader = _FakeDownloader(tmp_path) # succeeds ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader) @@ -1090,7 +1091,7 @@ async def test_clean_download_clears_below_threshold_failure( """A media with a sub-threshold failure that now downloads cleanly (tick) has its dead-letter row cleared.""" m1 = _media("p1", 1) - _seed_failed(sync_engine, source_id, _ledger_key(m1), 1) # not yet dead + _seed_failed(sync_engine, source_id, ledger_key(m1), 1) # not yet dead downloader = _FakeDownloader(tmp_path) ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader) @@ -1113,8 +1114,8 @@ async def test_mark_seen_is_idempotent(source_id, sync_engine, tmp_path): ing = _ingester(sync_engine, tmp_path, client, downloader) # Same media downloaded twice across two runs — the second marks seen again. - ing._mark_seen(source_id, [(_ledger_key(m1), "p1")]) - ing._mark_seen(source_id, [(_ledger_key(m1), "p1")]) + ing._mark_seen(source_id, [(ledger_key(m1), "p1")]) + ing._mark_seen(source_id, [(ledger_key(m1), "p1")]) assert _count_ledger(sync_engine, source_id) == 1 @@ -1180,7 +1181,7 @@ def test_ledger_key_video_has_no_filehash(): url="https://stream.mux.com/abc.m3u8", filename="clip.mp4", kind="postfile", filehash=None, post_id="p9", ) - assert _ledger_key(video) == "p9:clip.mp4" + assert ledger_key(video) == "p9:clip.mp4" # --- verify_patreon_credential (the verify counterpart, plan #697) --------- @@ -1350,7 +1351,7 @@ def _seed_all_seen(sync_engine, source_id, media): the tests that use it is invisible at the point of use. """ for m in media: - _seed_seen(sync_engine, source_id, _ledger_key(m), post_id=m.post_id) + _seed_seen(sync_engine, source_id, ledger_key(m), post_id=m.post_id) @pytest.mark.asyncio diff --git a/tests/test_subscribestar_native.py b/tests/test_subscribestar_native.py index 5b09809..9bb0cac 100644 --- a/tests/test_subscribestar_native.py +++ b/tests/test_subscribestar_native.py @@ -13,8 +13,8 @@ from pathlib import Path import requests -from backend.app.services.gallery_dl import DownloadResult, ErrorType -from backend.app.services.native_ingest_common import post_dir_name +from backend.app.services.download_result import DownloadResult, ErrorType +from backend.app.services.native_ingest_common import ledger_key, post_dir_name from backend.app.services.subscribestar_client import ( MediaItem, SubscribeStarAPIError, @@ -26,10 +26,7 @@ from backend.app.services.subscribestar_client import ( _split_creator_url, ) from backend.app.services.subscribestar_downloader import SubscribeStarDownloader -from backend.app.services.subscribestar_ingester import ( - SubscribeStarIngester, - _ledger_key, -) +from backend.app.services.subscribestar_ingester import SubscribeStarIngester from backend.app.utils.sidecar import find_sidecar _PNG_HEAD = b"\x89PNG\r\n\x1a\n" @@ -489,12 +486,12 @@ def test_skip_seen_does_not_download(tmp_path): # -- ingester ---------------------------------------------------------------- def test_ledger_key_prefers_filehash_then_post_media(): - assert _ledger_key(_img("a.jpg")) == "111:9001" # no filehash → post:media + assert ledger_key(_img("a.jpg")) == "111:9001" # no filehash → post:media hashed = MediaItem( url="u", filename="a.jpg", kind="image", filehash="deadbeef", post_id="111", media_id="9001", ) - assert _ledger_key(hashed) == "deadbeef" + assert ledger_key(hashed) == "deadbeef" def _ingester(tmp_path):