"""Movie ingestion pipeline (section 15): Provider -> raw data -> normalize -> match -> upsert Aurora entity -> satellite tables (genres, keywords, companies, credits, artwork, videos, external IDs) -> provenance record. Idempotent: re-running for the same provider + external_id updates the same Aurora movie rather than creating a duplicate (zero duplicate ingestion records — section 2). Satellite associations are replaced wholesale on each sync rather than diffed, which is simple and safe but means their Aurora IDs (e.g. individual credit IDs) are not stable across re-syncs — see docs/architecture.md for the tradeoff. """ 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 MovieCompany from app.models.credit import Credit from app.models.enums import MatchConfidence, MovieStatus, VideoType from app.models.genre import MovieGenre from app.models.keyword import MovieKeyword from app.models.movie import Movie, MovieRecommendation from app.models.video import Video 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 NormalizedMovie, normalize_movie logger = get_logger(__name__) async def _replace_movie_genres(session: AsyncSession, movie_id: uuid.UUID, names: list[str]) -> None: await session.execute(delete(MovieGenre).where(MovieGenre.movie_id == movie_id)) for name in names: genre = await refs.get_or_create_genre(session, name) session.add(MovieGenre(movie_id=movie_id, genre_id=genre.id)) async def _replace_movie_keywords(session: AsyncSession, movie_id: uuid.UUID, names: list[str]) -> None: await session.execute(delete(MovieKeyword).where(MovieKeyword.movie_id == movie_id)) for name in names: keyword = await refs.get_or_create_keyword(session, name) session.add(MovieKeyword(movie_id=movie_id, keyword_id=keyword.id)) async def _replace_movie_companies( session: AsyncSession, movie_id: uuid.UUID, provider_id: uuid.UUID, companies ) -> None: await session.execute(delete(MovieCompany).where(MovieCompany.movie_id == movie_id)) for ref 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, country_code=country_code, ) session.add(MovieCompany(movie_id=movie_id, company_id=company.id, role=ref.role)) async def _replace_movie_credits( session: AsyncSession, movie_id: uuid.UUID, provider_id: uuid.UUID, cast, crew ) -> None: await session.execute( delete(Credit).where(Credit.entity_type == "movie", Credit.entity_id == movie_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="movie", entity_id=movie_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: 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="movie", entity_id=movie_id, department=ref.department or "Crew", job=ref.job, provider_id=provider_id, ) ) _VIDEO_TYPES = {t.value for t in VideoType} async def _replace_movie_videos( session: AsyncSession, movie_id: uuid.UUID, provider_id: uuid.UUID, videos ) -> None: await session.execute( delete(Video).where(Video.entity_type == "movie", Video.entity_id == movie_id) ) for v in videos: video_type = v.video_type if v.video_type in _VIDEO_TYPES else "other" session.add( Video( entity_type="movie", entity_id=movie_id, video_type=VideoType(video_type), provider_id=provider_id, provider_video_id=v.provider_video_id, site=v.site, url=v.url, language_code=None, country_code=None, is_official=v.is_official, published_at=v.published_at, ) ) async def _link_external_ids( session: AsyncSession, movie_id: uuid.UUID, tmdb_provider_id: uuid.UUID, normalized: NormalizedMovie ) -> None: await refs.link_external_id( session, entity_type="movie", entity_id=movie_id, provider_id=tmdb_provider_id, external_id=normalized.external_id, ) for slug, external_id in normalized.external_ids.items(): provider = await get_or_create_provider(session, slug, slug.upper()) await refs.link_external_id( session, entity_type="movie", entity_id=movie_id, provider_id=provider.id, external_id=external_id, ) async def ingest_movie( session: AsyncSession, provider: MetadataProvider, external_id: str, *, ingestion_job_id: uuid.UUID | None = None, ) -> uuid.UUID: """Fetch, normalize, and upsert a single movie. Returns the Aurora movie 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="movie", external_id=external_id ) raw = await provider.get_movie(external_id) normalized = normalize_movie(raw) existing_movie: Movie | None = None if aurora_id_existing is not None: existing_movie = await session.get(Movie, aurora_id_existing) if existing_movie is None: aurora_id_existing = None movie: Movie if existing_movie is not None: movie = existing_movie else: movie = Movie(title=normalized.title, sort_title=_sort_title(normalized.title)) session.add(movie) await session.flush() movie.title = normalized.title movie.original_title = normalized.original_title movie.sort_title = _sort_title(normalized.title) movie.overview = normalized.overview movie.tagline = normalized.tagline movie.status = MovieStatus(normalized.status) movie.primary_release_date = normalized.release_date movie.runtime_minutes = normalized.runtime_minutes movie.budget = normalized.budget movie.revenue = normalized.revenue movie.popularity = normalized.popularity movie.vote_average = normalized.vote_average movie.vote_count = normalized.vote_count movie.adult = normalized.adult movie.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() ) movie.original_language_code = language.code if normalized.collection: ext_id, name = normalized.collection collection = await refs.get_or_create_collection( session, provider_id=tmdb_provider.id, external_id=ext_id, name=name ) movie.collection_id = collection.id await session.flush() await _replace_movie_genres(session, movie.id, normalized.genres) await _replace_movie_keywords(session, movie.id, normalized.keywords) await _replace_movie_companies(session, movie.id, tmdb_provider.id, normalized.companies) await _replace_movie_credits(session, movie.id, tmdb_provider.id, normalized.cast, normalized.crew) await polymorphic.sync_artwork( session, entity_type="movie", entity_id=movie.id, provider_id=tmdb_provider.id, artwork_list=normalized.artwork, ) await _replace_movie_videos(session, movie.id, tmdb_provider.id, normalized.videos) await _link_external_ids(session, movie.id, tmdb_provider.id, normalized) await session.execute( delete(MovieRecommendation).where(MovieRecommendation.movie_id == movie.id) ) if normalized.recommendation_external_ids: rec_ids = await refs.find_entities_by_external_ids( session, tmdb_provider.id, "movie", normalized.recommendation_external_ids ) for rank, rec_movie_id in enumerate(rec_ids): session.add( MovieRecommendation( movie_id=movie.id, recommended_movie_id=rec_movie_id, rank=rank, source="tmdb", ) ) await refs.link_external_id( session, entity_type="movie", entity_id=movie.id, provider_id=tmdb_provider.id, external_id=external_id, ) await upsert_provider_record( session, provider_id=tmdb_provider.id, entity_type="movie", external_id=external_id, raw_data=raw, aurora_entity_id=movie.id, match_confidence=confidence if aurora_id_existing else MatchConfidence.NEW, ingestion_job_id=ingestion_job_id, ) logger.info("movie_ingested", movie_id=str(movie.id), external_id=external_id) return movie.id _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