Aurora-Core/app/services/metadata/normalizer.py

"""Normalizes raw TMDb payloads into provider-agnostic dataclasses.

This is the seam between "provider shape" and "Aurora shape" (section 13).
A future provider adapter only needs its own `normalize_movie`-equivalent
that produces these same dataclasses; nothing downstream (matcher,
ingestion) needs to know which provider the data came from.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date

TMDB_IMAGE_BASE = "https://image.tmdb.org/t/p/original"


def _parse_date(value: str | None) -> date | None:
    if not value:
        return None
    try:
        return date.fromisoformat(value)
    except ValueError:
        return None


def _image_url(path: str | None) -> str | None:
    if not path:
        return None
    return f"{TMDB_IMAGE_BASE}{path}"


def _image_url_required(path: str) -> str:
    """Same as `_image_url`, for call sites already guarded by a truthy
    check on `path` — keeps `NormalizedArtwork.url: str` honest."""

    return f"{TMDB_IMAGE_BASE}{path}"


@dataclass
class NormalizedPersonRef:
    external_id: str
    name: str
    profile_path: str | None = None
    department: str | None = None
    job: str | None = None
    character: str | None = None
    billing_order: int | None = None
    gender: str | None = None


@dataclass
class NormalizedCompanyRef:
    external_id: str
    name: str
    country_code: str | None = None
    role: str = "production"
    logo_path: str | None = None  # TMDb's raw path, e.g. "/logo.png" — not a full URL yet


@dataclass
class NormalizedArtwork:
    artwork_type: str  # "poster" | "backdrop" | "logo" | "profile" | "still"
    url: str
    provider_asset_id: str | None
    width: int | None
    height: int | None
    aspect_ratio: float | None
    language: str | None
    vote_average: float | None = None
    # True for the single image the provider designates as *the* image for
    # this entity (TMDb's top-level poster_path/backdrop_path/profile_path),
    # as opposed to the full alternates list — this is what gets
    # auto-downloaded for local hosting (app/services/artwork/), so it's
    # deliberately narrow rather than "the first one in the list".
    is_primary: bool = False


def company_logo_artwork(logo_path: str | None) -> NormalizedArtwork | None:
    """A single-item artwork entry for a company/network's TMDb logo_path,
    or None if it doesn't have one. Shared by movie_ingestion.py and
    series_ingestion.py so the "how do we turn a relative TMDb path into a
    real image URL" logic stays in one place (this module), rather than the
    ingestion layer needing to know about TMDb's CDN itself."""

    if not logo_path:
        return None
    return NormalizedArtwork(
        artwork_type="logo",
        url=_image_url_required(logo_path),
        provider_asset_id=logo_path,
        width=None,
        height=None,
        aspect_ratio=None,
        language=None,
        is_primary=True,  # a company has exactly one logo ingested today
    )


@dataclass
class NormalizedVideo:
    video_type: str
    provider_video_id: str
    site: str | None
    url: str
    language: str | None
    country: str | None
    is_official: bool
    published_at: date | None


@dataclass
class NormalizedMovie:
    external_id: str
    title: str
    original_title: str | None
    overview: str | None
    tagline: str | None
    status: str
    original_language: str | None
    release_date: date | None
    runtime_minutes: int | None
    budget: int | None
    revenue: int | None
    popularity: float | None
    vote_average: float | None
    vote_count: int | None
    adult: bool
    genres: list[str] = field(default_factory=list)
    keywords: list[str] = field(default_factory=list)
    alternate_titles: list[str] = field(default_factory=list)
    companies: list[NormalizedCompanyRef] = field(default_factory=list)
    cast: list[NormalizedPersonRef] = field(default_factory=list)
    crew: list[NormalizedPersonRef] = field(default_factory=list)
    artwork: list[NormalizedArtwork] = field(default_factory=list)
    videos: list[NormalizedVideo] = field(default_factory=list)
    external_ids: dict[str, str] = field(default_factory=dict)
    collection: tuple[str, str] | None = None  # (external_id, name)
    recommendation_external_ids: list[str] = field(default_factory=list)


