NixReaperandClaude Sonnet 5
a52ac676ff
CI / lint (push) Canceled after 0s
CI / typecheck (push) Canceled after 0s
CI / unit-tests (push) Canceled after 0s
CI / integration-tests (push) Canceled after 0s
CI / container-build (push) Canceled after 0s
CI / dependency-audit (push) Canceled after 0s
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]>
82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
"""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
|