Aurora-Core/app/services/ingestion/movie_ingestion.py

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