Aurora-Core/app/repositories/polymorphic.py

"""Shared data access for the polymorphic satellite tables — credits,
artwork, videos, external IDs — that attach to more than one entity type via
an (entity_type, entity_id) pair rather than a typed foreign key.
"""

from __future__ import annotations

import uuid
from collections.abc import Sequence
from datetime import date
from typing import Protocol

from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.models.artwork import Artwork
from app.models.credit import Credit
from app.models.enums import ArtworkType
from app.models.external_id import ExternalID
from app.models.movie import ReleaseDate
from app.models.provider import Provider
from app.models.video import Video
from app.models.watch_provider import WatchProvider, WatchProviderOffering
from app.repositories import reference_repository as refs

_VALID_ARTWORK_TYPES = {t.value for t in ArtworkType}


class _NormalizedReleaseDateLike(Protocol):
    """Structural type for
    app.services.metadata.normalizer.NormalizedReleaseDate — see
    _NormalizedArtworkLike above for why a repository takes a Protocol
    instead of importing the services-layer dataclass directly."""

    country_code: str
    release_date: date
    release_type: str | None
    certification: tuple[str, str] | None
    note: str | None


class _NormalizedWatchProviderRefLike(Protocol):
    """Structural type for
    app.services.metadata.normalizer.NormalizedWatchProviderRef."""

    external_id: str
    name: str
    logo_path: str | None


class _NormalizedWatchProviderOfferingLike(Protocol):
    """Structural type for
    app.services.metadata.normalizer.NormalizedWatchProviderOffering. The
    nested `provider` is a `@property` rather than a plain annotation --
    mypy otherwise treats a Protocol attribute as read-write and requires
    invariant (exact) matching, which a genuinely different concrete class
    can never satisfy even when structurally identical; a property makes
    this member covariant, matching how it's actually used here (read
    only, never reassigned)."""

    country_code: str
    monetization_type: str
    display_priority: int | None
    link: str | None

    @property
    def provider(self) -> _NormalizedWatchProviderRefLike: ...


class _NormalizedArtworkLike(Protocol):
    """Structural type for app.services.metadata.normalizer.NormalizedArtwork
    — a repository shouldn't import from the services layer, so this
    describes the shape `sync_artwork` needs instead of the concrete class.
    """

    artwork_type: str
    url: str
    provider_asset_id: str | None
    width: int | None
    height: int | None
    aspect_ratio: float | None
    vote_average: float | None
    is_primary: bool


async def get_external_ids(
    session: AsyncSession, entity_type: str, entity_id: uuid.UUID
) -> dict[str, str]:
    stmt = (
        select(Provider.slug, ExternalID.external_id)
        .join(Provider, Provider.id == ExternalID.provider_id)
        .where(ExternalID.entity_type == entity_type, ExternalID.entity_id == entity_id)
    )
    result = await session.execute(stmt)
    return {slug: external_id for slug, external_id in result.all()}


async def get_external_ids_bulk(
    session: AsyncSession, entity_type: str, entity_ids: list[uuid.UUID]
) -> dict[uuid.UUID, dict[str, str]]:
    if not entity_ids:
        return {}
    stmt = (
        select(ExternalID.entity_id, Provider.slug, ExternalID.external_id)
        .join(Provider, Provider.id == ExternalID.provider_id)
        .where(ExternalID.entity_type == entity_type, ExternalID.entity_id.in_(entity_ids))
    )
    result = await session.execute(stmt)
    out: dict[uuid.UUID, dict[str, str]] = {eid: {} for eid in entity_ids}
    for entity_id, slug, external_id in result.all():
        out.setdefault(entity_id, {})[slug] = external_id
    return out


async def get_primary_artwork_ids(
    session: AsyncSession, entity_type: str, entity_ids: list[uuid.UUID]
) -> list[uuid.UUID]:
    """IDs of artwork rows marked `is_primary` for any of `entity_ids` — the
    small subset of a title's images (TMDb's own selected poster/backdrop/
    profile) that get auto-downloaded for local hosting, out of the dozens
    of alternates ingestion also stores. See app/services/artwork/."""

    if not entity_ids:
        return []
    stmt = select(Artwork.id).where(
        Artwork.entity_type == entity_type,
        Artwork.entity_id.in_(entity_ids),
        Artwork.is_primary.is_(True),
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())


