"""FC-3c download orchestrator. 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 3 — Import + persist (fresh session): attach_in_place each written file; clean up duplicates; persist rich metadata. """ from __future__ import annotations import asyncio import logging import re from datetime import UTC, datetime from pathlib import Path from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session as SyncSession from sqlalchemy.orm import joinedload from ..models import Artist, DownloadEvent, Source from .credential_service import CredentialService from .gallery_dl import ( BACKFILL_SKIP_VALUE, BACKFILL_TIMEOUT_SECONDS, TICK_SKIP_VALUE, ErrorType, GalleryDLService, SourceConfig, ) from .importer import Importer from .patreon_resolver import resolve_campaign_id from .platforms import auth_type_for from .scheduler_service import set_platform_cooldown log = logging.getLogger(__name__) _PATREON_VANITY_RE = re.compile( r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)", re.IGNORECASE, ) _CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id" def _extract_patreon_vanity(url: str) -> str | None: m = _PATREON_VANITY_RE.match(url) return m.group(1) if m else None def _looks_like_campaign_id_failure(stdout: str, stderr: str) -> bool: return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower() def _effective_url(platform: str, source_url: str, overrides: dict) -> str: if platform == "patreon" and overrides.get("patreon_campaign_id"): return f"https://www.patreon.com/id:{overrides['patreon_campaign_id']}" return source_url class DownloadService: """Async orchestrator. The Celery task runs `asyncio.run(svc.download_source(N))`. The Importer is sync (existing FC convention); calls to attach_in_place happen via run_in_executor. """ def __init__( self, async_session: AsyncSession, sync_session: SyncSession, gdl: GalleryDLService, importer: Importer, cred_service: CredentialService, ): self.async_session = async_session self.sync_session = sync_session self.gdl = gdl self.importer = importer self.cred_service = cred_service async def download_source(self, source_id: int) -> int: """Returns DownloadEvent.id. Idempotent: in-flight events are returned as-is.""" setup = await self._phase1_setup(source_id) if setup["status"] in ("skipped", "in_flight"): return setup["event_id"] ctx = setup # Release the phase-1 DB connections before the (up to ~19.5-min in # backfill) gallery-dl subprocess. Held checked-out across that idle # window, the asyncpg/psycopg connections get reaped by the server, # and phase 3's first query then hits a dead socket # (asyncpg ConnectionDoesNotExistError) → download_source autoretry → # _phase1_setup's in-flight guard no-ops the retry → the event # strands empty for the recovery sweep (Anduo #40014, 2026-06-04). # pool_pre_ping can't help a *held* connection — it only validates on # pool checkout. Closing returns them to the pool so phase 3 re- # acquires a live one (the async task engine uses NullPool, the sync # engine pre_ping + pool_recycle=300). This is what makes the # "Phase 2 — no DB connection" contract in the class docstring true. await self.async_session.close() self.sync_session.close() source_config = SourceConfig.from_dict(ctx["config_overrides"] or {}) # alembic 0031 / plan #544: derive skip_value + timeout from the # source's backfill_runs_remaining counter. When > 0, walk the full # post history (skip: True + 1170s); when 0, exit gallery-dl after # 20 contiguous archived items (skip: "exit:20" + the default # 870s). Operator sets backfill via POST /api/sources/{id}/backfill. backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0 if backfill_remaining > 0: skip_value: bool | str = BACKFILL_SKIP_VALUE source_config.timeout = BACKFILL_TIMEOUT_SECONDS else: skip_value = TICK_SKIP_VALUE effective_url = _effective_url( ctx["platform"], ctx["url"], ctx["config_overrides"] or {} ) dl_result = await self.gdl.download( url=effective_url, artist_slug=ctx["artist_slug"], platform=ctx["platform"], source_config=source_config, cookies_path=ctx["cookies_path"], auth_token=ctx["auth_token"], skip_value=skip_value, ) resolved_campaign_id: str | None = None if ( ctx["platform"] == "patreon" and not dl_result.success and "patreon_campaign_id" not in (ctx["config_overrides"] or {}) and _looks_like_campaign_id_failure(dl_result.stdout, dl_result.stderr) ): vanity = _extract_patreon_vanity(ctx["url"]) if vanity: log.info( "Attempting campaign-ID resolution for %s (%s)", ctx["artist_slug"], vanity, ) resolved_campaign_id = await resolve_campaign_id( vanity, ctx["cookies_path"] ) if resolved_campaign_id: dl_result = await self.gdl.download( url=f"https://www.patreon.com/id:{resolved_campaign_id}", artist_slug=ctx["artist_slug"], platform=ctx["platform"], source_config=source_config, cookies_path=ctx["cookies_path"], auth_token=ctx["auth_token"], skip_value=skip_value, ) return await self._phase3_persist( ctx["event_id"], ctx, dl_result, resolved_campaign_id, ) async def _phase1_setup(self, source_id: int) -> dict[str, Any]: source = (await self.async_session.execute( select(Source).options(joinedload(Source.artist)).where(Source.id == source_id) )).scalar_one_or_none() if source is None: raise LookupError(f"source {source_id} not found") if not source.enabled: ev = DownloadEvent(source_id=source_id, status="pending") self.async_session.add(ev) await self.async_session.commit() await self.async_session.refresh(ev) await self._finalize_event_and_source( event_id=ev.id, source_id=source_id, status="skipped", error_message="source disabled", ) return {"status": "skipped", "event_id": ev.id} existing = (await self.async_session.execute( select(DownloadEvent).where( DownloadEvent.source_id == source_id, DownloadEvent.status.in_(["pending", "running"]), ).order_by(DownloadEvent.id.desc()).limit(1) )).scalar_one_or_none() if existing and existing.status == "running": return {"status": "in_flight", "event_id": existing.id} if existing and existing.status == "pending": existing.status = "running" # Reset started_at on the pending→running transition so the # recovery sweep (DOWNLOAD_STALL_THRESHOLD_MINUTES, 30 min) # measures from real start, not from enqueue. On heavy-queue # days a freshly-promoted event whose original started_at # predated the cutoff would otherwise get swept mid-flight, # racing phase3's commit. Audit 2026-06-02. existing.started_at = datetime.now(UTC) await self.async_session.commit() event_id = existing.id else: ev = DownloadEvent(source_id=source_id, status="running") self.async_session.add(ev) await self.async_session.commit() await self.async_session.refresh(ev) event_id = ev.id artist = source.artist # Drive cookies-vs-token selection from the platform registry's # auth_type so a new 7th token-platform automatically picks the # right credential path. The hardcoded tuple here used to drift # out of sync with credential_service's auth_type_for(). Audit # 2026-06-02. if auth_type_for(source.platform) == "token": cookies_path = None auth_token = await self.cred_service.get_token(source.platform) else: cookies_path_obj = await self.cred_service.get_cookies_path(source.platform) cookies_path = str(cookies_path_obj) if cookies_path_obj else None auth_token = None return { "status": "ok", "event_id": event_id, "source_id": source_id, "url": source.url, "platform": source.platform, "artist_slug": artist.slug if artist else "_unknown", "artist_id": artist.id if artist else None, "config_overrides": dict(source.config_overrides or {}), "cookies_path": cookies_path, "auth_token": auth_token, "backfill_runs_remaining": source.backfill_runs_remaining or 0, } async def _phase3_persist( self, event_id: int, ctx: dict, dl_result, resolved_campaign_id: str | None, ) -> int: # Cache Patreon campaign id on the source. if resolved_campaign_id: src = self.sync_session.get(Source, ctx["source_id"]) if src is not None: src.config_overrides = { **(src.config_overrides or {}), "patreon_campaign_id": resolved_campaign_id, } self.sync_session.commit() artist = None source_row = None if ctx.get("artist_id"): artist = self.sync_session.get(Artist, ctx["artist_id"]) source_row = self.sync_session.get(Source, ctx["source_id"]) import_summary = {"attached": 0, "skipped": 0, "errors": 0} bytes_downloaded = 0 loop = asyncio.get_running_loop() for path_str in dl_result.written_paths or []: if path_str in (dl_result.quarantined_paths or []): continue path = Path(path_str) # Sync stdlib filesystem ops are intentional here: this orchestrator # runs under `asyncio.run` inside a Celery task — no other awaitables # compete for the loop. ASYNC240 suppressed below. if not path.exists(): # noqa: ASYNC240 continue def _attach(p=path): return self.importer.attach_in_place( p, artist=artist, source=source_row, ) result = await loop.run_in_executor(None, _attach) if result.status in ("imported", "superseded"): import_summary["attached"] += 1 try: bytes_downloaded += path.stat().st_size # noqa: ASYNC240 except OSError: pass # Enqueue thumbnail + ML for newly-attached images, matching # the filesystem-import path (tasks/import_file.py:228-239). # Importer.attach_in_place deliberately skips inline thumb # generation to keep the import queue moving; the calling # task is responsible for the enqueue. Operator-flagged # 2026-06-01: without this, every downloaded image stayed # at thumbnail_path=NULL until a periodic backfill swept # it up, surfacing as broken-thumbnail tiles in the gallery # for hours after a download landed. Lazy import to avoid # circular-import risk between this service and the # tasks/* modules that import it. from ..tasks.ml import tag_and_embed from ..tasks.thumbnail import generate_thumbnail ids = list(result.member_image_ids) if result.image_id is not None and result.image_id not in ids: ids.append(result.image_id) for img_id in ids: generate_thumbnail.delay(img_id) tag_and_embed.delay(img_id) elif result.status == "attached": # Non-media or extracted archive captured as PostAttachment # (FC-2d-iii). The canonical copy lives in the attachments # store; the original download path is now redundant — # mirror duplicate_hash cleanup so we don't keep two copies. # Operator-flagged 2026-06-02 (Lustria OST zip). import_summary["attached"] += 1 try: bytes_downloaded += path.stat().st_size # noqa: ASYNC240 except OSError: pass try: path.unlink(missing_ok=True) # noqa: ASYNC240 except OSError: pass elif result.status == "skipped" and result.skip_reason and result.skip_reason.value in ( "duplicate_hash", "duplicate_phash", ): import_summary["skipped"] += 1 try: path.unlink(missing_ok=True) # noqa: ASYNC240 except OSError: pass elif result.status == "skipped": # Soft skip (too_small, too_transparent, invalid_image) — # the file just didn't qualify, not a download/ingest # failure. Don't flag the run as error; the file stays # on disk for operator inspection. import_summary["skipped"] += 1 elif result.status == "failed": # Hard failure (today only: archive probe crash/timeout). # The original archive sits in /images/ as an orphan; the # filesystem scanner would re-import and re-crash on the # same file, so delete the source file and surface the # error in import_summary. Audit 2026-06-02. import_summary["errors"] += 1 try: path.unlink(missing_ok=True) # noqa: ASYNC240 except OSError: pass elif result.status == "refreshed": # Currently unreachable from attach_in_place (the download # path never runs in deep=True mode), but the importer's # ImportResult contract enumerates it. Treat the same as # 'attached' — work happened, no error. Audit 2026-06-02. import_summary["attached"] += 1 else: import_summary["errors"] += 1 ev = (await self.async_session.execute( 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 ) run_stats["quarantined_count"] = dl_result.files_quarantined stderr_summary = self.gdl._extract_errors_warnings(dl_result.stderr) # Plan #544: PARTIAL means the run downloaded ≥1 file but the # subprocess didn't finish in budget (typically wall-clock timeout # mid-walk). Real work happened; the next tick continues via # gallery-dl's archive. NOT a failure for status purposes. if dl_result.success and import_summary["errors"] == 0: status = "ok" elif dl_result.error_type == ErrorType.PARTIAL and import_summary["errors"] == 0: status = "ok" else: status = "error" ev.status = status ev.finished_at = datetime.now(UTC) ev.files_count = import_summary["attached"] ev.bytes_downloaded = bytes_downloaded ev.error = dl_result.error_message if status == "error" else None ev.metadata_ = { "run_stats": run_stats, "error_type": dl_result.error_type.value if dl_result.error_type else None, "stdout": self.gdl._truncate_log(dl_result.stdout) or None, "stderr": self.gdl._truncate_log(dl_result.stderr) or None, "stderr_errors_warnings": stderr_summary or None, "duration_seconds": dl_result.duration_seconds, "quarantined_paths": dl_result.quarantined_paths or None, "import_summary": import_summary, } await self._update_source_health( source_id=ctx["source_id"], status=status, error_message=ev.error, error_type=dl_result.error_type.value if dl_result.error_type else None, ) # Plan #544: backfill lifecycle — auto-complete when a clean # backfill run drained the queue (gallery-dl exited 0 + zero files # downloaded means there was nothing to fetch); otherwise decrement # the counter. Next tick falls back to tick mode once it hits 0. # # Audit 2026-06-02 gating: VALIDATION_FAILED also exits the # subprocess with return_code=0 and files_downloaded=0 (every # file was quarantined), which used to match the auto-complete # predicate exactly — zeroing the operator's armed budget on # the FIRST quarantine run instead of decrementing. Require # dl_result.success + no error_type so only genuinely-empty # successful runs drain the counter. backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0 if backfill_remaining > 0: src = (await self.async_session.execute( select(Source).where(Source.id == ctx["source_id"]) )).scalar_one() queue_drained = ( dl_result.success and dl_result.error_type is None and dl_result.return_code == 0 and dl_result.files_downloaded == 0 ) if queue_drained: src.backfill_runs_remaining = 0 else: src.backfill_runs_remaining = max(0, backfill_remaining - 1) await self.async_session.commit() return event_id async def _update_source_health( self, *, source_id: int, status: str, error_message: str | None, error_type: str | None = None, ) -> None: """FC-3d: update Source.{consecutive_failures, last_error, last_checked_at}. ok -> failures = 0, error = None, checked_at = now error -> failures += 1, error = error_message, checked_at = now skipped -> failures unchanged, error = None, checked_at = now When error_type == 'rate_limited', also stamps a platform-wide cooldown via scheduler_service.set_platform_cooldown so the next scan tick skips every source on this platform until the cooldown expires. Preventive half of the burst-prevention pair — consecutive_failures still backs the offending source off across ticks. """ source = (await self.async_session.execute( select(Source).where(Source.id == source_id) )).scalar_one() now = datetime.now(UTC) if status == "ok": source.consecutive_failures = 0 source.last_error = None # alembic 0032 — clear the failure-class chip on success. source.error_type = None elif status == "error": source.consecutive_failures = (source.consecutive_failures or 0) + 1 source.last_error = error_message # alembic 0032 — stamp the failure-class so FailingSourcesCard # can render a colored chip and operators can bulk-triage # by error class without opening Logs per row. source.error_type = error_type if error_type == "rate_limited": await set_platform_cooldown(self.async_session, source.platform) elif status == "skipped": source.last_error = None source.last_checked_at = now async def _finalize_event_and_source( self, *, event_id: int, source_id: int, status: str, error_message: str | None, ) -> None: """Finalize the event AND the source-health columns in one transactional step. Used by the skipped early-out path and by unit tests that exercise the source-health writes directly. """ ev = (await self.async_session.execute( select(DownloadEvent).where(DownloadEvent.id == event_id) )).scalar_one() ev.status = status ev.finished_at = datetime.now(UTC) ev.error = error_message await self._update_source_health( source_id=source_id, status=status, error_message=error_message, ) await self.async_session.commit()