diff --git a/backend/app/services/download_backends.py b/backend/app/services/download_backends.py index 737da4d..89b295a 100644 --- a/backend/app/services/download_backends.py +++ b/backend/app/services/download_backends.py @@ -20,8 +20,13 @@ function regardless of platform: from __future__ import annotations +import asyncio from pathlib import Path +from .gallery_dl import DownloadResult, ErrorType +from .patreon_ingester import PatreonIngester +from .patreon_resolver import resolve_campaign_id_for_source + # Platforms whose download + verify go through the native ingester rather than # gallery-dl. gallery-dl still serves every other platform (subscribestar, # hentaifoundry, discord, pixiv, deviantart) unchanged. @@ -34,6 +39,111 @@ def uses_native_ingester(platform: str) -> bool: return platform in NATIVE_INGESTER_PLATFORMS +async def run_download( + *, + ctx: dict, + source_config, + skip_value: bool | str, + mode: str | None, + gdl, + sync_session_factory, +) -> tuple[DownloadResult, str | None]: + """Uniform download across backends — the download counterpart to + `verify_source_credential`, so this module is the ONE place that knows how + each platform both downloads AND verifies (the seam that makes adding a + platform a bounded job). + + Returns `(DownloadResult, resolved_campaign_id)`; `resolved_campaign_id` is + non-None only when a native vanity lookup ran this call (so phase 3 caches + it). Native platforms route through their ingester in `mode` + (tick/backfill/recovery); gallery-dl platforms run the subprocess. The caller + (download_service) prepares `source_config`/`skip_value`/`mode` from the + backfill state machine and owns phase 3. + """ + platform = ctx["platform"] + if uses_native_ingester(platform): + return await _run_native_ingester( + ctx, source_config, mode, gdl, sync_session_factory + ) + result = await gdl.download( + url=ctx["url"], + artist_slug=ctx["artist_slug"], + platform=platform, + source_config=source_config, + cookies_path=ctx["cookies_path"], + auth_token=ctx["auth_token"], + skip_value=skip_value, + ) + return result, None + + +async def _run_native_ingester( + ctx: dict, source_config, mode: str | None, gdl, sync_session_factory, +) -> tuple[DownloadResult, str | None]: + """Patreon (today the only native platform): resolve the campaign id, then run + the native ingester in a worker thread (it is sync requests/subprocess). + + `resolved_campaign_id` is non-None only when we had to look it up from the + vanity URL this run, so phase 3 caches it the way the old gallery-dl retry + did. A campaign id we cannot resolve is a loud NOT_FOUND — never a silent + empty success. + """ + overrides = ctx["config_overrides"] or {} + campaign_id, resolved_campaign_id = await resolve_campaign_id_for_source( + ctx["url"], ctx["cookies_path"], overrides + ) + if not campaign_id: + return ( + DownloadResult( + success=False, + url=ctx["url"], + artist_slug=ctx["artist_slug"], + platform="patreon", + error_type=ErrorType.NOT_FOUND, + error_message=( + "Could not resolve Patreon campaign id from the source URL " + "(vanity lookup failed — cookies expired or creator moved?)" + ), + ), + None, + ) + + # Honor the operator's existing rate-limit knobs on the native path (plan + # #703): the global download_rate_limit_seconds (gallery-dl's `rate_limit`, + # here on the gdl service) paces media downloads; page fetches use the + # per-source sleep_request override, else `max(0.5, rate_limit/4)` — the same + # API-pacing default gallery-dl applied as its `sleep-request`. + rate_limit = gdl._rate_limit + request_sleep = ( + source_config.sleep_request + if source_config.sleep_request is not None + else max(0.5, rate_limit / 4) + ) + ingester = PatreonIngester( + images_root=gdl.images_root, + cookies_path=ctx["cookies_path"], + session_factory=sync_session_factory, + validate=gdl._validate_files, + rate_limit=rate_limit, + request_sleep=request_sleep, + ) + loop = asyncio.get_running_loop() + dl_result = await loop.run_in_executor( + None, + lambda: ingester.run( + source_id=ctx["source_id"], + campaign_id=campaign_id, + artist_slug=ctx["artist_slug"], + url=ctx["url"], + mode=mode, + resume_cursor=source_config.resume_cursor, + time_budget_seconds=source_config.timeout, + posts_base=int(overrides.get("_backfill_posts", 0)), + ), + ) + return dl_result, resolved_campaign_id + + async def verify_source_credential( *, platform: str, diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 86d8324..5295dcd 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -24,7 +24,7 @@ from sqlalchemy.orm import joinedload from ..models import Artist, DownloadEvent, Source from .credential_service import CredentialService -from .download_backends import uses_native_ingester +from .download_backends import run_download, uses_native_ingester from .gallery_dl import ( BACKFILL_CHUNK_SECONDS, BACKFILL_SKIP_VALUE, @@ -37,8 +37,6 @@ from .gallery_dl import ( truncate_log, ) from .importer import Importer -from .patreon_ingester import PatreonIngester -from .patreon_resolver import resolve_campaign_id_for_source from .platforms import auth_type_for from .scheduler_service import set_platform_cooldown @@ -122,32 +120,23 @@ class DownloadService: else: skip_value = TICK_SKIP_VALUE - resolved_campaign_id: str | None = None + # Phase 2 dispatch is uniform across backends (download_backends. + # run_download — the download counterpart to verify_source_credential): + # native platforms run their ingester in `mode` (zero per-file HEADs, + # native cursor/resume, loud drift detection); gallery-dl platforms run + # the subprocess. Either returns a DownloadResult-shaped object so phase 3 + # is untouched. `mode` is None for gallery-dl (run_download ignores it). + mode: str | None = None if uses_native_ingester(ctx["platform"]): - # Native ingester (plan #697) fully replaces gallery-dl for Patreon - # in phase 2 — zero per-file HEADs, native cursor/resume, loud drift - # detection. Returns a DownloadResult-shaped object so phase 3 is - # untouched. The gallery-dl Patreon path + its campaign-id retry are - # removed at the step-5 cutover. if in_backfill and bypass_seen: mode = "recovery" elif in_backfill: mode = "backfill" else: mode = "tick" - dl_result, resolved_campaign_id = await self._run_patreon_ingester( - ctx, source_config, mode, - ) - else: - dl_result = await self.gdl.download( - url=ctx["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, - ) + dl_result, resolved_campaign_id = await self._run_download( + ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode, + ) # A backfill chunk that hit its time-box but made forward progress is # NORMAL, not a failure — reclassify TIMEOUT → PARTIAL so it reads as @@ -171,82 +160,18 @@ class DownloadService: ctx["event_id"], ctx, dl_result, resolved_campaign_id, ) - async def _run_patreon_ingester( - self, ctx: dict, source_config, mode: str, + async def _run_download( + self, *, ctx: dict, source_config, skip_value, mode: str | None, ) -> tuple[DownloadResult, str | None]: - """Phase-2 Patreon branch: resolve the campaign id, then run the native - ingester in a worker thread (it is sync requests/subprocess). - - Returns (DownloadResult, resolved_campaign_id). `resolved_campaign_id` is - non-None only when we had to look it up from the vanity URL this run, so - phase 3 caches it on the source the same way the old gallery-dl retry did. - A campaign id we cannot resolve is a loud failure (NOT_FOUND) — never a - silent empty success. - """ - overrides = ctx["config_overrides"] or {} - # Shared resolution path (override / id: URL / vanity lookup) — the same - # helper the credential-verify probe uses. resolved_campaign_id is - # non-None only when a vanity lookup ran, so phase 3 caches it. - campaign_id, resolved_campaign_id = await resolve_campaign_id_for_source( - ctx["url"], ctx["cookies_path"], overrides + """Phase-2 dispatch → `download_backends.run_download`, passing this + service's gdl + sync sessionmaker. Kept as a thin instance method so tests + can stub the whole phase-2 dispatch on the service, and so the per-backend + construction lives in download_backends (the backend registry).""" + return await run_download( + ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode, + gdl=self.gdl, sync_session_factory=self.sync_session_factory, ) - if not campaign_id: - return ( - DownloadResult( - success=False, - url=ctx["url"], - artist_slug=ctx["artist_slug"], - platform="patreon", - error_type=ErrorType.NOT_FOUND, - error_message=( - "Could not resolve Patreon campaign id from the source " - "URL (vanity lookup failed — cookies expired or creator " - "moved?)" - ), - ), - None, - ) - - # Honor the operator's existing rate-limit knobs on the native path - # (plan #703): the global download_rate_limit_seconds (gallery-dl's - # `rate_limit`, here on self.gdl) paces media downloads. Page fetches use - # the per-source sleep_request override, else `max(0.5, rate_limit/4)` — - # the SAME API-pacing default gallery-dl applied as its `sleep-request` - # (gallery_dl._get_default_config), so the rate-limited /api/posts - # endpoint stays politely paced without over-throttling. - rate_limit = self.gdl._rate_limit - request_sleep = ( - source_config.sleep_request - if source_config.sleep_request is not None - else max(0.5, rate_limit / 4) - ) - ingester = PatreonIngester( - images_root=self.gdl.images_root, - cookies_path=ctx["cookies_path"], - session_factory=self.sync_session_factory, - validate=self.gdl._validate_files, - rate_limit=rate_limit, - request_sleep=request_sleep, - ) - loop = asyncio.get_running_loop() - dl_result = await loop.run_in_executor( - None, - lambda: ingester.run( - source_id=ctx["source_id"], - campaign_id=campaign_id, - artist_slug=ctx["artist_slug"], - url=ctx["url"], - mode=mode, - resume_cursor=source_config.resume_cursor, - time_budget_seconds=source_config.timeout, - # plan #704 #5: seed the live posts badge from prior chunks so the - # ingester writes a monotonic absolute mid-walk (it owns the key). - posts_base=int(overrides.get("_backfill_posts", 0)), - ), - ) - return 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) diff --git a/tests/test_download_service.py b/tests/test_download_service.py index d08eb12..354e04e 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -89,7 +89,7 @@ def _make_fake_dl_result( def _fake_gdl_with_result(result): fake = MagicMock() fake.download = AsyncMock(return_value=result) - # Real values for the attrs _run_patreon_ingester reads off the gdl service + # Real values for the attrs run_download's native branch reads off the gdl # (the native pacing config, plan #703) — a MagicMock would break the # arithmetic (max(0.5, rate_limit/4)). fake._rate_limit = 3.0 @@ -105,19 +105,24 @@ def _fake_gdl_with_result(result): def _stub_patreon_ingester(svc, result, resolved_campaign_id=None): - """Route the Patreon phase-2 branch (plan #697 native ingester) to a canned - DownloadResult so these tests exercise download_service's - phase-2-result→phase-3 handling (status mapping, backfill lifecycle, health) - without the real ingester's network/DB. The ingester itself is covered by - test_patreon_ingester.py. Returns a list capturing (ctx, source_config, mode) - per call so a test can assert mode / resume_cursor threading.""" + """Route the phase-2 dispatch (download_backends.run_download via + DownloadService._run_download, plan #707 A5) to a canned DownloadResult so + these tests exercise download_service's phase-2-result→phase-3 handling + (status mapping, backfill lifecycle, health) without the real ingester's + network/DB. The ingester itself is covered by test_patreon_ingester.py; the + native construction/resolution by the run_download tests below. Returns a list + capturing (ctx, source_config, skip_value, mode) per call so a test can assert + mode / resume_cursor threading.""" calls = [] - async def fake(ctx, source_config, mode): - calls.append({"ctx": ctx, "source_config": source_config, "mode": mode}) + async def fake(*, ctx, source_config, skip_value, mode): + calls.append({ + "ctx": ctx, "source_config": source_config, + "skip_value": skip_value, "mode": mode, + }) return result, resolved_campaign_id - svc._run_patreon_ingester = fake + svc._run_download = fake return calls @@ -285,20 +290,19 @@ async def test_patreon_resolved_campaign_id_is_cached( @pytest.mark.asyncio -async def test_run_patreon_ingester_resolves_vanity_and_runs( +async def test_run_download_native_resolves_vanity_and_runs( db, db_sync, tmp_path, seed_artist_and_source, monkeypatch, ): - """_run_patreon_ingester: with no cached campaign id, resolve the vanity and - pass the resolved id straight to the ingester; report it back so phase 3 can - cache it.""" - from backend.app.services import download_service as dl_mod - from backend.app.services.download_service import DownloadService + """run_download (native branch, plan #707 A5): with no cached campaign id, + resolve the vanity and pass the resolved id straight to the ingester; report + it back so phase 3 can cache it.""" + from backend.app.services import download_backends as db_mod from backend.app.services.gallery_dl import SourceConfig _artist, source = seed_artist_and_source monkeypatch.setattr( - dl_mod, "resolve_campaign_id_for_source", + db_mod, "resolve_campaign_id_for_source", AsyncMock(return_value=("4242", "4242")), ) run_kwargs = {} @@ -311,21 +315,19 @@ async def test_run_patreon_ingester_resolves_vanity_and_runs( run_kwargs.update(kw) return _make_fake_dl_result(success=True, written_paths=[]) - monkeypatch.setattr(dl_mod, "PatreonIngester", _FakeIngester) + monkeypatch.setattr(db_mod, "PatreonIngester", _FakeIngester) - svc = DownloadService( - async_session=db, sync_session=db_sync, - gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)), - importer=MagicMock(), cred_service=MagicMock(), - sync_session_factory=MagicMock(), - ) ctx = { + "platform": "patreon", "source_id": source.id, "url": "https://patreon.com/alice", "artist_slug": "alice", "cookies_path": None, "config_overrides": {}, } - result, resolved = await svc._run_patreon_ingester( - ctx, SourceConfig.from_dict({}), "tick", + result, resolved = await db_mod.run_download( + ctx=ctx, source_config=SourceConfig.from_dict({}), skip_value=True, + mode="tick", + gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)), + sync_session_factory=MagicMock(), ) assert result.success is True assert resolved == "4242" @@ -334,33 +336,29 @@ async def test_run_patreon_ingester_resolves_vanity_and_runs( @pytest.mark.asyncio -async def test_run_patreon_ingester_unresolvable_fails_loud( +async def test_run_download_native_unresolvable_fails_loud( db, db_sync, tmp_path, seed_artist_and_source, monkeypatch, ): """A campaign id we can't resolve is a loud NOT_FOUND failure, never a silent empty success.""" - from backend.app.services import download_service as dl_mod - from backend.app.services.download_service import DownloadService + from backend.app.services import download_backends as db_mod from backend.app.services.gallery_dl import ErrorType, SourceConfig _artist, source = seed_artist_and_source monkeypatch.setattr( - dl_mod, "resolve_campaign_id_for_source", + db_mod, "resolve_campaign_id_for_source", AsyncMock(return_value=(None, None)), ) - svc = DownloadService( - async_session=db, sync_session=db_sync, - gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(), - sync_session_factory=MagicMock(), - ) ctx = { + "platform": "patreon", "source_id": source.id, "url": "https://patreon.com/alice", "artist_slug": "alice", "cookies_path": None, "config_overrides": {}, } - result, resolved = await svc._run_patreon_ingester( - ctx, SourceConfig.from_dict({}), "tick", + result, resolved = await db_mod.run_download( + ctx=ctx, source_config=SourceConfig.from_dict({}), skip_value=True, + mode="tick", gdl=MagicMock(), sync_session_factory=MagicMock(), ) assert result.success is False assert result.error_type == ErrorType.NOT_FOUND @@ -926,11 +924,11 @@ async def test_releases_db_connections_before_subprocess( gdl=fake_gdl, importer=importer, cred_service=cred_service, ) - async def fake_ingest(ctx, source_config, mode): + async def fake_ingest(*, ctx, source_config, skip_value, mode): seen["closed_before_phase2"] = closed["async"] return _make_fake_dl_result(success=True, written_paths=[], stdout=""), None - svc._run_patreon_ingester = fake_ingest + svc._run_download = fake_ingest event_id = await svc.download_source(source.id) assert seen["closed_before_phase2"] is True