Aurora-Core/app/services/common.py

"""Shared helpers for turning ORM rows into response schemas."""

from __future__ import annotations

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.ids import aurora_id
from app.models.artwork import Artwork
from app.models.credit import Credit
from app.models.enums import ArtworkType
from app.models.movie import MovieAlternateTitle, ReleaseDate
from app.models.tv import SeriesAlternateTitle
from app.models.video import Video
from app.models.watch_provider import WatchProviderOffering
from app.repositories import polymorphic
from app.schemas.common import (
    AlternateTitleOut,
    ArtworkOut,
    CastCreditOut,
    CertificationOut,
    CrewCreditOut,
    ImagesOut,
    PersonBriefOut,
    ReleaseDateOut,
    VideoOut,
    WatchProviderOut,
    WatchProviderRegionOut,
)


def artwork_url(artwork: Artwork) -> str:
    return artwork.local_path or artwork.original_url


def to_artwork_out(artwork: Artwork) -> ArtworkOut:
    return ArtworkOut(
        id=aurora_id("artwork", str(artwork.id)),
        type=artwork.artwork_type.value,
        language=artwork.language_code,
        width=artwork.width,
        height=artwork.height,
        aspect_ratio=artwork.aspect_ratio,
        url=artwork_url(artwork),
        provider=artwork.provider.slug if artwork.provider else None,
        is_primary=artwork.is_primary,
        vote_average=artwork.vote_average,
    )


def to_images_out(artwork_list: list[Artwork]) -> ImagesOut:
    images = ImagesOut()
    for artwork in artwork_list:
        out = to_artwork_out(artwork)
        if artwork.artwork_type == ArtworkType.POSTER:
            images.posters.append(out)
        elif artwork.artwork_type == ArtworkType.BACKDROP:
            images.backdrops.append(out)
        elif artwork.artwork_type == ArtworkType.LOGO:
            images.logos.append(out)
    return images


def primary_artwork_path(artwork_list: list[Artwork], artwork_type: ArtworkType) -> str | None:
    for artwork in artwork_list:
        if artwork.artwork_type == artwork_type:
            return artwork_url(artwork)
    return None


def to_video_out(video: Video) -> VideoOut:
    return VideoOut(
        id=aurora_id("video", str(video.id)),
        type=video.video_type.value,
        site=video.site,
        url=video.url,
        language=video.language_code,
        country=video.country_code,
        is_official=video.is_official,
        published_at=video.published_at,
        provider=video.provider.slug if video.provider else None,
    )


def to_alternate_title_out(alt: MovieAlternateTitle | SeriesAlternateTitle) -> AlternateTitleOut:
    return AlternateTitleOut(title=alt.title, country_code=alt.country_code, type=alt.type)


def to_release_date_out(release_date: ReleaseDate) -> ReleaseDateOut:
    certification = None
    if release_date.certification is not None:
        certification = CertificationOut(
            system=release_date.certification.system, rating=release_date.certification.rating
        )
    return ReleaseDateOut(
        country_code=release_date.country_code,
        release_date=release_date.release_date,
        release_type=release_date.release_type,
        certification=certification,
        note=release_date.note,
    )


def pick_certification(release_dates: list[ReleaseDate], *, preferred_country: str = "US") -> str | None:
    """The single rating a client wants when it isn't picking a country
    itself -- TMDb's own site defaults to the US rating badge, so this
    does too, falling back to the first country that has one at all."""

    fallback: str | None = None
    for rd in release_dates:
        if rd.certification is None:
            continue
        if rd.country_code == preferred_country:
            return rd.certification.rating
        if fallback is None:
            fallback = rd.certification.rating
    return fallback


async def build_watch_providers_out(
    session: AsyncSession, offerings: list[WatchProviderOffering]
) -> dict[str, WatchProviderRegionOut]:
    out: dict[str, WatchProviderRegionOut] = {}
    logo_cache: dict[object, str | None] = {}
    for row in offerings:
        region = out.get(row.country_code)
        if region is None:
            region = WatchProviderRegionOut(link=row.link)
            out[row.country_code] = region
        elif region.link is None and row.link is not None:
            region.link = row.link

        wp = row.watch_provider
        if wp.id not in logo_cache:
            logos = await polymorphic.get_artwork(session, "watch_provider", wp.id, ArtworkType.LOGO)
            logo_cache[wp.id] = primary_artwork_path(logos, ArtworkType.LOGO)

        entry = WatchProviderOut(
            id=aurora_id("watch_provider", str(wp.id)),
            name=wp.name,
            logo_path=logo_cache[wp.id],
            display_priority=row.display_priority,
        )
        getattr(region, row.monetization_type).append(entry)
    return out


def split_credits(
    credits_: list[Credit], profile_paths: dict | None = None
) -> tuple[list[CastCreditOut], list[CrewCreditOut]]:
    profile_paths = profile_paths or {}
    cast: list[CastCreditOut] = []
    crew: list[CrewCreditOut] = []
    for credit in credits_:
        person_brief = PersonBriefOut(
            id=aurora_id("person", str(credit.person_id)),
            name=credit.person.name,
            profile_path=profile_paths.get(credit.person_id),
        )
        if credit.department == "Acting":
            cast.append(
                CastCreditOut(
                    id=aurora_id("credit", str(credit.id)),
                    person=person_brief,
                    character=credit.character,
                    billing_order=credit.billing_order,
                    episode_count=credit.episode_count,
                )
            )
        else:
            crew.append(
                CrewCreditOut(
                    id=aurora_id("credit", str(credit.id)),
                    person=person_brief,
                    department=credit.department,
                    job=credit.job,
                    episode_count=credit.episode_count,
                )
            )
    return cast, crew