"""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 NormalizedAlternateTitle: title: str country_code: str | None type: str | None = None def _alternate_titles(entries: list[dict]) -> list[NormalizedAlternateTitle]: return [ NormalizedAlternateTitle( title=t["title"], country_code=t.get("iso_3166_1") or None, type=t.get("type") or None ) for t in entries if t.get("title") ] # TMDb's per-country rating body isn't returned alongside a release_dates # certification or a content_ratings rating -- only the country and the # rating string itself are. This maps the handful of markets Aurora can # name with confidence; everywhere else falls back to the country code # itself (see NormalizedReleaseDate.certification) rather than fabricating # a body name TMDb never told us. _CERTIFICATION_SYSTEMS = { "US": "MPA", "GB": "BBFC", "DE": "FSK", "FR": "CNC", "JP": "Eirin", "AU": "ACB", "CA": "CHVRS", "KR": "KMRB", "BR": "ANCINE", "ES": "ICAA", "IT": "MiC", "NZ": "OFLC", "IE": "IFCO", "IN": "CBFC", } # TMDb's release_dates[].type: 1=Premiere, 2=Theatrical (limited), # 3=Theatrical, 4=Digital, 5=Physical, 6=TV -- collapsed onto the five # release_type values ReleaseDate.release_type documents. _RELEASE_TYPE_MAP = {1: "premiere", 2: "theatrical", 3: "theatrical", 4: "digital", 5: "physical", 6: "tv"} @dataclass class NormalizedReleaseDate: country_code: str release_date: date release_type: str | None certification: tuple[str, str] | None = None # (system, rating) note: str | None = None def _movie_release_dates(raw: dict) -> list[NormalizedReleaseDate]: out: list[NormalizedReleaseDate] = [] for country_block in (raw.get("release_dates") or {}).get("results", []): country_code = country_block.get("iso_3166_1") if not country_code: continue for entry in country_block.get("release_dates", []): release_date = _parse_date((entry.get("release_date") or "")[:10] or None) if release_date is None: continue rating = entry.get("certification") certification = ( (_CERTIFICATION_SYSTEMS.get(country_code, country_code), rating) if rating else None ) out.append( NormalizedReleaseDate( country_code=country_code, release_date=release_date, release_type=_RELEASE_TYPE_MAP.get(entry.get("type")), certification=certification, note=entry.get("note") or None, ) ) return out def _series_release_dates(raw: dict, first_air_date: date | None) -> list[NormalizedReleaseDate]: # Unlike /movie's release_dates, TMDb's /tv content_ratings has no date # at all -- just a country and a rating. ReleaseDate.release_date is # required, so each rating is anchored to the show's own first_air_date # (release_type="tv") rather than left dateless; a show with no # first_air_date yet has nothing to anchor to and is skipped. if first_air_date is None: return [] out: list[NormalizedReleaseDate] = [] for entry in (raw.get("content_ratings") or {}).get("results", []): country_code = entry.get("iso_3166_1") rating = entry.get("rating") if not country_code or not rating: continue out.append( NormalizedReleaseDate( country_code=country_code, release_date=first_air_date, release_type="tv", certification=(_CERTIFICATION_SYSTEMS.get(country_code, country_code), rating), ) ) return out @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[NormalizedAlternateTitle] = 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) release_dates: list[NormalizedReleaseDate] = 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=_alternate_titles(alt_titles), 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], release_dates=_movie_release_dates(raw), ) @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[NormalizedAlternateTitle] = 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) release_dates: list[NormalizedReleaseDate] = 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 ] first_air_date = _parse_date(raw.get("first_air_date")) 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=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=_alternate_titles(alt_titles), networks=networks, companies=companies, creators=creators, cast=cast, crew=crew, artwork=artwork, videos=normalized_videos, external_ids=external_ids, seasons=seasons, release_dates=_series_release_dates(raw, first_air_date), ) @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, )