@dataclass
class NormalizedCollection:
    external_id: str
    name: str
    overview: str | None
    artwork: list[NormalizedArtwork] = field(default_factory=list)
    # TMDb's "parts", ordered by release date rather than TMDb's own array
    # order (which isn't guaranteed to be chronological) -- see
    # normalize_collection.
    part_external_ids: list[str] = field(default_factory=list)


def normalize_collection(raw: dict) -> NormalizedCollection:
    artwork: list[NormalizedArtwork] = []
    if raw.get("poster_path"):
        artwork.append(
            NormalizedArtwork(
                "poster",
                _image_url_required(raw["poster_path"]),
                raw["poster_path"],
                None,
                None,
                None,
                None,
                is_primary=True,
            )
        )
    if raw.get("backdrop_path"):
        artwork.append(
            NormalizedArtwork(
                "backdrop",
                _image_url_required(raw["backdrop_path"]),
                raw["backdrop_path"],
                None,
                None,
                None,
                None,
                is_primary=True,
            )
        )

    parts = sorted(
        raw.get("parts") or [],
        key=lambda p: _parse_date(p.get("release_date")) or date.max,
    )

    return NormalizedCollection(
        external_id=str(raw["id"]),
        name=raw["name"],
        overview=raw.get("overview") or None,
        artwork=artwork,
        part_external_ids=[str(p["id"]) for p in parts],
    )


_STATUS_MAP = {
    "Rumored": "rumored",
    "Planned": "planned",
    "In Production": "in_production",
    "Post Production": "post_production",
    "Released": "released",
    "Canceled": "canceled",
}


def normalize_movie(raw: dict) -> NormalizedMovie:
    credits_ = raw.get("credits") or {}
    images = raw.get("images") or {}
    videos = (raw.get("videos") or {}).get("results") or []
    keywords = (raw.get("keywords") or {}).get("keywords") or []
    alt_titles = (raw.get("alternative_titles") or {}).get("titles") or []
    ext_ids = raw.get("external_ids") or {}
    collection = raw.get("belongs_to_collection")
    recommendations = (raw.get("recommendations") or {}).get("results") or []

    external_ids: dict[str, str] = {}
    if raw.get("imdb_id"):
        external_ids["imdb"] = raw["imdb_id"]
    elif ext_ids.get("imdb_id"):
        external_ids["imdb"] = ext_ids["imdb_id"]
    if ext_ids.get("wikidata_id"):
        external_ids["wikidata"] = ext_ids["wikidata_id"]

    cast = [
        NormalizedPersonRef(
            external_id=str(c["id"]),
            name=c["name"],
            profile_path=_image_url(c.get("profile_path")),
            department="Acting",
            character=c.get("character"),
            billing_order=c.get("order"),
            gender=_gender_label(c.get("gender")),
        )
        for c in credits_.get("cast", [])
    ]
    crew = [
        NormalizedPersonRef(
            external_id=str(c["id"]),
            name=c["name"],
            profile_path=_image_url(c.get("profile_path")),
            department=c.get("department") or "Crew",
            job=c.get("job"),
            gender=_gender_label(c.get("gender")),
        )
        for c in credits_.get("crew", [])
    ]

    companies = [
        NormalizedCompanyRef(
            external_id=str(pc["id"]),
            name=pc["name"],
            country_code=pc.get("origin_country") or None,
            logo_path=pc.get("logo_path"),
        )
        for pc in raw.get("production_companies", [])
    ]

    artwork: list[NormalizedArtwork] = []
    if raw.get("poster_path"):
        artwork.append(
            NormalizedArtwork(
                "poster",
                _image_url_required(raw["poster_path"]),
                raw["poster_path"],
                None,
                None,
                None,
                None,
                is_primary=True,
            )
        )
    if raw.get("backdrop_path"):
        artwork.append(
            NormalizedArtwork(
                "backdrop",
                _image_url_required(raw["backdrop_path"]),
                raw["backdrop_path"],
                None,
                None,
                None,
                None,
                is_primary=True,
            )
        )
    for kind, tmdb_key in (("poster", "posters"), ("backdrop", "backdrops"), ("logo", "logos")):
        for img in images.get(tmdb_key, []):
            artwork.append(
                NormalizedArtwork(
                    kind,
                    _image_url_required(img["file_path"]),
                    img["file_path"],
                    img.get("width"),
                    img.get("height"),
                    img.get("aspect_ratio"),
                    img.get("iso_639_1"),
                    img.get("vote_average"),
                )
            )

    normalized_videos = [
        NormalizedVideo(
            video_type=(v.get("type") or "other").lower().replace(" ", "_"),
            provider_video_id=v["id"],
            site=v.get("site"),
            url=_video_url(v),
            language=v.get("iso_639_1"),
            country=v.get("iso_3166_1"),
            is_official=bool(v.get("official")),
            published_at=_parse_date(v.get("published_at", "")[:10] if v.get("published_at") else None),
        )
        for v in videos
    ]

    return NormalizedMovie(
        external_id=str(raw["id"]),
        title=raw.get("title") or raw.get("original_title") or "",
        original_title=raw.get("original_title"),
        overview=raw.get("overview") or None,
        tagline=raw.get("tagline") or None,
        status=_STATUS_MAP.get(raw.get("status", "Released"), "released"),
        original_language=raw.get("original_language"),
        release_date=_parse_date(raw.get("release_date")),
        runtime_minutes=raw.get("runtime"),
        budget=raw.get("budget") or None,
        revenue=raw.get("revenue") or None,
        popularity=raw.get("popularity"),
        vote_average=raw.get("vote_average"),
        vote_count=raw.get("vote_count"),
        adult=bool(raw.get("adult", False)),
        genres=[g["name"] for g in raw.get("genres", [])],
        keywords=[k["name"] for k in keywords],
        alternate_titles=[t["title"] for t in alt_titles if t.get("title")],
        companies=companies,
        cast=cast,
        crew=crew,
        artwork=artwork,
        videos=normalized_videos,
        external_ids=external_ids,
        collection=(str(collection["id"]), collection["name"]) if collection else None,
        recommendation_external_ids=[str(m["id"]) for m in recommendations],
    )


