diff --git a/README.md b/README.md index 91082c5..2381b0f 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,40 @@ Ingesting a series also queues its posters/backdrops/profile images for local download automatically (only the provider's *primary* image per entity, not every alternate) — see the next section. +### Bulk-importing TMDb's entire catalog + +`app/scripts/bulk_import.py` enqueues an ingestion job for every id in +TMDb's [daily ID export files](https://developer.themoviedb.org/docs/daily-id-exports) +(published once a day specifically for full-catalog sync, unlike +`/discover`, which caps at 10,000 results per query) — the same +one-id-at-a-time pipeline the admin endpoints above use, just enqueued for +every id TMDb has rather than one at a time by hand: + +```bash +docker compose -f docker-compose.prod.yml exec worker \ + python -m app.scripts.bulk_import movie series person collection +``` + +As of 2026-09, TMDb's real catalog is roughly 1.24M movies, 231K TV +series, 4.9M people, and 9.8K collections — **~6.4M entities total**. At +this project's default worker concurrency (`WORKER_CONCURRENCY=4` × 2 +worker replicas), expect this to run for **on the order of weeks**, not +hours — it does not rate-limit itself against TMDb; TMDb's request rate is +governed entirely by how many workers are consuming the queue +concurrently (raise `WORKER_CONCURRENCY` / scale up workers to go faster, +mindful of TMDb's own rate limits — `app/providers/tmdb.py` backs off on a +429 using TMDb's own `Retry-After` hint, but sustained high concurrency +will still see more of them). Run it detached +(`docker compose ... exec -d worker ...`, or under `nohup`/`tmux`/`screen`) +so it survives a disconnect, and use `--dry-run` first to see counts +without enqueueing anything. Re-running (e.g. after an interruption) is +safe — every id is idempotency-keyed, so already-created jobs are reused, +never duplicated. + +Expect substantial disk usage from artwork self-hosting cascading on +every successful ingestion (see the next section) at this scale — budget +accordingly before running it against the full catalog. + ## Artwork Images are referenced by their provider CDN URL (TMDb) from the moment diff --git a/app/providers/base.py b/app/providers/base.py index b1c9cf5..e226355 100644 --- a/app/providers/base.py +++ b/app/providers/base.py @@ -27,6 +27,17 @@ class ProviderNotFoundError(ProviderError): """Raised when the provider has no record for the given external ID.""" +class ProviderRateLimitedError(ProviderError): + """Raised when the provider responds 429. Carries how long the provider + itself asked callers to wait (`Retry-After`, seconds) so a caller can + back off by the right amount rather than guessing — see + `app/providers/tmdb.py:TMDbProvider._get` for where this is retried.""" + + def __init__(self, message: str, *, retry_after: float) -> None: + super().__init__(message) + self.retry_after = retry_after + + class MetadataProvider(ABC): slug: str name: str diff --git a/app/providers/tmdb.py b/app/providers/tmdb.py index f8caf42..7d86ef6 100644 --- a/app/providers/tmdb.py +++ b/app/providers/tmdb.py @@ -10,7 +10,13 @@ from __future__ import annotations from typing import Any import httpx -from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential +from tenacity import ( + RetryCallState, + retry, + retry_if_exception_type, + stop_after_attempt, + wait_exponential, +) from app.core.config import get_settings from app.providers.base import ( @@ -18,12 +24,26 @@ from app.providers.base import ( ProviderError, ProviderNotConfiguredError, ProviderNotFoundError, + ProviderRateLimitedError, ) _MOVIE_APPEND = "credits,images,videos,keywords,alternative_titles,external_ids,release_dates,recommendations" _SERIES_APPEND = "credits,images,videos,keywords,alternative_titles,external_ids,content_ratings" _PERSON_APPEND = "external_ids,images,combined_credits" +_DEFAULT_EXPONENTIAL_WAIT = wait_exponential(multiplier=0.5, max=5) + + +def _tmdb_wait(retry_state: RetryCallState) -> float: + """A 429's own `Retry-After` (carried on ProviderRateLimitedError) takes + priority over guessing via exponential backoff -- TMDb told us exactly + how long to wait, so use that instead of a generic schedule.""" + + exc = retry_state.outcome.exception() if retry_state.outcome else None + if isinstance(exc, ProviderRateLimitedError): + return exc.retry_after + return _DEFAULT_EXPONENTIAL_WAIT(retry_state) + class TMDbProvider(MetadataProvider): slug = "tmdb" @@ -45,9 +65,9 @@ class TMDbProvider(MetadataProvider): @retry( reraise=True, - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=0.5, max=5), - retry=retry_if_exception_type(httpx.TransportError), + stop=stop_after_attempt(5), + wait=_tmdb_wait, + retry=retry_if_exception_type((httpx.TransportError, ProviderRateLimitedError)), ) async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: self._require_api_key() @@ -61,6 +81,17 @@ class TMDbProvider(MetadataProvider): raise ProviderNotFoundError(f"TMDb has no record at {path}") if response.status_code == 401: raise ProviderNotConfiguredError("TMDb rejected the configured TMDB_API_KEY.") + if response.status_code == 429: + try: + # Retry-After is defined (RFC 9110) as either a delay in + # seconds or an HTTP-date; TMDb sends seconds, but fall back + # rather than crash if that ever changes. + retry_after = float(response.headers.get("Retry-After", "1")) + except ValueError: + retry_after = 1.0 + raise ProviderRateLimitedError( + f"TMDb rate-limited {path}", retry_after=retry_after + ) if response.status_code >= 400: raise ProviderError(f"TMDb error {response.status_code} at {path}: {response.text[:300]}") return response.json() diff --git a/app/scripts/__init__.py b/app/scripts/__init__.py new file mode 100644 index 0000000..3768334 --- /dev/null +++ b/app/scripts/__init__.py @@ -0,0 +1,2 @@ +"""One-off operational scripts, run directly (`python -m app.scripts.`) +rather than through the API or arq worker task list.""" diff --git a/app/scripts/bulk_import.py b/app/scripts/bulk_import.py new file mode 100644 index 0000000..3c457bc --- /dev/null +++ b/app/scripts/bulk_import.py @@ -0,0 +1,151 @@ +"""One-off ops script: enqueue an ingestion job for every id in TMDb's +daily export, for one or more entity types -- a full-catalog bulk sync. + + python -m app.scripts.bulk_import movie series person collection + python -m app.scripts.bulk_import movie --dry-run + python -m app.scripts.bulk_import series --no-seasons + +Every id goes through the exact same IngestionJob + arq task pipeline the +admin API uses for a single id (get_or_create_ingestion_job is +idempotency-keyed on provider+entity_type+external_id), so interrupting +and re-running this script is safe -- already-created jobs are reused, +never duplicated, and re-enqueuing an already-succeeded job just re-syncs +that entity. + +This does not itself rate-limit against TMDb -- the actual API calls +happen later, inside the worker tasks this enqueues, so TMDb's request +rate is governed by WORKER_CONCURRENCY (how many workers process the +queue concurrently), not by how fast this script can enqueue. Enqueueing +does hit Postgres once per id (via get_or_create_ingestion_job), so this +runs a bounded number of ids concurrently to avoid one id's DB round-trip +serializing the next -- see _CONCURRENCY below. + +Expect this to run for hours across a full catalog (TMDb has roughly 1M+ +movies, 200K+ TV series, 3M+ people, 150K+ collections) -- run it inside +the worker container detached (e.g. `docker compose -f +docker-compose.prod.yml exec -d worker python -m app.scripts.bulk_import +...`) or under nohup/tmux/screen so it survives a disconnect, and expect +substantial TMDb API traffic, database growth, and (since artwork +self-hosting cascades on every successful ingestion, per +docs/database.md's artwork storage section) disk usage for downloaded +images while it runs. +""" + +from __future__ import annotations + +import argparse +import asyncio + +import httpx +from arq import ArqRedis + +from app.core.database import dispose_engine +from app.core.logging import configure_logging, get_logger +from app.core.queue import close_arq_pool, get_arq_pool +from app.services.bulk_import import fetch_daily_export_ids +from app.services.ingestion.runner import get_or_create_ingestion_job + +logger = get_logger(__name__) + +_TASK_NAME: dict[str, str] = { + "movie": "ingest_movie_task", + "series": "ingest_series_task", + "person": "ingest_person_task", + "collection": "ingest_collection_task", +} +_JOB_TYPE: dict[str, str] = { + "movie": "movie_sync", + "series": "series_sync", + "person": "person_sync", + "collection": "collection_sync", +} + +# How many ids to create IngestionJob rows / enqueue for concurrently. +# Bounded so this doesn't try to open millions of DB round-trips at once, +# but high enough that per-id latency isn't fully serialized. +_CONCURRENCY = 50 + + +async def _enqueue_one( + queue: ArqRedis, entity_type: str, external_id: str, *, include_seasons: bool +) -> None: + job_id = await get_or_create_ingestion_job(_JOB_TYPE[entity_type], entity_type, external_id) + task_name = _TASK_NAME[entity_type] + if entity_type == "series": + await queue.enqueue_job(task_name, str(job_id), external_id, include_seasons) + else: + await queue.enqueue_job(task_name, str(job_id), external_id) + + +async def _bulk_enqueue(entity_type: str, *, dry_run: bool, include_seasons: bool) -> None: + async with httpx.AsyncClient(timeout=120.0) as client: + ids = await fetch_daily_export_ids(entity_type, client=client) + logger.info("bulk_import_fetched_ids", entity_type=entity_type, count=len(ids)) + + if dry_run: + logger.info("bulk_import_dry_run", entity_type=entity_type, would_enqueue=len(ids)) + return + + queue = await get_arq_pool() + semaphore = asyncio.Semaphore(_CONCURRENCY) + enqueued = 0 + lock = asyncio.Lock() + + async def _worker(external_id: str) -> None: + nonlocal enqueued + async with semaphore: + await _enqueue_one(queue, entity_type, external_id, include_seasons=include_seasons) + async with lock: + enqueued += 1 + if enqueued % 5000 == 0: + logger.info( + "bulk_import_progress", + entity_type=entity_type, + enqueued=enqueued, + total=len(ids), + ) + + await asyncio.gather(*(_worker(external_id) for external_id in ids)) + logger.info("bulk_import_complete", entity_type=entity_type, enqueued=enqueued) + + +async def _main(entity_types: list[str], *, dry_run: bool, include_seasons: bool) -> None: + configure_logging() + try: + for entity_type in entity_types: + await _bulk_enqueue(entity_type, dry_run=dry_run, include_seasons=include_seasons) + finally: + await close_arq_pool() + await dispose_engine() + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Bulk-enqueue ingestion jobs from TMDb's daily ID export files." + ) + parser.add_argument( + "entity_types", + nargs="+", + choices=list(_TASK_NAME), + help="One or more entity types to bulk-import.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Fetch and count ids without enqueueing anything.", + ) + parser.add_argument( + "--no-seasons", + action="store_true", + help="For 'series': don't cascade into per-season/episode ingestion " + "(cuts total TMDb calls roughly in half for TV, at the cost of " + "season/episode data). Cascading is the default for a full pull.", + ) + args = parser.parse_args() + asyncio.run( + _main(args.entity_types, dry_run=args.dry_run, include_seasons=not args.no_seasons) + ) + + +if __name__ == "__main__": + main() diff --git a/app/services/bulk_import.py b/app/services/bulk_import.py new file mode 100644 index 0000000..de97a0d --- /dev/null +++ b/app/services/bulk_import.py @@ -0,0 +1,81 @@ +"""TMDb daily ID export fetching, for full-catalog bulk sync. + +TMDb publishes a full list of every movie/TV/person/collection id that +exists in their database once a day, specifically to support this exact +use case -- syncing an entire local mirror is what these files are for, +as opposed to paginating `/discover`, which caps at 10,000 results per +query regardless of how many actually match. + +Format: https://developer.themoviedb.org/docs/daily-id-exports -- a +gzipped file, one JSON object per line, e.g. for movies: +`{"adult":false,"id":3924,"original_title":"Blondie","popularity":1.61,"video":false}`. +Only `id` is used here; ingestion re-fetches full details per entity +through the normal provider pipeline (app/services/ingestion/), so +nothing else in the export line matters. +""" + +from __future__ import annotations + +import gzip +import io +import json +from datetime import date, timedelta + +import httpx + +from app.core.logging import get_logger + +logger = get_logger(__name__) + +_EXPORT_URL = "http://files.tmdb.org/p/exports/{kind}_ids_{date}.json.gz" + +# Aurora's entity_type -> the TMDb export file's own name for that kind. +_EXPORT_KIND: dict[str, str] = { + "movie": "movie", + "series": "tv_series", + "person": "person", + "collection": "collection", +} + + +class DailyExportUnavailableError(Exception): + pass + + +async def fetch_daily_export_ids(entity_type: str, *, client: httpx.AsyncClient) -> list[str]: + """Fetch and parse today's (or, failing that, yesterday's) TMDb daily + export for `entity_type`, returning every id as a string. TMDb + publishes a new file once a day at a fixed time (early UTC morning); + "today's" file may not exist yet depending on when this runs, so this + tries today first and falls back to yesterday rather than failing.""" + + kind = _EXPORT_KIND[entity_type] + last_status: int | None = None + for day_offset in (0, 1): + export_date = date.today() - timedelta(days=day_offset) + url = _EXPORT_URL.format(kind=kind, date=export_date.strftime("%m_%d_%Y")) + response = await client.get(url, follow_redirects=True) + if response.status_code == 200: + ids = _parse_export(response.content) + logger.info( + "tmdb_daily_export_fetched", entity_type=entity_type, url=url, count=len(ids) + ) + return ids + last_status = response.status_code + logger.info("tmdb_daily_export_not_found", url=url, status=response.status_code) + + raise DailyExportUnavailableError( + f"no TMDb daily export found for {entity_type} (tried today and yesterday, " + f"last status {last_status})" + ) + + +def _parse_export(raw_gzip: bytes) -> list[str]: + ids: list[str] = [] + with gzip.GzipFile(fileobj=io.BytesIO(raw_gzip)) as f: + for line in f: + line = line.strip() + if not line: + continue + ids.append(str(json.loads(line)["id"])) + return ids diff --git a/tests/unit/test_bulk_import.py b/tests/unit/test_bulk_import.py new file mode 100644 index 0000000..fbf3a1d --- /dev/null +++ b/tests/unit/test_bulk_import.py @@ -0,0 +1,85 @@ +"""Unit tests for app/services/bulk_import.py: parsing TMDb's daily ID +export format and the today-then-yesterday fallback, using +httpx.MockTransport so these run without real network access.""" + +from __future__ import annotations + +import gzip +import json + +import httpx +import pytest + +from app.services.bulk_import import DailyExportUnavailableError, fetch_daily_export_ids + + +def _gzip_ndjson(rows: list[dict]) -> bytes: + body = "\n".join(json.dumps(row) for row in rows).encode("utf-8") + return gzip.compress(body) + + +def _client_for(handler) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + +@pytest.mark.asyncio +async def test_fetch_daily_export_ids_parses_today_file(): + payload = _gzip_ndjson( + [ + {"adult": False, "id": 603, "original_title": "The Matrix", "popularity": 45.5}, + {"adult": False, "id": 604, "original_title": "The Matrix Reloaded", "popularity": 18.1}, + ] + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=payload) + + async with _client_for(handler) as client: + ids = await fetch_daily_export_ids("movie", client=client) + + assert ids == ["603", "604"] + + +@pytest.mark.asyncio +async def test_fetch_daily_export_ids_falls_back_to_yesterday_when_todays_missing(): + payload = _gzip_ndjson([{"id": 1396, "name": "Breaking Bad"}]) + calls: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(str(request.url)) + if len(calls) == 1: + return httpx.Response(404) + return httpx.Response(200, content=payload) + + async with _client_for(handler) as client: + ids = await fetch_daily_export_ids("series", client=client) + + assert ids == ["1396"] + assert len(calls) == 2 + assert "tv_series_ids" in calls[0] + + +@pytest.mark.asyncio +async def test_fetch_daily_export_ids_raises_when_neither_day_available(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + async with _client_for(handler) as client: + with pytest.raises(DailyExportUnavailableError): + await fetch_daily_export_ids("person", client=client) + + +@pytest.mark.asyncio +async def test_fetch_daily_export_ids_skips_blank_lines(): + body = b"\n".join( + [json.dumps({"id": 1}).encode(), b"", json.dumps({"id": 2}).encode(), b""] + ) + payload = gzip.compress(body) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=payload) + + async with _client_for(handler) as client: + ids = await fetch_daily_export_ids("collection", client=client) + + assert ids == ["1", "2"] diff --git a/tests/unit/test_tmdb_provider.py b/tests/unit/test_tmdb_provider.py new file mode 100644 index 0000000..ae6eb06 --- /dev/null +++ b/tests/unit/test_tmdb_provider.py @@ -0,0 +1,87 @@ +"""Unit tests for TMDbProvider's 429 (rate-limited) handling -- added +before running a full-catalog bulk import (app/scripts/bulk_import.py), +where sustained high request volume makes 429s an expected, recoverable +condition rather than a rare edge case. Uses httpx.MockTransport, and +Retry-After: 0 in the mocked responses so retries don't actually slow +these tests down.""" + +from __future__ import annotations + +from unittest.mock import Mock + +import httpx +import pytest + +from app.providers.base import ProviderRateLimitedError +from app.providers.tmdb import TMDbProvider, _tmdb_wait + + +def _client_for(handler) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="https://api.themoviedb.org/3") + + +def _provider(handler) -> TMDbProvider: + return TMDbProvider(api_key="test-key", client=_client_for(handler)) + + +@pytest.mark.asyncio +async def test_get_retries_after_a_single_429_and_succeeds(): + calls: list[int] = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(1) + if len(calls) == 1: + return httpx.Response(429, headers={"Retry-After": "0"}) + return httpx.Response(200, json={"id": 603, "title": "The Matrix"}) + + provider = _provider(handler) + data = await provider.get_movie("603") + + assert data == {"id": 603, "title": "The Matrix"} + assert len(calls) == 2 + + +@pytest.mark.asyncio +async def test_get_reraises_provider_rate_limited_error_after_exhausting_retries(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(429, headers={"Retry-After": "0"}) + + provider = _provider(handler) + + with pytest.raises(ProviderRateLimitedError): + await provider.get_movie("603") + + +@pytest.mark.asyncio +async def test_get_defaults_retry_after_when_header_is_not_a_number(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(429, headers={"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"}) + + provider = _provider(handler) + + with pytest.raises(ProviderRateLimitedError) as exc_info: + await provider.get_movie("603") + + assert exc_info.value.retry_after == 1.0 + + +def _fake_retry_state(exception: Exception | None) -> Mock: + state = Mock() + if exception is None: + state.outcome = None + else: + state.outcome.exception.return_value = exception + return state + + +def test_tmdb_wait_uses_retry_after_for_rate_limit_errors(): + state = _fake_retry_state(ProviderRateLimitedError("rate limited", retry_after=12.5)) + assert _tmdb_wait(state) == 12.5 + + +def test_tmdb_wait_falls_back_to_exponential_backoff_for_other_errors(): + state = _fake_retry_state(httpx.TransportError("boom")) + state.attempt_number = 1 + waited = _tmdb_wait(state) + assert isinstance(waited, float) + assert waited >= 0