Add series/season/episode ingestion pipeline

Movie and person ingestion existed, but the series/season/episode API
endpoints had no way to actually populate them from TMDb -- the
biggest remaining gap in Core 0.1's television scope.

- normalize_series/normalize_season in the metadata normalizer,
  including two TMDb quirks that would otherwise silently produce
  empty data: /tv nests keywords and alternative_titles under
  "results" (not "keywords"/"titles" like /movie), and series-level
  creators come from a separate created_by array, not the credits
  cast/crew lists.
- ingest_series upserts the Series row plus season *stubs* (number,
  title, air date, poster -- everything TMDb inlines on /tv/{id});
  ingest_season fetches and upserts one season's full episode list,
  and requires the series to already exist (TMDb has no single
  endpoint for a whole series' episodes, so this has to be a separate
  per-season call).
- Wired through the same job-lifecycle path as movie/person ingestion
  (runner.py, arq tasks, admin endpoints), with the shared boilerplate
  between them factored into runner._run_ingestion and
  admin.py:_enqueue rather than copy-pasted a third and fourth time.
- Unit tests for both normalizers (runnable without a DB) and
  integration tests for the full ingest_series -> ingest_season flow,
  including the series-must-exist-first error and idempotency on
  re-ingestion.

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

app/services/ingestion/series_ingestion.py