@dataclass
class NormalizedPerson:
    external_id: str
    name: str
    biography: str | None
    birthday: date | None
    deathday: date | None
    birthplace: str | None
    gender: str | None
    known_for_department: str | None
    popularity: float | None
    aliases: list[str]
    artwork: list[NormalizedArtwork]
    external_ids: dict[str, str]


def _gender_label(tmdb_gender: int | None) -> str | None:
    return {1: "female", 2: "male", 3: "non_binary"}.get(tmdb_gender or 0)


def _video_url(video: dict) -> str:
    site = (video.get("site") or "").lower()
    key = video.get("key", "")
    if site == "youtube":
        return f"https://www.youtube.com/watch?v={key}"
    if site == "vimeo":
        return f"https://vimeo.com/{key}"
    return key


def normalize_person(raw: dict) -> NormalizedPerson:
    images = (raw.get("images") or {}).get("profiles") or []
    ext_ids = raw.get("external_ids") or {}
    external_ids: dict[str, str] = {}
    if raw.get("imdb_id"):
        external_ids["imdb"] = raw["imdb_id"]
    elif ext_ids.get("imdb_id"):
        external_ids["imdb"] = ext_ids["imdb_id"]
    if ext_ids.get("wikidata_id"):
        external_ids["wikidata"] = ext_ids["wikidata_id"]

    artwork = [
        NormalizedArtwork(
            "profile",
            _image_url_required(img["file_path"]),
            img["file_path"],
            img.get("width"),
            img.get("height"),
            img.get("aspect_ratio"),
            img.get("iso_639_1"),
            img.get("vote_average"),
        )
        for img in images
    ]
    if raw.get("profile_path"):
        existing = next(
            (a for a in artwork if a.provider_asset_id == raw["profile_path"]), None
        )
        if existing is not None:
            existing.is_primary = True
        else:
            artwork.insert(
                0,
                NormalizedArtwork(
                    "profile",
                    _image_url_required(raw["profile_path"]),
                    raw["profile_path"],
                    None,
                    None,
                    None,
                    None,
                    is_primary=True,
                ),
            )

    also_known_as = raw.get("also_known_as") or []

    return NormalizedPerson(
        external_id=str(raw["id"]),
        name=raw.get("name", ""),
        biography=raw.get("biography") or None,
        birthday=_parse_date(raw.get("birthday")),
        deathday=_parse_date(raw.get("deathday")),
        birthplace=raw.get("place_of_birth") or None,
        gender=_gender_label(raw.get("gender")),
        known_for_department=raw.get("known_for_department"),
        popularity=raw.get("popularity"),
        aliases=list(also_known_as),
        artwork=artwork,
        external_ids=external_ids,
    )