async def sync_artwork(
    session: AsyncSession,
    *,
    entity_type: str,
    entity_id: uuid.UUID,
    provider_id: uuid.UUID,
    artwork_list: Sequence[_NormalizedArtworkLike],
) -> None:
    """Upsert an entity's artwork rows keyed by `provider_asset_id` (TMDb's
    stable per-image file path), instead of deleting and recreating them on
    every re-sync.

    This matters specifically because of `app/services/artwork/` (section
    10's self-hosting): a delete-then-recreate replace would generate a
    fresh Artwork.id with `local_path=None` every time a title gets
    re-ingested, silently throwing away every previously-downloaded local
    copy and forcing a full re-download on each sync — defeating the point
    of caching them locally at all. Rows for images no longer present in
    `artwork_list` (dropped from the provider's alternates) are still
    deleted, same as before.
    """

    existing = (
        (
            await session.execute(
                select(Artwork).where(
                    Artwork.entity_type == entity_type, Artwork.entity_id == entity_id
                )
            )
        )
        .scalars()
        .all()
    )
    by_asset_id: dict[str, Artwork] = {
        row.provider_asset_id: row for row in existing if row.provider_asset_id is not None
    }
    # The incoming batch is authoritative for which image(s) are primary
    # this sync — reset up front so a provider dropping its "primary" pick
    # on an existing image actually clears here, not just adds.
    for row in existing:
        row.is_primary = False

    seen_asset_ids: set[str] = set()
    for art in artwork_list:
        art_type = art.artwork_type if art.artwork_type in _VALID_ARTWORK_TYPES else "poster"
        asset_id = art.provider_asset_id
        matched = by_asset_id.get(asset_id) if asset_id else None
        if matched is not None:
            matched.artwork_type = ArtworkType(art_type)
            matched.width = art.width
            matched.height = art.height
            matched.aspect_ratio = art.aspect_ratio
            matched.vote_average = art.vote_average
            # OR, never a plain overwrite: the *same* image commonly
            # appears twice in one provider fetch (TMDb's selected
            # poster_path also shows up in the general posters alternates
            # list) — once any occurrence in this batch says "primary",
            # a later, non-primary occurrence of that same image must not
            # un-set it.
            matched.is_primary = matched.is_primary or art.is_primary
            matched.original_url = art.url
            # local_path/checksum deliberately untouched — see docstring.
        else:
            new_row = 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=asset_id,
                original_url=art.url,
                vote_average=art.vote_average,
                is_primary=art.is_primary,
            )
            session.add(new_row)
            if asset_id:
                # Register immediately so a second occurrence of this same
                # new asset_id later in the batch updates this row instead
                # of inserting a duplicate.
                by_asset_id[asset_id] = new_row
        if asset_id:
            seen_asset_ids.add(asset_id)

    for row in existing:
        if row.provider_asset_id not in seen_asset_ids:
            await session.delete(row)

    await session.flush()


async def get_artwork(
    session: AsyncSession,
    entity_type: str,
    entity_id: uuid.UUID,
    artwork_type: ArtworkType | None = None,
) -> list[Artwork]:
    stmt = (
        select(Artwork)
        .options(selectinload(Artwork.provider))
        .where(Artwork.entity_type == entity_type, Artwork.entity_id == entity_id)
        .order_by(Artwork.is_primary.desc(), Artwork.vote_average.desc().nulls_last())
    )
    if artwork_type is not None:
        stmt = stmt.where(Artwork.artwork_type == artwork_type)
    result = await session.execute(stmt)
    return list(result.scalars().all())


async def get_videos(session: AsyncSession, entity_type: str, entity_id: uuid.UUID) -> list[Video]:
    stmt = (
        select(Video)
        .options(selectinload(Video.provider))
        .where(Video.entity_type == entity_type, Video.entity_id == entity_id)
        .order_by(Video.is_official.desc(), Video.published_at.desc().nulls_last())
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())


async def get_credits(
    session: AsyncSession, entity_type: str, entity_id: uuid.UUID
) -> list[Credit]:
    stmt = (
        select(Credit)
        .options(selectinload(Credit.person))
        .where(Credit.entity_type == entity_type, Credit.entity_id == entity_id)
        .order_by(Credit.billing_order.asc().nulls_last())
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())


async def get_release_dates(
    session: AsyncSession, entity_type: str, entity_id: uuid.UUID
) -> list[ReleaseDate]:
    stmt = (
        select(ReleaseDate)
        .options(selectinload(ReleaseDate.certification))
        .where(ReleaseDate.entity_type == entity_type, ReleaseDate.entity_id == entity_id)
        .order_by(ReleaseDate.release_date.asc())
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())


