"""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.company import SeriesCompany from app.models.credit import Credit from app.models.enums import 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 polymorphic, tv_repository 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 ( NormalizedSeries, normalize_season, normalize_series, ) logger = get_logger(__name__) _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_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 polymorphic.sync_artwork( session, entity_type="series", entity_id=series.id, provider_id=tmdb_provider.id, artwork_list=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 polymorphic.sync_artwork( session, entity_type="season", entity_id=season.id, provider_id=tmdb_provider.id, artwork_list=[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 polymorphic.sync_artwork( session, entity_type="season", entity_id=season.id, provider_id=tmdb_provider.id, artwork_list=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 polymorphic.sync_artwork( session, entity_type="episode", entity_id=episode.id, provider_id=tmdb_provider.id, artwork_list=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