feat(fc3c): DownloadService orchestrator — 3-phase pipeline + Patreon retry
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,273 @@
|
||||
"""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, 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="skipped",
|
||||
error="source disabled",
|
||||
finished_at=datetime.now(UTC),
|
||||
)
|
||||
self.async_session.add(ev)
|
||||
await self.async_session.commit()
|
||||
await self.async_session.refresh(ev)
|
||||
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)
|
||||
if not path.exists():
|
||||
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
|
||||
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)
|
||||
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)
|
||||
|
||||
ev.status = "ok" if (dl_result.success and import_summary["errors"] == 0) else "error"
|
||||
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.async_session.commit()
|
||||
return event_id
|
||||
@@ -0,0 +1,254 @@
|
||||
"""DownloadService orchestrator tests.
|
||||
|
||||
GalleryDLService is mocked to return a canned DownloadResult; tests
|
||||
provide on-disk files for the importer to attach_in_place.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.models import Artist, DownloadEvent, ImportSettings, Source
|
||||
from backend.app.services.credential_crypto import CredentialCrypto
|
||||
from backend.app.services.credential_service import CredentialService
|
||||
from backend.app.services.thumbnailer import Thumbnailer
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def seed_artist_and_source(db, db_sync):
|
||||
artist = Artist(name="Alice", slug="alice")
|
||||
db.add(artist)
|
||||
await db.flush()
|
||||
source = Source(
|
||||
artist_id=artist.id, platform="patreon",
|
||||
url="https://patreon.com/alice", enabled=True,
|
||||
config_overrides={},
|
||||
)
|
||||
db.add(source)
|
||||
await db.commit()
|
||||
return artist, source
|
||||
|
||||
|
||||
def _make_jpg(path: Path):
|
||||
from PIL import Image
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
Image.new("RGB", (32, 32), (100, 100, 100)).save(path, "JPEG")
|
||||
|
||||
|
||||
def _make_fake_dl_result(
|
||||
*, success=True, written_paths=None, quarantined_paths=None,
|
||||
files_downloaded=0, error_type=None, error_message=None,
|
||||
stdout="", stderr="",
|
||||
):
|
||||
return SimpleNamespace(
|
||||
success=success,
|
||||
url="https://patreon.com/alice",
|
||||
artist_slug="alice",
|
||||
platform="patreon",
|
||||
files_downloaded=files_downloaded,
|
||||
files_quarantined=len(quarantined_paths or []),
|
||||
quarantined_paths=quarantined_paths or [],
|
||||
written_paths=written_paths or [],
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
return_code=0 if success else 1,
|
||||
error_type=error_type,
|
||||
error_message=error_message,
|
||||
duration_seconds=1.23,
|
||||
started_at="2026-05-20T14:00:00+00:00",
|
||||
completed_at="2026-05-20T14:01:00+00:00",
|
||||
)
|
||||
|
||||
|
||||
def _fake_gdl_with_result(result):
|
||||
fake = MagicMock()
|
||||
fake.download = AsyncMock(return_value=result)
|
||||
fake._compute_run_stats = lambda *a, **k: {
|
||||
"exit_code": 0, "downloaded_count": len(result.written_paths),
|
||||
"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
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_source_attaches_written_files(
|
||||
db, db_sync, tmp_path, seed_artist_and_source,
|
||||
):
|
||||
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"
|
||||
f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
|
||||
f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
|
||||
_make_jpg(f1)
|
||||
_make_jpg(f2)
|
||||
|
||||
fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
|
||||
success=True,
|
||||
written_paths=[str(f1), str(f2)],
|
||||
files_downloaded=2,
|
||||
stdout=f"{f1}\n{f2}\n",
|
||||
))
|
||||
|
||||
sync_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=sync_settings,
|
||||
)
|
||||
crypto = CredentialCrypto(tmp_path / "key.b64")
|
||||
cred_service = CredentialService(db, crypto)
|
||||
|
||||
svc = DownloadService(
|
||||
async_session=db, sync_session=db_sync,
|
||||
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
||||
)
|
||||
event_id = await svc.download_source(source.id)
|
||||
|
||||
ev = (await db.execute(
|
||||
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
||||
)).scalar_one()
|
||||
assert ev.status == "ok"
|
||||
assert ev.files_count == 2
|
||||
assert ev.metadata_["import_summary"]["attached"] == 2
|
||||
assert ev.metadata_["run_stats"]["downloaded_count"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_source_skipped_when_disabled(
|
||||
db, db_sync, tmp_path, seed_artist_and_source,
|
||||
):
|
||||
from backend.app.services.download_service import DownloadService
|
||||
from backend.app.services.importer import Importer
|
||||
|
||||
_artist, source = seed_artist_and_source
|
||||
source.enabled = False
|
||||
await db.commit()
|
||||
|
||||
fake_gdl = MagicMock()
|
||||
fake_gdl.download = AsyncMock()
|
||||
|
||||
sync_settings = db_sync.execute(
|
||||
select(ImportSettings).where(ImportSettings.id == 1)
|
||||
).scalar_one()
|
||||
importer = Importer(
|
||||
session=db_sync, images_root=tmp_path,
|
||||
import_root=tmp_path,
|
||||
thumbnailer=Thumbnailer(images_root=tmp_path),
|
||||
settings=sync_settings,
|
||||
)
|
||||
cred_service = CredentialService(db, MagicMock())
|
||||
|
||||
svc = DownloadService(
|
||||
async_session=db, sync_session=db_sync,
|
||||
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
||||
)
|
||||
event_id = await svc.download_source(source.id)
|
||||
|
||||
ev = (await db.execute(
|
||||
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
||||
)).scalar_one()
|
||||
assert ev.status == "skipped"
|
||||
assert ev.error == "source disabled"
|
||||
fake_gdl.download.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_in_flight_idempotency_returns_existing_event(
|
||||
db, db_sync, tmp_path, seed_artist_and_source,
|
||||
):
|
||||
from backend.app.services.download_service import DownloadService
|
||||
|
||||
_artist, source = seed_artist_and_source
|
||||
running = DownloadEvent(source_id=source.id, status="running")
|
||||
db.add(running)
|
||||
await db.commit()
|
||||
await db.refresh(running)
|
||||
|
||||
fake_gdl = MagicMock()
|
||||
fake_gdl.download = AsyncMock()
|
||||
importer_stub = MagicMock()
|
||||
cred_service = CredentialService(db, MagicMock())
|
||||
|
||||
svc = DownloadService(
|
||||
async_session=db, sync_session=db_sync,
|
||||
gdl=fake_gdl, importer=importer_stub, cred_service=cred_service,
|
||||
)
|
||||
event_id = await svc.download_source(source.id)
|
||||
assert event_id == running.id
|
||||
fake_gdl.download.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_patreon_retry_caches_campaign_id(
|
||||
db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
|
||||
):
|
||||
from backend.app.services.download_service import DownloadService
|
||||
from backend.app.services.importer import Importer
|
||||
|
||||
_artist, source = seed_artist_and_source
|
||||
|
||||
failed = _make_fake_dl_result(
|
||||
success=False, written_paths=[], stdout="",
|
||||
stderr="[patreon][error] Failed to extract campaign ID for alice\n",
|
||||
)
|
||||
succeeded = _make_fake_dl_result(success=True, written_paths=[])
|
||||
|
||||
download_calls = []
|
||||
|
||||
async def fake_download(url, **k):
|
||||
download_calls.append(url)
|
||||
return failed if len(download_calls) == 1 else succeeded
|
||||
|
||||
fake_gdl = MagicMock()
|
||||
fake_gdl.download = fake_download
|
||||
fake_gdl._compute_run_stats = lambda *a, **k: {
|
||||
"exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
|
||||
"per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
|
||||
}
|
||||
fake_gdl._extract_errors_warnings = lambda *a, **k: ""
|
||||
fake_gdl._truncate_log = lambda x, **k: x
|
||||
|
||||
monkeypatch.setattr(
|
||||
"backend.app.services.download_service.resolve_campaign_id",
|
||||
AsyncMock(return_value="99"),
|
||||
)
|
||||
|
||||
sync_settings = db_sync.execute(
|
||||
select(ImportSettings).where(ImportSettings.id == 1)
|
||||
).scalar_one()
|
||||
importer = Importer(
|
||||
session=db_sync, images_root=tmp_path,
|
||||
import_root=tmp_path,
|
||||
thumbnailer=Thumbnailer(images_root=tmp_path),
|
||||
settings=sync_settings,
|
||||
)
|
||||
cred_service = CredentialService(db, MagicMock())
|
||||
|
||||
svc = DownloadService(
|
||||
async_session=db, sync_session=db_sync,
|
||||
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
||||
)
|
||||
await svc.download_source(source.id)
|
||||
|
||||
assert len(download_calls) == 2
|
||||
assert "id:99" in download_calls[1]
|
||||
|
||||
overrides = db_sync.execute(
|
||||
select(Source.config_overrides).where(Source.id == source.id)
|
||||
).scalar_one()
|
||||
assert overrides.get("patreon_campaign_id") == "99"
|
||||
Reference in New Issue
Block a user