"""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.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 _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()