_SERIES_STATUS_MAP = {
    "Planned": "planned",
    "Pilot": "pilot",
    "In Production": "in_production",
    "Returning Series": "returning_series",
    "Ended": "ended",
    "Canceled": "canceled",
}


@dataclass
class NormalizedSeasonStub:
    """A season summary as it appears embedded in a series payload — enough
    to create/update the Season row, but without episodes (those require a
    separate per-season fetch — see `normalize_season`)."""

    season_number: int
    title: str | None
    overview: str | None
    air_date: date | None
    episode_count: int | None
    poster: NormalizedArtwork | None


@dataclass
class NormalizedSeries:
    external_id: str
    title: str
    original_title: str | None
    overview: str | None
    tagline: str | None
    status: str
    original_language: str | None
    first_air_date: date | None
    last_air_date: date | None
    number_of_seasons: int | None
    number_of_episodes: int | None
    popularity: float | None
    vote_average: float | None
    vote_count: int | None
    genres: list[str] = field(default_factory=list)
    keywords: list[str] = field(default_factory=list)
    alternate_titles: list[str] = field(default_factory=list)
    networks: list[NormalizedCompanyRef] = field(default_factory=list)
    companies: list[NormalizedCompanyRef] = field(default_factory=list)
    creators: list[NormalizedPersonRef] = field(default_factory=list)
    cast: list[NormalizedPersonRef] = field(default_factory=list)
    crew: list[NormalizedPersonRef] = field(default_factory=list)
    artwork: list[NormalizedArtwork] = field(default_factory=list)
    videos: list[NormalizedVideo] = field(default_factory=list)
    external_ids: dict[str, str] = field(default_factory=dict)
    seasons: list[NormalizedSeasonStub] = field(default_factory=list)