@@ -0,0 +1,419 @@
+"""Series/season/episode ingestion pipeline (section 15).
+
+Mirrors `movie_ingestion.py`'s flow (fetch -> normalize -> match -> upsert
+-> provenance) but split into two entry points because TMDb requires a
+separate API call per season for episode data:
+
+- `ingest_series` upserts the Series row and season *stubs* (number, title,
+  air date, poster — everything TMDb includes inline on `/tv/{id}`), but
+  not episodes.
+- `ingest_season` fetches and upserts one season's full episode list. It
+  requires the series to have been ingested already (episodes attach to an
+  Aurora series_id that has to already exist).
+"""
+
+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.artwork import Artwork
+from app.models.company import SeriesCompany
+from app.models.credit import Credit
+from app.models.enums import ArtworkType, CompanyType, MatchConfidence, SeriesStatus, VideoType
+from app.models.genre import SeriesGenre
+from app.models.keyword import SeriesKeyword
+from app.models.tv import Series
+from app.models.video import Video
+from app.providers.base import MetadataProvider, ProviderNotFoundError
+from app.repositories import reference_repository as refs
+from app.repositories import tv_repository
+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 (
+    NormalizedSeries,
+    normalize_season,
+    normalize_series,
+)
+
+logger = get_logger(__name__)
+
+_ARTWORK_TYPES = {t.value for t in ArtworkType}
+_VIDEO_TYPES = {t.value for t in VideoType}
+
+_LEADING_ARTICLES = ("the ", "a ", "an ")
+
+
+def _sort_title(title: str) -> str:
+    lowered = title.lower()
+    for article in _LEADING_ARTICLES:
+        if lowered.startswith(article):
+            return title[len(article) :].strip()
+    return title
+
+
+async def _replace_series_genres(session: AsyncSession, series_id: uuid.UUID, names: list[str]) -> None:
+    await session.execute(delete(SeriesGenre).where(SeriesGenre.series_id == series_id))
+    for name in names:
+        genre = await refs.get_or_create_genre(session, name)
+        session.add(SeriesGenre(series_id=series_id, genre_id=genre.id))
+
+
+async def _replace_series_keywords(
+    session: AsyncSession, series_id: uuid.UUID, names: list[str]
+) -> None:
+    await session.execute(delete(SeriesKeyword).where(SeriesKeyword.series_id == series_id))
+    for name in names:
+        keyword = await refs.get_or_create_keyword(session, name)
+        session.add(SeriesKeyword(series_id=series_id, keyword_id=keyword.id))
+
+
+async def _replace_series_companies(
+    session: AsyncSession,
+    series_id: uuid.UUID,
+    provider_id: uuid.UUID,
+    networks,
+    companies,
+) -> None:
+    await session.execute(delete(SeriesCompany).where(SeriesCompany.series_id == series_id))
+    for ref, company_type, role in (
+        *[(n, CompanyType.NETWORK, "network") for n in networks],
+        *[(c, CompanyType.PRODUCTION_COMPANY, "production") for c in companies],
+    ):
+        country_code = None
+        if ref.country_code:
+            country = await refs.get_or_create_country(session, ref.country_code, ref.country_code)
+            country_code = country.code
+        company = await refs.get_or_create_company(
+            session,
+            provider_id=provider_id,
+            external_id=ref.external_id,
+            name=ref.name,
+            company_type=company_type,
+            country_code=country_code,
+        )
+        session.add(SeriesCompany(series_id=series_id, company_id=company.id, role=role))
+
+
+async def _replace_series_credits(
+    session: AsyncSession,
+    series_id: uuid.UUID,
+    provider_id: uuid.UUID,
+    cast,
+    crew,
+    creators,
+) -> None:
+    await session.execute(
+        delete(Credit).where(Credit.entity_type == "series", Credit.entity_id == series_id)
+    )
+    for order, ref in enumerate(cast):
+        person = await refs.get_or_create_person(
+            session, provider_id=provider_id, external_id=ref.external_id, name=ref.name, gender=ref.gender
+        )
+        session.add(
+            Credit(
+                person_id=person.id,
+                entity_type="series",
+                entity_id=series_id,
+                department="Acting",
+                character=ref.character,
+                billing_order=ref.billing_order if ref.billing_order is not None else order,
+                provider_id=provider_id,
+            )
+        )
+    for ref in (*crew, *creators):
+        person = await refs.get_or_create_person(
+            session, provider_id=provider_id, external_id=ref.external_id, name=ref.name, gender=ref.gender
+        )
+        session.add(
+            Credit(
+                person_id=person.id,
+                entity_type="series",
+                entity_id=series_id,
+                department=ref.department or "Crew",
+                job=ref.job,
+                provider_id=provider_id,
+            )
+        )
+
+
+async def _replace_entity_artwork(
+    session: AsyncSession, entity_type: str, entity_id: uuid.UUID, provider_id: uuid.UUID, artwork_list
+) -> None:
+    await session.execute(
+        delete(Artwork).where(Artwork.entity_type == entity_type, Artwork.entity_id == entity_id)
+    )
+    for art in artwork_list:
+        art_type = art.artwork_type if art.artwork_type in _ARTWORK_TYPES else "poster"
+        session.add(
+            Artwork(
+                entity_type=entity_type,
+                entity_id=entity_id,
+                artwork_type=ArtworkType(art_type),
+                width=art.width,
+                height=art.height,
+                aspect_ratio=art.aspect_ratio,
+                provider_id=provider_id,
+                provider_asset_id=art.provider_asset_id,
+                original_url=art.url,
+                vote_average=art.vote_average,
+            )
+        )
+
+
+async def _replace_entity_videos(
+    session: AsyncSession, entity_type: str, entity_id: uuid.UUID, provider_id: uuid.UUID, videos
+) -> None:
+    await session.execute(
+        delete(Video).where(Video.entity_type == entity_type, Video.entity_id == entity_id)
+    )
+    for v in videos:
+        video_type = v.video_type if v.video_type in _VIDEO_TYPES else "other"
+        session.add(
+            Video(
+                entity_type=entity_type,
+                entity_id=entity_id,
+                video_type=VideoType(video_type),
+                provider_id=provider_id,
+                provider_video_id=v.provider_video_id,
+                site=v.site,
+                url=v.url,
+                is_official=v.is_official,
+                published_at=v.published_at,
+            )
+        )
+
+
+async def ingest_series(
+    session: AsyncSession,
+    provider: MetadataProvider,
+    external_id: str,
+    *,
+    ingestion_job_id: uuid.UUID | None = None,
+) -> uuid.UUID:
+    """Fetch, normalize, and upsert a series plus its season *stubs*
+    (without episodes — see `ingest_season`). Returns the Aurora series
+    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="series", external_id=external_id
+    )
+
+    raw = await provider.get_series(external_id)
+    normalized: NormalizedSeries = normalize_series(raw)
+
+    existing_series: Series | None = None
+    if aurora_id_existing is not None:
+        existing_series = await session.get(Series, aurora_id_existing)
+        if existing_series is None:
+            aurora_id_existing = None
+
+    series: Series
+    if existing_series is not None:
+        series = existing_series
+    else:
+        series = Series(title=normalized.title, sort_title=_sort_title(normalized.title))
+        session.add(series)
+        await session.flush()
+
+    series.title = normalized.title
+    series.original_title = normalized.original_title
+    series.sort_title = _sort_title(normalized.title)
+    series.overview = normalized.overview
+    series.tagline = normalized.tagline
+    series.status = SeriesStatus(normalized.status)
+    series.first_air_date = normalized.first_air_date
+    series.last_air_date = normalized.last_air_date
+    series.number_of_seasons = normalized.number_of_seasons
+    series.number_of_episodes = normalized.number_of_episodes
+    series.popularity = normalized.popularity
+    series.vote_average = normalized.vote_average
+    series.vote_count = normalized.vote_count
+    series.source_provider_id = tmdb_provider.id
+
+    if normalized.original_language:
+        language = await refs.get_or_create_language(
+            session, normalized.original_language, normalized.original_language.upper()
+        )
+        series.original_language_code = language.code
+
+    await session.flush()
+
+    await _replace_series_genres(session, series.id, normalized.genres)
+    await _replace_series_keywords(session, series.id, normalized.keywords)
+    await _replace_series_companies(
+        session, series.id, tmdb_provider.id, normalized.networks, normalized.companies
+    )
+    await _replace_series_credits(
+        session, series.id, tmdb_provider.id, normalized.cast, normalized.crew, normalized.creators
+    )
+    await _replace_entity_artwork(session, "series", series.id, tmdb_provider.id, normalized.artwork)
+    await _replace_entity_videos(session, "series", series.id, tmdb_provider.id, normalized.videos)
+
+    for stub in normalized.seasons:
+        season = await tv_repository.get_or_create_season(session, series.id, stub.season_number)
+        season.title = stub.title
+        season.overview = stub.overview
+        season.air_date = stub.air_date
+        season.episode_count = stub.episode_count
+        await session.flush()
+        await _replace_entity_artwork(
+            session,
+            "season",
+            season.id,
+            tmdb_provider.id,
+            [stub.poster] if stub.poster else [],
+        )
+
+    await refs.link_external_id(
+        session,
+        entity_type="series",
+        entity_id=series.id,
+        provider_id=tmdb_provider.id,
+        external_id=external_id,
+    )
+    for slug, ext_id in normalized.external_ids.items():
+        ext_provider = await get_or_create_provider(session, slug, slug.upper())
+        await refs.link_external_id(
+            session,
+            entity_type="series",
+            entity_id=series.id,
+            provider_id=ext_provider.id,
+            external_id=ext_id,
+        )
+
+    await upsert_provider_record(
+        session,
+        provider_id=tmdb_provider.id,
+        entity_type="series",
+        external_id=external_id,
+        raw_data=raw,
+        aurora_entity_id=series.id,
+        match_confidence=confidence if aurora_id_existing else MatchConfidence.NEW,
+        ingestion_job_id=ingestion_job_id,
+    )
+
+    logger.info("series_ingested", series_id=str(series.id), external_id=external_id)
+    return series.id
+
+
+async def _replace_episode_credits(
+    session: AsyncSession, episode_id: uuid.UUID, provider_id: uuid.UUID, guest_stars, crew
+) -> None:
+    await session.execute(
+        delete(Credit).where(Credit.entity_type == "episode", Credit.entity_id == episode_id)
+    )
+    for ref in guest_stars:
+        person = await refs.get_or_create_person(
+            session, provider_id=provider_id, external_id=ref.external_id, name=ref.name, gender=ref.gender
+        )
+        session.add(
+            Credit(
+                person_id=person.id,
+                entity_type="episode",
+                entity_id=episode_id,
+                department="Acting",
+                character=ref.character,
+                billing_order=ref.billing_order,
+                is_guest=True,
+                provider_id=provider_id,
+            )
+        )
+    for ref in crew:
+        person = await refs.get_or_create_person(
+            session, provider_id=provider_id, external_id=ref.external_id, name=ref.name, gender=ref.gender
+        )
+        session.add(
+            Credit(
+                person_id=person.id,
+                entity_type="episode",
+                entity_id=episode_id,
+                department=ref.department or "Crew",
+                job=ref.job,
+                provider_id=provider_id,
+            )
+        )
+
+
+async def ingest_season(
+    session: AsyncSession,
+    provider: MetadataProvider,
+    series_external_id: str,
+    season_number: int,
+    *,
+    ingestion_job_id: uuid.UUID | None = None,
+) -> uuid.UUID:
+    """Fetch, normalize, and upsert one season's full episode list. The
+    series must already be ingested (`ingest_series`) — episodes attach to
+    an Aurora series_id that has to exist first."""
+
+    tmdb_provider = await get_or_create_provider(session, provider.slug, provider.name)
+
+    series_id = await refs.find_entity_by_external_id(
+        session, tmdb_provider.id, "series", series_external_id
+    )
+    if series_id is None:
+        raise ProviderNotFoundError(
+            f"Cannot ingest season {season_number}: series tmdb:{series_external_id} has not "
+            "been ingested yet. Call POST /admin/ingest/series first."
+        )
+
+    raw = await provider.get_season(series_external_id, season_number)
+    normalized = normalize_season(raw)
+
+    season = await tv_repository.get_or_create_season(session, series_id, season_number)
+    season.title = normalized.title
+    season.overview = normalized.overview
+    season.air_date = normalized.air_date
+    season.episode_count = normalized.episode_count
+    await session.flush()
+
+    await _replace_entity_artwork(session, "season", season.id, tmdb_provider.id, normalized.artwork)
+
+    for ep in normalized.episodes:
+        episode = await tv_repository.get_or_create_episode(
+            session,
+            series_id=series_id,
+            season_id=season.id,
+            season_number=season_number,
+            episode_number=ep.episode_number,
+        )
+        episode.title = ep.title
+        episode.overview = ep.overview
+        episode.air_date = ep.air_date
+        episode.runtime_minutes = ep.runtime_minutes
+        await session.flush()
+
+        await _replace_episode_credits(session, episode.id, tmdb_provider.id, ep.guest_stars, ep.crew)
+        await _replace_entity_artwork(session, "episode", episode.id, tmdb_provider.id, ep.stills)
+        await refs.link_external_id(
+            session,
+            entity_type="episode",
+            entity_id=episode.id,
+            provider_id=tmdb_provider.id,
+            external_id=ep.external_id,
+        )
+
+    await upsert_provider_record(
+        session,
+        provider_id=tmdb_provider.id,
+        entity_type="season",
+        external_id=f"{series_external_id}:{season_number}",
+        raw_data=raw,
+        aurora_entity_id=season.id,
+        match_confidence=MatchConfidence.NEW,
+        ingestion_job_id=ingestion_job_id,
+    )
+
+    logger.info(
+        "season_ingested",
+        season_id=str(season.id),
+        series_external_id=series_external_id,
+        season_number=season_number,
+    )
+    return season.id