Add full-catalog bulk import from TMDb's daily ID exports

app/services/bulk_import.py fetches and parses TMDb's daily ID export
files (published once a day specifically for full-catalog sync, unlike
/discover which caps at 10,000 results per query) -- today's file with
a fallback to yesterday's, since TMDb publishes at a fixed time and
"today's" may not exist yet.

app/scripts/bulk_import.py is a standalone ops script (not an arq task,
since enqueueing millions of jobs would blow past arq's job_timeout)
that fans every id from the export into the existing per-entity
ingestion pipeline -- get_or_create_ingestion_job + the same arq task
each admin endpoint already uses, so a single id's behavior is
identical whether it came from the admin API or this script. Bounded
concurrency (50) on the enqueue phase; idempotency-keyed jobs make
interrupting and re-running the script safe.

Real TMDb catalog size (checked live): ~1.24M movies, 231K series,
4.9M people, 9.8K collections -- ~6.4M entities. At default worker
concurrency this is realistically weeks of continuous processing, not
hours, documented as such in README.md.

Also added 429 (rate-limited) handling to TMDbProvider before running
this at volume: previously a 429 fell into the generic >=400 branch and
failed the request outright with zero retry, since only httpx.TransportError
triggered the existing retry decorator. Now a 429 raises
ProviderRateLimitedError carrying the response's own Retry-After hint,
and tenacity's wait strategy honors that hint directly instead of
guessing via exponential backoff.

Co-Authored-By: Claude Sonnet 5 <[email protected]>

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

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()

app/scripts/__init__.py

@@ -0,0 +1,2 @@
+"""One-off operational scripts, run directly (`python -m app.scripts.<name>`)
+rather than through the API or arq worker task list."""

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()

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