diff --git a/app/api/v1/admin.py b/app/api/v1/admin.py index 6a2bec3..b13aa9f 100644 --- a/app/api/v1/admin.py +++ b/app/api/v1/admin.py @@ -49,6 +49,10 @@ class IngestSeasonRequest(BaseModel): season_number: int +class IngestCollectionRequest(BaseModel): + tmdb_id: str + + class IngestionEnqueuedOut(BaseModel): job_id: str arq_job_id: str @@ -146,6 +150,24 @@ async def enqueue_season_ingestion( ) +@router.post("/ingest/collection", response_model=IngestionEnqueuedOut) +async def enqueue_collection_ingestion( + body: IngestCollectionRequest, queue: ArqRedis = Depends(get_queue) +) -> IngestionEnqueuedOut: + # Membership only links movies already ingested into Aurora (never a + # stub) -- see app/services/ingestion/collection_ingestion.py -- so + # ingesting the franchise's movies first (or re-running this afterward) + # is what fills the list in. + return await _enqueue( + queue, + job_type="collection_sync", + entity_type="collection", + external_id=body.tmdb_id, + task_name="ingest_collection_task", + task_args=(body.tmdb_id,), + ) + + @router.get("/ingest/jobs/{job_id}", response_model=IngestionJobOut) async def get_ingestion_job_status( job_id: uuid.UUID, session: AsyncSession = Depends(get_db) diff --git a/app/providers/base.py b/app/providers/base.py index f11a559..b1c9cf5 100644 --- a/app/providers/base.py +++ b/app/providers/base.py @@ -49,6 +49,11 @@ class MetadataProvider(ABC): async def get_person(self, external_id: str) -> dict[str, Any]: """Fetch raw person data, including external IDs and images.""" + @abstractmethod + async def get_collection(self, external_id: str) -> dict[str, Any]: + """Fetch raw collection/franchise data: name, overview, poster/ + backdrop, and its member movies (TMDb's "parts").""" + @abstractmethod async def search_movie(self, query: str) -> list[dict[str, Any]]: """Search this provider's catalog for movies matching `query`. Used diff --git a/app/providers/tmdb.py b/app/providers/tmdb.py index a5f4165..f8caf42 100644 --- a/app/providers/tmdb.py +++ b/app/providers/tmdb.py @@ -80,6 +80,9 @@ class TMDbProvider(MetadataProvider): async def get_person(self, external_id: str) -> dict[str, Any]: return await self._get(f"/person/{external_id}", {"append_to_response": _PERSON_APPEND}) + async def get_collection(self, external_id: str) -> dict[str, Any]: + return await self._get(f"/collection/{external_id}") + async def search_movie(self, query: str) -> list[dict[str, Any]]: data = await self._get("/search/movie", {"query": query}) return data.get("results", []) diff --git a/app/repositories/reference_repository.py b/app/repositories/reference_repository.py index 691ada4..68fdf83 100644 --- a/app/repositories/reference_repository.py +++ b/app/repositories/reference_repository.py @@ -208,6 +208,15 @@ async def get_or_create_collection( existing = await _lookup() if existing is not None: + existing.name = name + # Only overwrite overview when the caller actually has one -- a + # movie's bare `belongs_to_collection` reference passes no overview + # at all, and must not blank out real data a full `ingest_collection` + # run already filled in (see get_or_create_person for the same + # partial-data guard). + if overview is not None: + existing.overview = overview + await session.flush() return existing async def _create() -> Collection: diff --git a/app/services/ingestion/collection_ingestion.py b/app/services/ingestion/collection_ingestion.py new file mode 100644 index 0000000..21b69f8 --- /dev/null +++ b/app/services/ingestion/collection_ingestion.py @@ -0,0 +1,95 @@ +"""Collection/franchise ingestion (section 15). + +Movie ingestion already creates a bare `Collection` stub (name only) the +moment it sees a `belongs_to_collection` reference (see +`movie_ingestion.py`). This module is the real sync: fetching TMDb's +`/collection/{id}` payload for overview, poster/backdrop artwork, and the +franchise's member movies. + +Membership (`CollectionMovie`) is only ever linked for movies Aurora has +already ingested -- the same "never fabricate an entity from a bare +reference" rule `ingest_movie` applies to recommendations. A newly-ingested +collection can list zero movies if none of its parts have been ingested yet; +re-running `ingest_collection` after ingesting more of them fills the list +in, since it's a full re-sync, not an append. +""" + +from __future__ import annotations + +import uuid + +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.logging import get_logger +from app.models.collection import CollectionMovie +from app.models.enums import MatchConfidence +from app.providers.base import MetadataProvider +from app.repositories import polymorphic +from app.repositories import reference_repository as refs +from app.repositories.provider_repository import get_or_create_provider, upsert_provider_record +from app.services.metadata.matcher import match_entity +from app.services.metadata.normalizer import normalize_collection + +logger = get_logger(__name__) + + +async def ingest_collection( + session: AsyncSession, + provider: MetadataProvider, + external_id: str, + *, + ingestion_job_id: uuid.UUID | None = None, +) -> uuid.UUID: + """Fetch, normalize, and upsert a single collection. Returns the Aurora + collection UUID (new or existing).""" + + tmdb_provider = await get_or_create_provider(session, provider.slug, provider.name) + + aurora_id_existing, confidence = await match_entity( + session, provider_id=tmdb_provider.id, entity_type="collection", external_id=external_id + ) + + raw = await provider.get_collection(external_id) + normalized = normalize_collection(raw) + + collection = await refs.get_or_create_collection( + session, + provider_id=tmdb_provider.id, + external_id=external_id, + name=normalized.name, + overview=normalized.overview, + ) + + await polymorphic.sync_artwork( + session, + entity_type="collection", + entity_id=collection.id, + provider_id=tmdb_provider.id, + artwork_list=normalized.artwork, + ) + + await session.execute( + delete(CollectionMovie).where(CollectionMovie.collection_id == collection.id) + ) + movie_ids = await refs.find_entities_by_external_ids( + session, tmdb_provider.id, "movie", normalized.part_external_ids + ) + for sort_order, movie_id in enumerate(movie_ids): + session.add( + CollectionMovie(collection_id=collection.id, movie_id=movie_id, sort_order=sort_order) + ) + + await upsert_provider_record( + session, + provider_id=tmdb_provider.id, + entity_type="collection", + external_id=external_id, + raw_data=raw, + aurora_entity_id=collection.id, + match_confidence=confidence if aurora_id_existing else MatchConfidence.NEW, + ingestion_job_id=ingestion_job_id, + ) + + logger.info("collection_ingested", collection_id=str(collection.id), external_id=external_id) + return collection.id diff --git a/app/services/ingestion/runner.py b/app/services/ingestion/runner.py index 761037b..c076ec2 100644 --- a/app/services/ingestion/runner.py +++ b/app/services/ingestion/runner.py @@ -33,6 +33,7 @@ from app.repositories.provider_repository import ( ) from app.repositories.tv_repository import get_seasons_for_series, get_series_companies from app.services.cache import build_key, cache_delete +from app.services.ingestion.collection_ingestion import ingest_collection from app.services.ingestion.movie_ingestion import ingest_movie from app.services.ingestion.person_ingestion import ingest_person from app.services.ingestion.series_ingestion import ingest_season, ingest_series @@ -164,6 +165,16 @@ async def run_series_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID ) +async def run_collection_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: + return await _run_ingestion( + job_id, + "collection_ingestion", + lambda session, provider: ingest_collection( + session, provider, external_id, ingestion_job_id=job_id + ), + ) + + async def run_season_ingestion( job_id: uuid.UUID, series_external_id: str, season_number: int ) -> uuid.UUID: diff --git a/app/services/metadata/normalizer.py b/app/services/metadata/normalizer.py index 6d6374e..a71e9cc 100644 --- a/app/services/metadata/normalizer.py +++ b/app/services/metadata/normalizer.py @@ -138,6 +138,61 @@ class NormalizedMovie: recommendation_external_ids: list[str] = field(default_factory=list) +@dataclass +class NormalizedCollection: + external_id: str + name: str + overview: str | None + artwork: list[NormalizedArtwork] = field(default_factory=list) + # TMDb's "parts", ordered by release date rather than TMDb's own array + # order (which isn't guaranteed to be chronological) -- see + # normalize_collection. + part_external_ids: list[str] = field(default_factory=list) + + +def normalize_collection(raw: dict) -> NormalizedCollection: + artwork: list[NormalizedArtwork] = [] + if raw.get("poster_path"): + artwork.append( + NormalizedArtwork( + "poster", + _image_url_required(raw["poster_path"]), + raw["poster_path"], + None, + None, + None, + None, + is_primary=True, + ) + ) + if raw.get("backdrop_path"): + artwork.append( + NormalizedArtwork( + "backdrop", + _image_url_required(raw["backdrop_path"]), + raw["backdrop_path"], + None, + None, + None, + None, + is_primary=True, + ) + ) + + parts = sorted( + raw.get("parts") or [], + key=lambda p: _parse_date(p.get("release_date")) or date.max, + ) + + return NormalizedCollection( + external_id=str(raw["id"]), + name=raw["name"], + overview=raw.get("overview") or None, + artwork=artwork, + part_external_ids=[str(p["id"]) for p in parts], + ) + + _STATUS_MAP = { "Rumored": "rumored", "Planned": "planned", diff --git a/app/workers/tasks.py b/app/workers/tasks.py index 2e51010..8854c21 100644 --- a/app/workers/tasks.py +++ b/app/workers/tasks.py @@ -28,6 +28,7 @@ from app.services.ingestion.runner import ( get_series_company_ids, get_series_season_ids, get_series_season_numbers, + run_collection_ingestion, run_movie_ingestion, run_person_ingestion, run_season_ingestion, @@ -102,6 +103,12 @@ async def ingest_season_task( return str(season_id) +async def ingest_collection_task(ctx: dict[str, Any], job_id: str, external_id: str) -> str: + collection_id = await run_collection_ingestion(uuid.UUID(job_id), external_id) + await _cascade_artwork_downloads(ctx, "collection", [collection_id]) + return str(collection_id) + + async def download_artwork_task(ctx: dict[str, Any], artwork_id: str) -> bool: async with session_scope() as session: return await download_artwork(session, get_artwork_storage(), uuid.UUID(artwork_id)) diff --git a/app/workers/worker.py b/app/workers/worker.py index db156b4..488c8a1 100644 --- a/app/workers/worker.py +++ b/app/workers/worker.py @@ -15,6 +15,7 @@ from app.core.config import get_settings from app.core.logging import configure_logging, get_logger from app.workers.tasks import ( download_artwork_task, + ingest_collection_task, ingest_movie_task, ingest_person_task, ingest_season_task, @@ -39,6 +40,7 @@ class WorkerSettings: ingest_person_task, ingest_series_task, ingest_season_task, + ingest_collection_task, download_artwork_task, ] on_startup = startup diff --git a/docs/api.md b/docs/api.md index 58c0730..e94719f 100644 --- a/docs/api.md +++ b/docs/api.md @@ -127,6 +127,12 @@ network and a production company are both `Company` rows distinguished by | GET | `/api/v1/collections/{aurora_id}` | | GET | `/api/v1/companies/{aurora_id}` | +Collection detail's `movies` list only includes movies already ingested +into Aurora — `POST /api/v1/admin/ingest/collection` links a franchise's +member movies purely by external-id lookup, the same never-fabricate-a-stub +rule `recommendations` follows, so a franchise fills in as more of its +movies get ingested rather than all at once. + ## Administrative (section 17: "separately protected") Require header `X-API-Key: `. Never exposed to @@ -139,6 +145,7 @@ you want network-level isolation too. | POST | `/api/v1/admin/ingest/person` | `{"tmdb_id": "6384"}` | Same, for a person profile. | | POST | `/api/v1/admin/ingest/series` | `{"tmdb_id": "1396", "include_seasons": false}` | Ingests the series and season *stubs* (number, title, air date, poster) — no episodes yet. With `include_seasons: true`, also enqueues an `ingest/season` job for every season once the series finishes (opt-in: some shows have dozens of seasons, each a separate TMDb call). | | POST | `/api/v1/admin/ingest/season` | `{"tmdb_id": "1396", "season_number": 1}` | Ingests one season's full episode list. Requires the series to already be ingested (`ingest/series` first) — fails with a clear error otherwise. | +| POST | `/api/v1/admin/ingest/collection` | `{"tmdb_id": "2344"}` | Ingests a franchise's own metadata (overview, poster/backdrop) and links its already-ingested member movies (see "Collections / Companies" above). Movie ingestion also lazily creates a bare name-only stub the first time it sees a `belongs_to_collection` reference — this is what actually fills it in. | | GET | `/api/v1/admin/ingest/jobs/{job_id}` | — | Poll job status: `pending` → `running` → `succeeded`/`failed`/`retrying`. | ## Errors diff --git a/docs/database.md b/docs/database.md index 11a5b4e..18dd913 100644 --- a/docs/database.md +++ b/docs/database.md @@ -60,7 +60,14 @@ and a movie studio have the same shape (name, logo, country, parent company), so a separate `Network` entity would just duplicate columns. `collections` (franchises) are first-class, with their own artwork and `collection_relations` for shared-universe links between collections -(section 9). +(section 9). A collection starts as a bare name-only stub the first time +movie ingestion sees a `belongs_to_collection` reference; its real overview, +poster/backdrop, and ordered `collection_movies` membership are filled in by +`app/services/ingestion/collection_ingestion.py:ingest_collection` +(`POST /api/v1/admin/ingest/collection`), which — like recommendations — +only links movies Aurora has already ingested rather than fabricating stubs +for the rest of the franchise. `collection_relations` and `collection_series` +exist in the schema for future use but nothing populates them yet. ## Artwork / Videos diff --git a/tests/integration/test_collection_ingestion.py b/tests/integration/test_collection_ingestion.py new file mode 100644 index 0000000..cb577d7 --- /dev/null +++ b/tests/integration/test_collection_ingestion.py @@ -0,0 +1,182 @@ +"""Integration tests for collection/franchise ingestion (section 15): a fake +provider stands in for TMDb so these run without network access, while +still exercising the real normalize -> match -> upsert -> provenance path +against PostgreSQL.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.ids import aurora_id +from app.models.artwork import Artwork +from app.models.collection import Collection, CollectionMovie +from app.providers.base import MetadataProvider +from app.services import collection_service +from app.services.ingestion.collection_ingestion import ingest_collection +from app.services.ingestion.movie_ingestion import ingest_movie +from tests.unit.test_normalizer import FAKE_COLLECTION, FAKE_MOVIE + + +class FakeTMDbCollectionProvider(MetadataProvider): + slug = "tmdb" + name = "The Movie Database (TMDb)" + + def __init__( + self, movie_payloads: dict[str, dict[str, Any]], collection_payload: dict[str, Any] + ) -> None: + self._movie_payloads = movie_payloads + self._collection_payload = collection_payload + + async def get_movie(self, external_id: str) -> dict[str, Any]: + return self._movie_payloads[external_id] + + async def get_series(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + + async def get_season(self, series_external_id: str, season_number: int) -> dict[str, Any]: + raise NotImplementedError + + async def get_person(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + + async def get_collection(self, external_id: str) -> dict[str, Any]: + return self._collection_payload + + async def search_movie(self, query: str) -> list[dict[str, Any]]: + raise NotImplementedError + + +@pytest.mark.asyncio +async def test_ingest_collection_only_links_already_ingested_movies(db_session: AsyncSession): + """Two of FAKE_COLLECTION's four parts (603, 604) are ingested first; + 605 and 999 are not -- CollectionMovie must link only the two Aurora + already knows about, in release-date order, never fabricate a stub for + the other two (section 15).""" + + provider = FakeTMDbCollectionProvider( + movie_payloads={ + "603": {**FAKE_MOVIE, "id": 603, "title": "The Matrix"}, + "604": {**FAKE_MOVIE, "id": 604, "title": "The Matrix Reloaded"}, + }, + collection_payload=FAKE_COLLECTION, + ) + + await ingest_movie(db_session, provider, "603") + await ingest_movie(db_session, provider, "604") + await db_session.flush() + + collection_id = await ingest_collection(db_session, provider, "2344") + await db_session.commit() + + collection = await db_session.get(Collection, collection_id) + assert collection is not None + assert collection.name == "The Matrix Collection" + assert collection.overview == "A group of films..." + + links = ( + ( + await db_session.execute( + select(CollectionMovie) + .where(CollectionMovie.collection_id == collection_id) + .order_by(CollectionMovie.sort_order.asc()) + ) + ) + .scalars() + .all() + ) + assert len(links) == 2 + + detail = await collection_service.get_collection_detail( + db_session, aurora_id("collection", str(collection_id)) + ) + assert [m.title for m in detail.movies] == ["The Matrix", "The Matrix Reloaded"] + assert detail.poster_path is not None + assert detail.poster_path.endswith("/collection-poster.jpg") + + +@pytest.mark.asyncio +async def test_ingest_collection_is_idempotent_and_does_not_duplicate_links( + db_session: AsyncSession, +): + provider = FakeTMDbCollectionProvider( + movie_payloads={"603": {**FAKE_MOVIE, "id": 603, "title": "The Matrix"}}, + collection_payload=FAKE_COLLECTION, + ) + + await ingest_movie(db_session, provider, "603") + await db_session.flush() + + first_id = await ingest_collection(db_session, provider, "2344") + await db_session.flush() + second_id = await ingest_collection(db_session, provider, "2344") + await db_session.commit() + + assert first_id == second_id + + all_collections = (await db_session.execute(select(Collection))).scalars().all() + assert len(all_collections) == 1 + + links = ( + ( + await db_session.execute( + select(CollectionMovie).where(CollectionMovie.collection_id == first_id) + ) + ) + .scalars() + .all() + ) + assert len(links) == 1 + + +@pytest.mark.asyncio +async def test_ingest_collection_does_not_clobber_overview_from_movie_stub( + db_session: AsyncSession, +): + """A movie referencing this collection via belongs_to_collection creates + a bare stub (name only, overview=None) *after* the real ingest_collection + already ran -- get_or_create_collection must not blank the real overview + back to None just because the stub call has none to offer.""" + + provider = FakeTMDbCollectionProvider( + movie_payloads={ + "603": {**FAKE_MOVIE, "id": 603, "title": "The Matrix"}, + "604": {**FAKE_MOVIE, "id": 604, "title": "The Matrix Reloaded"}, + }, + collection_payload=FAKE_COLLECTION, + ) + + collection_id = await ingest_collection(db_session, provider, "2344") + await db_session.flush() + + await ingest_movie(db_session, provider, "604") + await db_session.commit() + + collection = await db_session.get(Collection, collection_id) + assert collection.overview == "A group of films..." + + +@pytest.mark.asyncio +async def test_ingest_collection_ingests_poster_and_backdrop_artwork(db_session: AsyncSession): + provider = FakeTMDbCollectionProvider(movie_payloads={}, collection_payload=FAKE_COLLECTION) + + collection_id = await ingest_collection(db_session, provider, "2344") + await db_session.commit() + + artwork = ( + ( + await db_session.execute( + select(Artwork).where( + Artwork.entity_type == "collection", Artwork.entity_id == collection_id + ) + ) + ) + .scalars() + .all() + ) + types = {a.artwork_type.value for a in artwork} + assert types == {"poster", "backdrop"} + assert all(a.is_primary for a in artwork) diff --git a/tests/integration/test_movie_ingestion.py b/tests/integration/test_movie_ingestion.py index 61c8059..a871e95 100644 --- a/tests/integration/test_movie_ingestion.py +++ b/tests/integration/test_movie_ingestion.py @@ -44,6 +44,9 @@ class FakeTMDbProvider(MetadataProvider): async def get_person(self, external_id: str) -> dict[str, Any]: raise NotImplementedError + async def get_collection(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + async def search_movie(self, query: str) -> list[dict[str, Any]]: raise NotImplementedError diff --git a/tests/integration/test_series_ingestion.py b/tests/integration/test_series_ingestion.py index b288042..acfc99f 100644 --- a/tests/integration/test_series_ingestion.py +++ b/tests/integration/test_series_ingestion.py @@ -39,6 +39,9 @@ class FakeTMDbSeriesProvider(MetadataProvider): async def get_person(self, external_id: str) -> dict[str, Any]: raise NotImplementedError + async def get_collection(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + async def search_movie(self, query: str) -> list[dict[str, Any]]: raise NotImplementedError diff --git a/tests/unit/test_normalizer.py b/tests/unit/test_normalizer.py index dd14bdb..17a2f34 100644 --- a/tests/unit/test_normalizer.py +++ b/tests/unit/test_normalizer.py @@ -7,6 +7,7 @@ from __future__ import annotations from datetime import date from app.services.metadata.normalizer import ( + normalize_collection, normalize_movie, normalize_person, normalize_season, @@ -403,3 +404,50 @@ def test_normalize_season_maps_episode_crew_and_stills(): def test_normalize_season_handles_episode_without_still(): season = normalize_season(FAKE_SEASON) assert season.episodes[1].stills == [] + + +FAKE_COLLECTION = { + "id": 2344, + "name": "The Matrix Collection", + "overview": "A group of films...", + "poster_path": "/collection-poster.jpg", + "backdrop_path": "/collection-backdrop.jpg", + "parts": [ + {"id": 605, "title": "The Matrix Revolutions", "release_date": "2003-11-05"}, + {"id": 603, "title": "The Matrix", "release_date": "1999-03-30"}, + {"id": 604, "title": "The Matrix Reloaded", "release_date": "2003-05-15"}, + {"id": 999, "title": "Untitled Prequel", "release_date": ""}, + ], +} + + +def test_normalize_collection_maps_basic_fields(): + collection = normalize_collection(FAKE_COLLECTION) + assert collection.external_id == "2344" + assert collection.name == "The Matrix Collection" + assert collection.overview == "A group of films..." + + +def test_normalize_collection_maps_poster_and_backdrop_as_primary_artwork(): + collection = normalize_collection(FAKE_COLLECTION) + poster = next(a for a in collection.artwork if a.artwork_type == "poster") + backdrop = next(a for a in collection.artwork if a.artwork_type == "backdrop") + assert poster.is_primary is True + assert poster.url.endswith("/collection-poster.jpg") + assert backdrop.is_primary is True + assert backdrop.url.endswith("/collection-backdrop.jpg") + + +def test_normalize_collection_orders_parts_by_release_date_not_array_order(): + collection = normalize_collection(FAKE_COLLECTION) + # Input order is [605, 603, 604, 999] -- release-date order puts the + # undated "999" entry last rather than fabricating a sort position. + assert collection.part_external_ids == ["603", "604", "605", "999"] + + +def test_normalize_collection_handles_missing_overview_and_parts(): + minimal = {"id": 1, "name": "Minimal Collection"} + collection = normalize_collection(minimal) + assert collection.overview is None + assert collection.artwork == [] + assert collection.part_external_ids == []