async def sync_release_dates(
    session: AsyncSession,
    *,
    entity_type: str,
    entity_id: uuid.UUID,
    release_dates: Sequence[_NormalizedReleaseDateLike],
) -> None:
    """Replace an entity's release-date/certification rows wholesale, same
    delete-then-recreate approach as credits/videos (see module docstring)
    -- unlike Artwork, a ReleaseDate has no local-file state worth
    preserving across re-syncs."""

    await session.execute(
        delete(ReleaseDate).where(
            ReleaseDate.entity_type == entity_type, ReleaseDate.entity_id == entity_id
        )
    )
    for rd in release_dates:
        await refs.get_or_create_country(session, rd.country_code, rd.country_code)
        certification_id = None
        if rd.certification is not None:
            system, rating = rd.certification
            certification = await refs.get_or_create_certification(
                session, country_code=rd.country_code, system=system, rating=rating
            )
            certification_id = certification.id
        session.add(
            ReleaseDate(
                entity_type=entity_type,
                entity_id=entity_id,
                country_code=rd.country_code,
                release_date=rd.release_date,
                release_type=rd.release_type,
                certification_id=certification_id,
                note=rd.note,
            )
        )
    await session.flush()


async def get_watch_providers(
    session: AsyncSession, entity_type: str, entity_id: uuid.UUID
) -> list[WatchProviderOffering]:
    stmt = (
        select(WatchProviderOffering)
        .options(selectinload(WatchProviderOffering.watch_provider))
        .where(
            WatchProviderOffering.entity_type == entity_type,
            WatchProviderOffering.entity_id == entity_id,
        )
        .order_by(
            WatchProviderOffering.country_code.asc(),
            WatchProviderOffering.monetization_type.asc(),
            WatchProviderOffering.display_priority.asc().nulls_last(),
        )
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())


async def sync_watch_providers(
    session: AsyncSession,
    *,
    entity_type: str,
    entity_id: uuid.UUID,
    tmdb_provider_id: uuid.UUID,
    offerings: Sequence[_NormalizedWatchProviderOfferingLike],
) -> list[tuple[WatchProvider, str | None]]:
    """Replace an entity's watch-provider offerings wholesale, same
    delete-then-recreate as release_dates/credits/videos. Returns each
    distinct WatchProvider touched paired with its current logo_path so the
    caller can sync its logo artwork via company_logo_artwork + sync_artwork
    -- this repository layer doesn't import the services layer to build a
    NormalizedArtwork itself (see _NormalizedArtworkLike above).

    Resolves each distinct provider (by external_id) at most once, even
    when it appears in several countries/monetization types for the same
    entity (e.g. Netflix offering both a US and a GB flatrate row) --
    calling get_or_create_watch_provider a second time for the same
    external_id later in the same uncommitted session produced a second,
    duplicate WatchProvider row instead of finding the first (caught
    against real Postgres; watch providers was the first place in the
    ingestion pipeline where the same external entity is ever resolved
    twice within one ingest before anything commits)."""

    await session.execute(
        delete(WatchProviderOffering).where(
            WatchProviderOffering.entity_type == entity_type,
            WatchProviderOffering.entity_id == entity_id,
        )
    )
    touched: dict[uuid.UUID, tuple[WatchProvider, str | None]] = {}
    provider_by_external_id: dict[str, WatchProvider] = {}
    for offering in offerings:
        await refs.get_or_create_country(session, offering.country_code, offering.country_code)
        ext_id = offering.provider.external_id
        watch_provider = provider_by_external_id.get(ext_id)
        if watch_provider is None:
            watch_provider = await refs.get_or_create_watch_provider(
                session,
                provider_id=tmdb_provider_id,
                external_id=ext_id,
                name=offering.provider.name,
            )
            provider_by_external_id[ext_id] = watch_provider
        touched[watch_provider.id] = (watch_provider, offering.provider.logo_path)
        session.add(
            WatchProviderOffering(
                entity_type=entity_type,
                entity_id=entity_id,
                country_code=offering.country_code,
                monetization_type=offering.monetization_type,
                watch_provider_id=watch_provider.id,
                display_priority=offering.display_priority,
                link=offering.link,
            )
        )
    await session.flush()
    return list(touched.values())