def normalize_series(raw: dict) -> NormalizedSeries:
    credits_ = raw.get("credits") or {}
    images = raw.get("images") or {}
    videos = (raw.get("videos") or {}).get("results") or []
    # Unlike /movie, TMDb's /tv keywords and alternative_titles nest under
    # "results", not "keywords"/"titles".
    keywords = (raw.get("keywords") or {}).get("results") or []
    alt_titles = (raw.get("alternative_titles") or {}).get("results") or []
    ext_ids = raw.get("external_ids") or {}

    external_ids: dict[str, str] = {}
    if ext_ids.get("imdb_id"):
        external_ids["imdb"] = ext_ids["imdb_id"]
    if ext_ids.get("tvdb_id"):
        external_ids["tvdb"] = str(ext_ids["tvdb_id"])
    if ext_ids.get("wikidata_id"):
        external_ids["wikidata"] = ext_ids["wikidata_id"]

    cast = [
        NormalizedPersonRef(
            external_id=str(c["id"]),
            name=c["name"],
            profile_path=_image_url(c.get("profile_path")),
            department="Acting",
            character=c.get("character"),
            billing_order=c.get("order"),
            gender=_gender_label(c.get("gender")),
        )
        for c in credits_.get("cast", [])
    ]
    crew = [
        NormalizedPersonRef(
            external_id=str(c["id"]),
            name=c["name"],
            profile_path=_image_url(c.get("profile_path")),
            department=c.get("department") or "Crew",
            job=c.get("job"),
            gender=_gender_label(c.get("gender")),
        )
        for c in credits_.get("crew", [])
    ]
    creators = [
        NormalizedPersonRef(
            external_id=str(c["id"]),
            name=c["name"],
            profile_path=_image_url(c.get("profile_path")),
            department="Creator",
            gender=_gender_label(c.get("gender")),
        )
        for c in raw.get("created_by", [])
    ]

    networks = [
        NormalizedCompanyRef(
            external_id=str(n["id"]),
            name=n["name"],
            country_code=n.get("origin_country") or None,
            logo_path=n.get("logo_path"),
        )
        for n in raw.get("networks", [])
    ]
    companies = [
        NormalizedCompanyRef(
            external_id=str(pc["id"]),
            name=pc["name"],
            country_code=pc.get("origin_country") or None,
            logo_path=pc.get("logo_path"),
        )
        for pc in raw.get("production_companies", [])
    ]

    artwork: list[NormalizedArtwork] = []
    if raw.get("poster_path"):
        artwork.append(
            NormalizedArtwork(
                "poster",
                _image_url_required(raw["poster_path"]),
                raw["poster_path"],
                None,
                None,
                None,
                None,
                is_primary=True,
            )
        )
    if raw.get("backdrop_path"):
        artwork.append(
            NormalizedArtwork(
                "backdrop",
                _image_url_required(raw["backdrop_path"]),
                raw["backdrop_path"],
                None,
                None,
                None,
                None,
                is_primary=True,
            )
        )
    for kind, tmdb_key in (("poster", "posters"), ("backdrop", "backdrops"), ("logo", "logos")):
        for img in images.get(tmdb_key, []):
            artwork.append(
                NormalizedArtwork(
                    kind,
                    _image_url_required(img["file_path"]),
                    img["file_path"],
                    img.get("width"),
                    img.get("height"),
                    img.get("aspect_ratio"),
                    img.get("iso_639_1"),
                    img.get("vote_average"),
                )
            )

    normalized_videos = [
        NormalizedVideo(
            video_type=(v.get("type") or "other").lower().replace(" ", "_"),
            provider_video_id=v["id"],
            site=v.get("site"),
            url=_video_url(v),
            language=v.get("iso_639_1"),
            country=v.get("iso_3166_1"),
            is_official=bool(v.get("official")),
            published_at=_parse_date(v.get("published_at", "")[:10] if v.get("published_at") else None),
        )
        for v in videos
    ]

    seasons = [
        NormalizedSeasonStub(
            season_number=s["season_number"],
            title=s.get("name"),
            overview=s.get("overview") or None,
            air_date=_parse_date(s.get("air_date")),
            episode_count=s.get("episode_count"),
            poster=(
                NormalizedArtwork(
                    "poster", _image_url_required(s["poster_path"]), s["poster_path"],
                    None, None, None, None, is_primary=True,
                )
                if s.get("poster_path")
                else None
            ),
        )
        for s in raw.get("seasons", [])
        # TMDb includes a "season 0" for specials; keep it (it's a real,
        # commonly-referenced season) but drop entries missing a number.
        if s.get("season_number") is not None
    ]

    return NormalizedSeries(
        external_id=str(raw["id"]),
        title=raw.get("name") or raw.get("original_name") or "",
        original_title=raw.get("original_name"),
        overview=raw.get("overview") or None,
        tagline=raw.get("tagline") or None,
        status=_SERIES_STATUS_MAP.get(raw.get("status", "Returning Series"), "returning_series"),
        original_language=raw.get("original_language"),
        first_air_date=_parse_date(raw.get("first_air_date")),
        last_air_date=_parse_date(raw.get("last_air_date")),
        number_of_seasons=raw.get("number_of_seasons"),
        number_of_episodes=raw.get("number_of_episodes"),
        popularity=raw.get("popularity"),
        vote_average=raw.get("vote_average"),
        vote_count=raw.get("vote_count"),
        genres=[g["name"] for g in raw.get("genres", [])],
        keywords=[k["name"] for k in keywords],
        alternate_titles=[t["title"] for t in alt_titles if t.get("title")],
        networks=networks,
        companies=companies,
        creators=creators,
        cast=cast,
        crew=crew,
        artwork=artwork,
        videos=normalized_videos,
        external_ids=external_ids,
        seasons=seasons,
    )


