From 3f6332559a1d92ea05d238aa8c047a0a6b09c11c Mon Sep 17 00:00:00 2001 From: NixReaper Date: Fri, 4 Sep 2026 19:43:56 -0500 Subject: [PATCH] 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 --- app/api/v1/admin.py | 91 ++++- app/repositories/tv_repository.py | 54 +++ app/services/ingestion/runner.py | 82 ++-- app/services/ingestion/series_ingestion.py | 419 +++++++++++++++++++++ app/services/metadata/normalizer.py | 348 +++++++++++++++++ app/workers/tasks.py | 19 +- app/workers/worker.py | 9 +- docs/api.md | 2 + docs/database.md | 11 + tests/integration/test_series_ingestion.py | 121 ++++++ tests/unit/test_normalizer.py | 181 ++++++++- 11 files changed, 1288 insertions(+), 49 deletions(-) create mode 100644 app/services/ingestion/series_ingestion.py create mode 100644 tests/integration/test_series_ingestion.py diff --git a/app/api/v1/admin.py b/app/api/v1/admin.py index e691cce..32eaf8b 100644 --- a/app/api/v1/admin.py +++ b/app/api/v1/admin.py @@ -34,6 +34,15 @@ class IngestPersonRequest(BaseModel): tmdb_id: str +class IngestSeriesRequest(BaseModel): + tmdb_id: str + + +class IngestSeasonRequest(BaseModel): + tmdb_id: str + season_number: int + + class IngestionEnqueuedOut(BaseModel): job_id: str arq_job_id: str @@ -51,16 +60,21 @@ class IngestionJobOut(BaseModel): result_aurora_id: str | None -@router.post("/ingest/movie", response_model=IngestionEnqueuedOut) -async def enqueue_movie_ingestion( - body: IngestMovieRequest, queue: ArqRedis = Depends(get_queue) +async def _enqueue( + queue: ArqRedis, + *, + job_type: str, + entity_type: str, + external_id: str, + task_name: str, + task_args: tuple = (), ) -> IngestionEnqueuedOut: - # Created synchronously (and idempotently — same tmdb_id reuses the same - # row) so the response's job_id is immediately valid against - # GET /ingest/jobs/{job_id}. arq's own job id is a separate, unrelated - # identifier used only for queue bookkeeping. - job_id = await get_or_create_ingestion_job("movie_sync", "movie", body.tmdb_id) - arq_job = await queue.enqueue_job("ingest_movie_task", str(job_id), body.tmdb_id) + # The IngestionJob row is created synchronously (and idempotently — the + # same external_id reuses the same row) so the response's job_id is + # immediately valid against GET /ingest/jobs/{job_id}. arq's own job id + # is a separate, unrelated identifier used only for queue bookkeeping. + job_id = await get_or_create_ingestion_job(job_type, entity_type, external_id) + arq_job = await queue.enqueue_job(task_name, str(job_id), *task_args) if arq_job is None: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, "Could not enqueue ingestion job (queue busy)." @@ -68,17 +82,62 @@ async def enqueue_movie_ingestion( return IngestionEnqueuedOut(job_id=str(job_id), arq_job_id=arq_job.job_id) +@router.post("/ingest/movie", response_model=IngestionEnqueuedOut) +async def enqueue_movie_ingestion( + body: IngestMovieRequest, queue: ArqRedis = Depends(get_queue) +) -> IngestionEnqueuedOut: + return await _enqueue( + queue, + job_type="movie_sync", + entity_type="movie", + external_id=body.tmdb_id, + task_name="ingest_movie_task", + task_args=(body.tmdb_id,), + ) + + @router.post("/ingest/person", response_model=IngestionEnqueuedOut) async def enqueue_person_ingestion( body: IngestPersonRequest, queue: ArqRedis = Depends(get_queue) ) -> IngestionEnqueuedOut: - job_id = await get_or_create_ingestion_job("person_sync", "person", body.tmdb_id) - arq_job = await queue.enqueue_job("ingest_person_task", str(job_id), body.tmdb_id) - if arq_job is None: - raise HTTPException( - status.HTTP_503_SERVICE_UNAVAILABLE, "Could not enqueue ingestion job (queue busy)." - ) - return IngestionEnqueuedOut(job_id=str(job_id), arq_job_id=arq_job.job_id) + return await _enqueue( + queue, + job_type="person_sync", + entity_type="person", + external_id=body.tmdb_id, + task_name="ingest_person_task", + task_args=(body.tmdb_id,), + ) + + +@router.post("/ingest/series", response_model=IngestionEnqueuedOut) +async def enqueue_series_ingestion( + body: IngestSeriesRequest, queue: ArqRedis = Depends(get_queue) +) -> IngestionEnqueuedOut: + return await _enqueue( + queue, + job_type="series_sync", + entity_type="series", + external_id=body.tmdb_id, + task_name="ingest_series_task", + task_args=(body.tmdb_id,), + ) + + +@router.post("/ingest/season", response_model=IngestionEnqueuedOut) +async def enqueue_season_ingestion( + body: IngestSeasonRequest, queue: ArqRedis = Depends(get_queue) +) -> IngestionEnqueuedOut: + # Season ingestion requires the series to already be ingested — see + # app/services/ingestion/series_ingestion.py:ingest_season. + return await _enqueue( + queue, + job_type="season_sync", + entity_type="season", + external_id=f"{body.tmdb_id}:{body.season_number}", + task_name="ingest_season_task", + task_args=(body.tmdb_id, body.season_number), + ) @router.get("/ingest/jobs/{job_id}", response_model=IngestionJobOut) diff --git a/app/repositories/tv_repository.py b/app/repositories/tv_repository.py index 9614579..7c389ca 100644 --- a/app/repositories/tv_repository.py +++ b/app/repositories/tv_repository.py @@ -19,6 +19,60 @@ async def get_series_by_id(session: AsyncSession, series_id: uuid.UUID) -> Serie return result.scalar_one_or_none() +async def get_season_by_number( + session: AsyncSession, series_id: uuid.UUID, season_number: int +) -> Season | None: + stmt = select(Season).where( + Season.series_id == series_id, Season.season_number == season_number + ) + result = await session.execute(stmt) + return result.scalar_one_or_none() + + +async def get_or_create_season( + session: AsyncSession, series_id: uuid.UUID, season_number: int +) -> Season: + season = await get_season_by_number(session, series_id, season_number) + if season is not None: + return season + season = Season(series_id=series_id, season_number=season_number) + session.add(season) + await session.flush() + return season + + +async def get_episode_by_number( + session: AsyncSession, season_id: uuid.UUID, episode_number: int +) -> Episode | None: + stmt = select(Episode).where( + Episode.season_id == season_id, Episode.episode_number == episode_number + ) + result = await session.execute(stmt) + return result.scalar_one_or_none() + + +async def get_or_create_episode( + session: AsyncSession, + *, + series_id: uuid.UUID, + season_id: uuid.UUID, + season_number: int, + episode_number: int, +) -> Episode: + episode = await get_episode_by_number(session, season_id, episode_number) + if episode is not None: + return episode + episode = Episode( + series_id=series_id, + season_id=season_id, + season_number=season_number, + episode_number=episode_number, + ) + session.add(episode) + await session.flush() + return episode + + async def get_seasons_for_series(session: AsyncSession, series_id: uuid.UUID) -> list[Season]: stmt = ( select(Season) diff --git a/app/services/ingestion/runner.py b/app/services/ingestion/runner.py index dedb033..3250052 100644 --- a/app/services/ingestion/runner.py +++ b/app/services/ingestion/runner.py @@ -1,19 +1,21 @@ """Orchestrates one ingestion job end-to-end: create/track the `IngestionJob` row, run the provider fetch + normalize + upsert pipeline, and record success/failure. Used by both the admin trigger endpoint (section 17) and -the arq worker task (section 15) so the two entry points share identical +the arq worker task (section 15) so every entry point shares identical job-lifecycle and idempotency behavior. """ from __future__ import annotations import uuid +from collections.abc import Awaitable, Callable from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import session_scope from app.core.logging import get_logger from app.models.provider import IngestionJob +from app.providers.base import MetadataProvider from app.providers.tmdb import TMDbProvider from app.repositories.provider_repository import ( create_ingestion_job, @@ -25,6 +27,7 @@ from app.repositories.provider_repository import ( ) 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 logger = get_logger(__name__) @@ -64,9 +67,14 @@ async def get_or_create_ingestion_job(job_type: str, entity_type: str, external_ return job.id -async def run_movie_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: - """Run the pipeline for an `IngestionJob` row that already exists (see - `get_or_create_ingestion_job`) — this never creates the row itself.""" +async def _run_ingestion( + job_id: uuid.UUID, + label: str, + work: Callable[[AsyncSession, MetadataProvider], Awaitable[uuid.UUID]], +) -> uuid.UUID: + """Shared job-lifecycle wrapper: mark running, run `work` in its own + session/transaction, mark succeeded/failed. `work` does the actual + fetch+normalize+upsert and returns the resulting Aurora entity id.""" async with session_scope() as session: job = await _require_job(session, job_id) @@ -75,13 +83,13 @@ async def run_movie_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: provider = TMDbProvider() try: async with session_scope() as session: - movie_id = await ingest_movie(session, provider, external_id, ingestion_job_id=job_id) + result_id = await work(session, provider) async with session_scope() as session: job = await _require_job(session, job_id) - await mark_job_succeeded(session, job, movie_id) - return movie_id + await mark_job_succeeded(session, job, result_id) + return result_id except Exception as exc: - logger.error("movie_ingestion_failed", external_id=external_id, error=str(exc)) + logger.error(f"{label}_failed", job_id=str(job_id), error=str(exc)) async with session_scope() as session: job = await _require_job(session, job_id) await mark_job_failed(session, job, str(exc)) @@ -90,27 +98,43 @@ async def run_movie_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: await provider.aclose() +async def run_movie_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: + return await _run_ingestion( + job_id, + "movie_ingestion", + lambda session, provider: ingest_movie( + session, provider, external_id, ingestion_job_id=job_id + ), + ) + + async def run_person_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: - """Run the pipeline for an `IngestionJob` row that already exists (see - `get_or_create_ingestion_job`) — this never creates the row itself.""" + return await _run_ingestion( + job_id, + "person_ingestion", + lambda session, provider: ingest_person( + session, provider, external_id, ingestion_job_id=job_id + ), + ) - async with session_scope() as session: - job = await _require_job(session, job_id) - await mark_job_running(session, job) - provider = TMDbProvider() - try: - async with session_scope() as session: - person_id = await ingest_person(session, provider, external_id, ingestion_job_id=job_id) - async with session_scope() as session: - job = await _require_job(session, job_id) - await mark_job_succeeded(session, job, person_id) - return person_id - except Exception as exc: - logger.error("person_ingestion_failed", external_id=external_id, error=str(exc)) - async with session_scope() as session: - job = await _require_job(session, job_id) - await mark_job_failed(session, job, str(exc)) - raise - finally: - await provider.aclose() +async def run_series_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID: + return await _run_ingestion( + job_id, + "series_ingestion", + lambda session, provider: ingest_series( + 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: + return await _run_ingestion( + job_id, + "season_ingestion", + lambda session, provider: ingest_season( + session, provider, series_external_id, season_number, ingestion_job_id=job_id + ), + ) diff --git a/app/services/ingestion/series_ingestion.py b/app/services/ingestion/series_ingestion.py new file mode 100644 index 0000000..c05965b --- /dev/null +++ b/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 diff --git a/app/services/metadata/normalizer.py b/app/services/metadata/normalizer.py index 8892b0a..fba4ae2 100644 --- a/app/services/metadata/normalizer.py +++ b/app/services/metadata/normalizer.py @@ -339,3 +339,351 @@ def normalize_person(raw: dict) -> NormalizedPerson: artwork=artwork, external_ids=external_ids, ) + + +_SERIES_STATUS_MAP = { + "Planned": "planned", + "Pilot": "pilot", + "In Production": "in_production", + "Returning Series": "returning_series", + "Ended": "ended", + "Canceled": "canceled", +} + + +@dataclass +class NormalizedSeasonStub: + """A season summary as it appears embedded in a series payload — enough + to create/update the Season row, but without episodes (those require a + separate per-season fetch — see `normalize_season`).""" + + season_number: int + title: str | None + overview: str | None + air_date: date | None + episode_count: int | None + poster: NormalizedArtwork | None + + +@dataclass +class NormalizedSeries: + external_id: str + title: str + original_title: str | None + overview: str | None + tagline: str | None + status: str + original_language: str | None + first_air_date: date | None + last_air_date: date | None + number_of_seasons: int | None + number_of_episodes: int | None + popularity: float | None + vote_average: float | None + vote_count: int | None + genres: list[str] = field(default_factory=list) + keywords: list[str] = field(default_factory=list) + alternate_titles: list[str] = field(default_factory=list) + networks: list[NormalizedCompanyRef] = field(default_factory=list) + companies: list[NormalizedCompanyRef] = field(default_factory=list) + creators: list[NormalizedPersonRef] = field(default_factory=list) + cast: list[NormalizedPersonRef] = field(default_factory=list) + crew: list[NormalizedPersonRef] = field(default_factory=list) + artwork: list[NormalizedArtwork] = field(default_factory=list) + videos: list[NormalizedVideo] = field(default_factory=list) + external_ids: dict[str, str] = field(default_factory=dict) + seasons: list[NormalizedSeasonStub] = field(default_factory=list) + + +def normalize_series(raw: dict) -> NormalizedSeries: + credits_ = raw.get("credits") or {} + images = raw.get("images") or {} + videos = (raw.get("videos") or {}).get("results") or [] + # Unlike /movie, TMDb's /tv keywords and alternative_titles nest under + # "results", not "keywords"/"titles". + keywords = (raw.get("keywords") or {}).get("results") or [] + alt_titles = (raw.get("alternative_titles") or {}).get("results") or [] + ext_ids = raw.get("external_ids") or {} + + external_ids: dict[str, str] = {} + if ext_ids.get("imdb_id"): + external_ids["imdb"] = ext_ids["imdb_id"] + if ext_ids.get("tvdb_id"): + external_ids["tvdb"] = str(ext_ids["tvdb_id"]) + if ext_ids.get("wikidata_id"): + external_ids["wikidata"] = ext_ids["wikidata_id"] + + cast = [ + NormalizedPersonRef( + external_id=str(c["id"]), + name=c["name"], + profile_path=_image_url(c.get("profile_path")), + department="Acting", + character=c.get("character"), + billing_order=c.get("order"), + gender=_gender_label(c.get("gender")), + ) + for c in credits_.get("cast", []) + ] + crew = [ + NormalizedPersonRef( + external_id=str(c["id"]), + name=c["name"], + profile_path=_image_url(c.get("profile_path")), + department=c.get("department") or "Crew", + job=c.get("job"), + gender=_gender_label(c.get("gender")), + ) + for c in credits_.get("crew", []) + ] + creators = [ + NormalizedPersonRef( + external_id=str(c["id"]), + name=c["name"], + profile_path=_image_url(c.get("profile_path")), + department="Creator", + gender=_gender_label(c.get("gender")), + ) + for c in raw.get("created_by", []) + ] + + networks = [ + NormalizedCompanyRef( + external_id=str(n["id"]), name=n["name"], country_code=n.get("origin_country") or None + ) + for n in raw.get("networks", []) + ] + companies = [ + NormalizedCompanyRef( + external_id=str(pc["id"]), + name=pc["name"], + country_code=pc.get("origin_country") or None, + ) + for pc in raw.get("production_companies", []) + ] + + 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, + ) + ) + if raw.get("backdrop_path"): + artwork.append( + NormalizedArtwork( + "backdrop", + _image_url_required(raw["backdrop_path"]), + raw["backdrop_path"], + None, + None, + None, + None, + ) + ) + for kind, tmdb_key in (("poster", "posters"), ("backdrop", "backdrops"), ("logo", "logos")): + for img in images.get(tmdb_key, []): + artwork.append( + NormalizedArtwork( + kind, + _image_url_required(img["file_path"]), + img["file_path"], + img.get("width"), + img.get("height"), + img.get("aspect_ratio"), + img.get("iso_639_1"), + img.get("vote_average"), + ) + ) + + normalized_videos = [ + NormalizedVideo( + video_type=(v.get("type") or "other").lower().replace(" ", "_"), + provider_video_id=v["id"], + site=v.get("site"), + url=_video_url(v), + language=v.get("iso_639_1"), + country=v.get("iso_3166_1"), + is_official=bool(v.get("official")), + published_at=_parse_date(v.get("published_at", "")[:10] if v.get("published_at") else None), + ) + for v in videos + ] + + seasons = [ + NormalizedSeasonStub( + season_number=s["season_number"], + title=s.get("name"), + overview=s.get("overview") or None, + air_date=_parse_date(s.get("air_date")), + episode_count=s.get("episode_count"), + poster=( + NormalizedArtwork( + "poster", _image_url_required(s["poster_path"]), s["poster_path"], + None, None, None, None, + ) + if s.get("poster_path") + else None + ), + ) + for s in raw.get("seasons", []) + # TMDb includes a "season 0" for specials; keep it (it's a real, + # commonly-referenced season) but drop entries missing a number. + if s.get("season_number") is not None + ] + + return NormalizedSeries( + external_id=str(raw["id"]), + title=raw.get("name") or raw.get("original_name") or "", + original_title=raw.get("original_name"), + overview=raw.get("overview") or None, + tagline=raw.get("tagline") or None, + status=_SERIES_STATUS_MAP.get(raw.get("status", "Returning Series"), "returning_series"), + original_language=raw.get("original_language"), + first_air_date=_parse_date(raw.get("first_air_date")), + last_air_date=_parse_date(raw.get("last_air_date")), + number_of_seasons=raw.get("number_of_seasons"), + number_of_episodes=raw.get("number_of_episodes"), + popularity=raw.get("popularity"), + vote_average=raw.get("vote_average"), + vote_count=raw.get("vote_count"), + genres=[g["name"] for g in raw.get("genres", [])], + keywords=[k["name"] for k in keywords], + alternate_titles=[t["title"] for t in alt_titles if t.get("title")], + networks=networks, + companies=companies, + creators=creators, + cast=cast, + crew=crew, + artwork=artwork, + videos=normalized_videos, + external_ids=external_ids, + seasons=seasons, + ) + + +@dataclass +class NormalizedEpisode: + external_id: str + episode_number: int + title: str | None + overview: str | None + air_date: date | None + runtime_minutes: int | None + cast: list[NormalizedPersonRef] = field(default_factory=list) + guest_stars: list[NormalizedPersonRef] = field(default_factory=list) + crew: list[NormalizedPersonRef] = field(default_factory=list) + stills: list[NormalizedArtwork] = field(default_factory=list) + + +@dataclass +class NormalizedSeason: + season_number: int + title: str | None + overview: str | None + air_date: date | None + episode_count: int | None + artwork: list[NormalizedArtwork] = field(default_factory=list) + episodes: list[NormalizedEpisode] = field(default_factory=list) + external_ids: dict[str, str] = field(default_factory=dict) + + +def _episode_person_refs(entries: list[dict]) -> list[NormalizedPersonRef]: + return [ + NormalizedPersonRef( + external_id=str(e["id"]), + name=e["name"], + profile_path=_image_url(e.get("profile_path")), + department="Acting", + character=e.get("character"), + billing_order=e.get("order"), + gender=_gender_label(e.get("gender")), + ) + for e in entries + ] + + +def normalize_season(raw: dict) -> NormalizedSeason: + images = raw.get("images") or {} + ext_ids = raw.get("external_ids") or {} + external_ids: dict[str, str] = {} + if ext_ids.get("tvdb_id"): + external_ids["tvdb"] = str(ext_ids["tvdb_id"]) + + 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, + ) + ) + for img in images.get("posters", []): + artwork.append( + NormalizedArtwork( + "poster", + _image_url_required(img["file_path"]), + img["file_path"], + img.get("width"), + img.get("height"), + img.get("aspect_ratio"), + img.get("iso_639_1"), + img.get("vote_average"), + ) + ) + + episodes = [] + for ep in raw.get("episodes", []): + crew = [ + NormalizedPersonRef( + external_id=str(c["id"]), + name=c["name"], + profile_path=_image_url(c.get("profile_path")), + department=c.get("department") or "Crew", + job=c.get("job"), + gender=_gender_label(c.get("gender")), + ) + for c in ep.get("crew", []) + ] + stills: list[NormalizedArtwork] = [] + if ep.get("still_path"): + stills.append( + NormalizedArtwork( + "still", _image_url_required(ep["still_path"]), ep["still_path"], + None, None, None, None, + ) + ) + episodes.append( + NormalizedEpisode( + external_id=str(ep["id"]), + episode_number=ep["episode_number"], + title=ep.get("name"), + overview=ep.get("overview") or None, + air_date=_parse_date(ep.get("air_date")), + runtime_minutes=ep.get("runtime"), + # TMDb's per-episode payload has no separate "main cast" + # list distinct from guest_stars/crew — the series' regular + # cast is assumed, not repeated per episode. + cast=[], + guest_stars=_episode_person_refs(ep.get("guest_stars", [])), + crew=crew, + stills=stills, + ) + ) + + return NormalizedSeason( + season_number=raw["season_number"], + title=raw.get("name"), + overview=raw.get("overview") or None, + air_date=_parse_date(raw.get("air_date")), + episode_count=len(episodes) or None, + artwork=artwork, + episodes=episodes, + external_ids=external_ids, + ) diff --git a/app/workers/tasks.py b/app/workers/tasks.py index 2910cf4..a6ef234 100644 --- a/app/workers/tasks.py +++ b/app/workers/tasks.py @@ -13,7 +13,12 @@ import uuid from typing import Any from app.core.logging import get_logger -from app.services.ingestion.runner import run_movie_ingestion, run_person_ingestion +from app.services.ingestion.runner import ( + run_movie_ingestion, + run_person_ingestion, + run_season_ingestion, + run_series_ingestion, +) logger = get_logger(__name__) @@ -26,3 +31,15 @@ async def ingest_movie_task(ctx: dict[str, Any], job_id: str, external_id: str) async def ingest_person_task(ctx: dict[str, Any], job_id: str, external_id: str) -> str: person_id = await run_person_ingestion(uuid.UUID(job_id), external_id) return str(person_id) + + +async def ingest_series_task(ctx: dict[str, Any], job_id: str, external_id: str) -> str: + series_id = await run_series_ingestion(uuid.UUID(job_id), external_id) + return str(series_id) + + +async def ingest_season_task( + ctx: dict[str, Any], job_id: str, series_external_id: str, season_number: int +) -> str: + season_id = await run_season_ingestion(uuid.UUID(job_id), series_external_id, season_number) + return str(season_id) diff --git a/app/workers/worker.py b/app/workers/worker.py index 8e762a0..7a183a5 100644 --- a/app/workers/worker.py +++ b/app/workers/worker.py @@ -13,7 +13,12 @@ from arq.connections import RedisSettings from app.core.config import get_settings from app.core.logging import configure_logging, get_logger -from app.workers.tasks import ingest_movie_task, ingest_person_task +from app.workers.tasks import ( + ingest_movie_task, + ingest_person_task, + ingest_season_task, + ingest_series_task, +) logger = get_logger(__name__) @@ -28,7 +33,7 @@ async def shutdown(ctx: dict) -> None: class WorkerSettings: - functions = [ingest_movie_task, ingest_person_task] + functions = [ingest_movie_task, ingest_person_task, ingest_series_task, ingest_season_task] on_startup = startup on_shutdown = shutdown redis_settings = RedisSettings.from_dsn(get_settings().REDIS_URL) diff --git a/docs/api.md b/docs/api.md index 9520205..0ca8f4f 100644 --- a/docs/api.md +++ b/docs/api.md @@ -120,6 +120,8 @@ you want network-level isolation too. |---|---|---|---| | POST | `/api/v1/admin/ingest/movie` | `{"tmdb_id": "603"}` | Enqueues an arq job; returns `{"job_id": "...", "status": "enqueued"}` immediately (section 15: ingestion never runs inline in a request). | | POST | `/api/v1/admin/ingest/person` | `{"tmdb_id": "6384"}` | Same, for a person profile. | +| POST | `/api/v1/admin/ingest/series` | `{"tmdb_id": "1396"}` | Ingests the series and season *stubs* (number, title, air date, poster) — no episodes yet. | +| 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. | | 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 1f9e9e4..836dd4c 100644 --- a/docs/database.md +++ b/docs/database.md @@ -163,3 +163,14 @@ indexes on the alternate-title tables. See links) are replaced wholesale on each re-sync rather than diffed, so their Aurora IDs are not stable across re-ingestion — see `app/services/ingestion/movie_ingestion.py`. +- Seasons don't get their own `external_ids` row (identified instead by the + natural `(series_id, season_number)` key) — episodes do, since TMDb gives + each episode a distinct global id. Revisit if a provider ever needs a + season-level external reference Aurora doesn't already have another way + to resolve. +- `ingest_series` only upserts season *stubs* (number, title, air date, + poster) — episodes require a separate `ingest_season` call per season + (TMDb has no single endpoint returning a whole series' episodes), and + that call requires the series to already exist. There's no orchestration + yet that walks all of a series' seasons automatically after + `ingest_series` succeeds. diff --git a/tests/integration/test_series_ingestion.py b/tests/integration/test_series_ingestion.py new file mode 100644 index 0000000..936486a --- /dev/null +++ b/tests/integration/test_series_ingestion.py @@ -0,0 +1,121 @@ +"""Integration tests for the series/season ingestion pipeline (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.models.tv import Episode, Season, Series +from app.providers.base import MetadataProvider, ProviderNotFoundError +from app.services.ingestion.series_ingestion import ingest_season, ingest_series +from tests.unit.test_normalizer import FAKE_SEASON, FAKE_SERIES + + +class FakeTMDbSeriesProvider(MetadataProvider): + slug = "tmdb" + name = "The Movie Database (TMDb)" + + def __init__(self, series_payload: dict[str, Any], season_payload: dict[str, Any]) -> None: + self._series_payload = series_payload + self._season_payload = season_payload + + async def get_movie(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + + async def get_series(self, external_id: str) -> dict[str, Any]: + return self._series_payload + + async def get_season(self, series_external_id: str, season_number: int) -> dict[str, Any]: + return self._season_payload + + async def get_person(self, external_id: str) -> dict[str, Any]: + raise NotImplementedError + + async def search_movie(self, query: str) -> list[dict[str, Any]]: + raise NotImplementedError + + +@pytest.mark.asyncio +async def test_ingest_series_creates_series_and_season_stubs(db_session: AsyncSession): + provider = FakeTMDbSeriesProvider(FAKE_SERIES, FAKE_SEASON) + + series_id = await ingest_series(db_session, provider, "1396") + await db_session.commit() + + series = await db_session.get(Series, series_id) + assert series is not None + assert series.title == "Ozymandias" + assert series.status.value == "ended" + + seasons = ( + (await db_session.execute(select(Season).where(Season.series_id == series_id))) + .scalars() + .all() + ) + season_numbers = sorted(s.season_number for s in seasons) + assert season_numbers == [0, 1] + # Season stubs have no episodes yet — that's ingest_season's job. + assert all( + (await db_session.execute(select(Episode).where(Episode.season_id == s.id))) + .scalars() + .first() + is None + for s in seasons + ) + + +@pytest.mark.asyncio +async def test_ingest_season_requires_series_ingested_first(db_session: AsyncSession): + provider = FakeTMDbSeriesProvider(FAKE_SERIES, FAKE_SEASON) + + with pytest.raises(ProviderNotFoundError): + await ingest_season(db_session, provider, "1396", 1) + + +@pytest.mark.asyncio +async def test_ingest_season_creates_episodes_with_credits_and_stills(db_session: AsyncSession): + provider = FakeTMDbSeriesProvider(FAKE_SERIES, FAKE_SEASON) + + await ingest_series(db_session, provider, "1396") + await db_session.flush() + + season_id = await ingest_season(db_session, provider, "1396", 1) + await db_session.commit() + + episodes = ( + (await db_session.execute(select(Episode).where(Episode.season_id == season_id))) + .scalars() + .all() + ) + assert len(episodes) == 2 + pilot = next(e for e in episodes if e.episode_number == 1) + assert pilot.title == "Pilot" + assert pilot.runtime_minutes == 58 + + +@pytest.mark.asyncio +async def test_ingest_season_is_idempotent(db_session: AsyncSession): + provider = FakeTMDbSeriesProvider(FAKE_SERIES, FAKE_SEASON) + + await ingest_series(db_session, provider, "1396") + await db_session.flush() + + first_season_id = await ingest_season(db_session, provider, "1396", 1) + await db_session.flush() + second_season_id = await ingest_season(db_session, provider, "1396", 1) + await db_session.commit() + + assert first_season_id == second_season_id + episodes = ( + (await db_session.execute(select(Episode).where(Episode.season_id == first_season_id))) + .scalars() + .all() + ) + # Re-running must not duplicate episode rows. + assert len(episodes) == 2 diff --git a/tests/unit/test_normalizer.py b/tests/unit/test_normalizer.py index 6ec7f20..b095e11 100644 --- a/tests/unit/test_normalizer.py +++ b/tests/unit/test_normalizer.py @@ -6,7 +6,12 @@ from __future__ import annotations from datetime import date -from app.services.metadata.normalizer import normalize_movie, normalize_person +from app.services.metadata.normalizer import ( + normalize_movie, + normalize_person, + normalize_season, + normalize_series, +) FAKE_MOVIE = { "id": 603, @@ -182,3 +187,177 @@ def test_normalize_person_profile_artwork_deduplicated(): # not be inserted twice. paths = [a.provider_asset_id for a in person.artwork] assert paths.count("/keanu.jpg") == 1 + + +FAKE_SERIES = { + "id": 1396, + "name": "Ozymandias", + "original_name": "Ozymandias", + "overview": "A chemistry teacher turns to crime.", + "tagline": "Change the equation.", + "status": "Ended", + "original_language": "en", + "first_air_date": "2008-01-20", + "last_air_date": "2013-09-29", + "number_of_seasons": 5, + "number_of_episodes": 62, + "popularity": 120.5, + "vote_average": 8.9, + "vote_count": 15000, + "genres": [{"id": 18, "name": "Drama"}, {"id": 80, "name": "Crime"}], + "created_by": [{"id": 66633, "name": "Vince Gilligan", "gender": 2}], + "networks": [{"id": 174, "name": "AMC", "origin_country": "US"}], + "production_companies": [ + {"id": 2605, "name": "Gran Via Productions", "origin_country": "US"} + ], + "poster_path": "/poster.jpg", + "backdrop_path": "/backdrop.jpg", + "credits": { + "cast": [ + { + "id": 17419, + "name": "Bryan Cranston", + "character": "Walter White", + "order": 0, + "gender": 2, + } + ], + "crew": [], + }, + "images": {"posters": [], "backdrops": [], "logos": []}, + "videos": {"results": []}, + "keywords": {"results": [{"id": 1, "name": "drug dealer"}]}, + "alternative_titles": {"results": [{"title": "Breaking Bad", "iso_3166_1": "US"}]}, + "external_ids": {"imdb_id": "tt0903747", "tvdb_id": 81189}, + "seasons": [ + { + "season_number": 1, + "name": "Season 1", + "overview": "The first season.", + "air_date": "2008-01-20", + "episode_count": 7, + "poster_path": "/season1.jpg", + }, + { + # TMDb includes a "specials" season with number 0 — must not be + # dropped. + "season_number": 0, + "name": "Specials", + "overview": None, + "air_date": None, + "episode_count": 3, + "poster_path": None, + }, + ], +} + +FAKE_SEASON = { + "season_number": 1, + "name": "Season 1", + "overview": "The first season.", + "air_date": "2008-01-20", + "poster_path": "/season1.jpg", + "images": {"posters": []}, + "external_ids": {"tvdb_id": 12345}, + "episodes": [ + { + "id": 62085, + "episode_number": 1, + "name": "Pilot", + "overview": "Walter White begins his transformation.", + "air_date": "2008-01-20", + "runtime": 58, + "still_path": "/still1.jpg", + "crew": [ + {"id": 66633, "name": "Vince Gilligan", "department": "Directing", "job": "Director", "gender": 2} + ], + "guest_stars": [ + {"id": 99999, "name": "Some Guest", "character": "Random Guy", "order": 5, "gender": 2} + ], + }, + { + "id": 62086, + "episode_number": 2, + "name": "Cat's in the Bag...", + "overview": None, + "air_date": "2008-01-27", + "runtime": 48, + "still_path": None, + "crew": [], + "guest_stars": [], + }, + ], +} + + +def test_normalize_series_maps_core_fields(): + series = normalize_series(FAKE_SERIES) + assert series.external_id == "1396" + assert series.title == "Ozymandias" + assert series.status == "ended" + assert series.first_air_date == date(2008, 1, 20) + assert series.number_of_seasons == 5 + + +def test_normalize_series_maps_networks_vs_production_companies(): + series = normalize_series(FAKE_SERIES) + assert [n.name for n in series.networks] == ["AMC"] + assert [c.name for c in series.companies] == ["Gran Via Productions"] + + +def test_normalize_series_maps_creators_separately_from_crew(): + series = normalize_series(FAKE_SERIES) + assert len(series.creators) == 1 + assert series.creators[0].name == "Vince Gilligan" + assert series.creators[0].department == "Creator" + assert all(c.department != "Creator" for c in series.crew) + + +def test_normalize_series_keywords_and_alt_titles_use_tv_shaped_keys(): + # TMDb nests these under "results" for /tv, unlike "keywords"/"titles" + # for /movie — a normalizer bug here would silently return empty lists. + series = normalize_series(FAKE_SERIES) + assert series.keywords == ["drug dealer"] + assert series.alternate_titles == ["Breaking Bad"] + + +def test_normalize_series_keeps_season_zero_specials(): + series = normalize_series(FAKE_SERIES) + season_numbers = sorted(s.season_number for s in series.seasons) + assert season_numbers == [0, 1] + + +def test_normalize_series_external_ids_include_tvdb(): + series = normalize_series(FAKE_SERIES) + assert series.external_ids["imdb"] == "tt0903747" + assert series.external_ids["tvdb"] == "81189" + + +def test_normalize_season_maps_episodes(): + season = normalize_season(FAKE_SEASON) + assert season.season_number == 1 + assert len(season.episodes) == 2 + assert season.episodes[0].external_id == "62085" + assert season.episodes[0].title == "Pilot" + assert season.episodes[0].runtime_minutes == 58 + + +def test_normalize_season_separates_guest_stars_from_main_cast(): + season = normalize_season(FAKE_SEASON) + episode = season.episodes[0] + assert episode.cast == [] + assert len(episode.guest_stars) == 1 + assert episode.guest_stars[0].character == "Random Guy" + + +def test_normalize_season_maps_episode_crew_and_stills(): + season = normalize_season(FAKE_SEASON) + episode = season.episodes[0] + assert episode.crew[0].job == "Director" + assert len(episode.stills) == 1 + assert episode.stills[0].artwork_type == "still" + + +def test_normalize_season_handles_episode_without_still(): + season = normalize_season(FAKE_SEASON) + assert season.episodes[1].stills == []