"""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 GalleryDLService, SourceConfig from .importer import Importer from .patreon_resolver import resolve_campaign_id 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 source_config = SourceConfig.from_dict(ctx["config_overrides"] or {}) 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"], ) 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"], ) 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" 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 if source.platform in ("discord", "pixiv"): 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, } 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 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 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) status = "ok" if (dl_result.success and import_summary["errors"] == 0) else "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 not dl_result.success 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, ) await self.async_session.commit() return event_id async def _update_source_health( self, *, source_id: int, status: str, error_message: str | 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 """ 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 elif status == "error": source.consecutive_failures = (source.consecutive_failures or 0) + 1 source.last_error = error_message 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()