Add collection/franchise ingestion

Collections were bare stubs (name only) lazily created the moment a
movie's belongs_to_collection reference was seen. Adds a real
ingest_collection pipeline: TMDb's /collection/{id} for overview and
poster/backdrop artwork, plus ordered CollectionMovie membership --
linking only movies already ingested into Aurora, the same
never-fabricate-a-stub rule movie recommendations already follow.

get_or_create_collection now refreshes name/overview on an existing
stub instead of only creating, guarded so a bare movie-stub call
(overview=None) never clobbers real data a full ingestion already
filled in.

New POST /api/v1/admin/ingest/collection endpoint and
ingest_collection_task; GET /api/v1/collections/{id} needed no changes
since its read path (collection_service.py) already assembled the full
detail shape.

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

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)

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

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", [])

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:

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

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:

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",

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

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

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: <one of AURORA_ADMIN_API_KEYS>`. 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

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
 

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)

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
 

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
 

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 == []