@dataclass
class NormalizedEpisode:
    external_id: str
    episode_number: int
    title: str | None
    overview: str | None
    air_date: date | None
    runtime_minutes: int | None
    cast: list[NormalizedPersonRef] = field(default_factory=list)
    guest_stars: list[NormalizedPersonRef] = field(default_factory=list)
    crew: list[NormalizedPersonRef] = field(default_factory=list)
    stills: list[NormalizedArtwork] = field(default_factory=list)


@dataclass
class NormalizedSeason:
    season_number: int
    title: str | None
    overview: str | None
    air_date: date | None
    episode_count: int | None
    artwork: list[NormalizedArtwork] = field(default_factory=list)
    episodes: list[NormalizedEpisode] = field(default_factory=list)
    external_ids: dict[str, str] = field(default_factory=dict)


def _episode_person_refs(entries: list[dict]) -> list[NormalizedPersonRef]:
    return [
        NormalizedPersonRef(
            external_id=str(e["id"]),
            name=e["name"],
            profile_path=_image_url(e.get("profile_path")),
            department="Acting",
            character=e.get("character"),
            billing_order=e.get("order"),
            gender=_gender_label(e.get("gender")),
        )
        for e in entries
    ]


def normalize_season(raw: dict) -> NormalizedSeason:
    images = raw.get("images") or {}
    ext_ids = raw.get("external_ids") or {}
    external_ids: dict[str, str] = {}
    if ext_ids.get("tvdb_id"):
        external_ids["tvdb"] = str(ext_ids["tvdb_id"])

    artwork: list[NormalizedArtwork] = []
    if raw.get("poster_path"):
        artwork.append(
            NormalizedArtwork(
                "poster", _image_url_required(raw["poster_path"]), raw["poster_path"],
                None, None, None, None, is_primary=True,
            )
        )
    for img in images.get("posters", []):
        artwork.append(
            NormalizedArtwork(
                "poster",
                _image_url_required(img["file_path"]),
                img["file_path"],
                img.get("width"),
                img.get("height"),
                img.get("aspect_ratio"),
                img.get("iso_639_1"),
                img.get("vote_average"),
            )
        )

    episodes = []
    for ep in raw.get("episodes", []):
        crew = [
            NormalizedPersonRef(
                external_id=str(c["id"]),
                name=c["name"],
                profile_path=_image_url(c.get("profile_path")),
                department=c.get("department") or "Crew",
                job=c.get("job"),
                gender=_gender_label(c.get("gender")),
            )
            for c in ep.get("crew", [])
        ]
        stills: list[NormalizedArtwork] = []
        if ep.get("still_path"):
            stills.append(
                NormalizedArtwork(
                    "still", _image_url_required(ep["still_path"]), ep["still_path"],
                    None, None, None, None, is_primary=True,
                )
            )
        episodes.append(
            NormalizedEpisode(
                external_id=str(ep["id"]),
                episode_number=ep["episode_number"],
                title=ep.get("name"),
                overview=ep.get("overview") or None,
                air_date=_parse_date(ep.get("air_date")),
                runtime_minutes=ep.get("runtime"),
                # TMDb's per-episode payload has no separate "main cast"
                # list distinct from guest_stars/crew — the series' regular
                # cast is assumed, not repeated per episode.
                cast=[],
                guest_stars=_episode_person_refs(ep.get("guest_stars", [])),
                crew=crew,
                stills=stills,
            )
        )

    return NormalizedSeason(
        season_number=raw["season_number"],
        title=raw.get("name"),
        overview=raw.get("overview") or None,
        air_date=_parse_date(raw.get("air_date")),
        episode_count=len(episodes) or None,
        artwork=artwork,
        episodes=episodes,
        external_ids=external_ids,
    )