"""TMDB to NFO model mapper. This module converts TMDB API data to TVShowNFO Pydantic models, keeping the mapping logic separate from the service orchestration. Example: >>> model = tmdb_to_nfo_model(tmdb_data, content_ratings, get_image_url, "original") """ import logging from datetime import datetime from typing import Any, Callable, Dict, List, Optional from src.core.entities.nfo_models import ( ActorInfo, ImageInfo, NamedSeason, RatingInfo, TVShowNFO, UniqueID, ) logger = logging.getLogger(__name__) def _extract_rating_by_country( content_ratings: Dict[str, Any], country_code: str, ) -> Optional[str]: """Extract content rating for a specific country from TMDB content ratings. Args: content_ratings: TMDB content ratings response dict with "results" list. country_code: ISO 3166-1 alpha-2 country code (e.g., "DE", "US"). Returns: Raw rating string for the requested country, or None if not found. Example: >>> _extract_rating_by_country({"results": [{"iso_3166_1": "US", "rating": "TV-14"}]}, "US") 'TV-14' """ if not content_ratings or "results" not in content_ratings: return None for rating in content_ratings["results"]: if rating.get("iso_3166_1") == country_code: return rating.get("rating") or None return None def _extract_fsk_rating(content_ratings: Dict[str, Any]) -> Optional[str]: """Extract German FSK rating from TMDB content ratings. Delegates to :func:`_extract_rating_by_country` and then normalises the raw TMDB string into the 'FSK XX' format expected by Kodi/Jellyfin. Args: content_ratings: TMDB content ratings response. Returns: Formatted FSK string (e.g., 'FSK 12') or None. """ raw = _extract_rating_by_country(content_ratings, "DE") if raw is None: return None fsk_mapping: Dict[str, str] = { "0": "FSK 0", "6": "FSK 6", "12": "FSK 12", "16": "FSK 16", "18": "FSK 18", } if raw in fsk_mapping: return fsk_mapping[raw] # Try to extract numeric part (ordered high→low to avoid partial matches) for key in ["18", "16", "12", "6", "0"]: if key in raw: return fsk_mapping[key] if raw.startswith("FSK"): return raw logger.debug("Unmapped German rating: %s", raw) return None def tmdb_to_nfo_model( tmdb_data: Dict[str, Any], content_ratings: Optional[Dict[str, Any]], get_image_url: Callable[[str, str], str], image_size: str = "original", ) -> TVShowNFO: """Convert TMDB API data to a fully-populated TVShowNFO model. All required NFO tags are explicitly set in this function so that newly created files are complete without a subsequent repair pass. Args: tmdb_data: TMDB TV show details (with credits, external_ids, images appended via ``append_to_response``). content_ratings: TMDB content ratings response, or None. get_image_url: Callable ``(path, size) -> url`` for TMDB images. image_size: TMDB image size parameter (e.g., ``"original"``, ``"w500"``). Returns: TVShowNFO Pydantic model with all available fields populated. """ title: str = tmdb_data["name"] original_title: str = tmdb_data.get("original_name") or title # --- Year and dates --- first_air_date: Optional[str] = tmdb_data.get("first_air_date") or None year: Optional[int] = int(first_air_date[:4]) if first_air_date else None # --- Ratings --- ratings: List[RatingInfo] = [] if tmdb_data.get("vote_average"): ratings.append(RatingInfo( name="themoviedb", value=float(tmdb_data["vote_average"]), votes=tmdb_data.get("vote_count", 0), max_rating=10, default=True, )) # --- External IDs --- external_ids: Dict[str, Any] = tmdb_data.get("external_ids", {}) imdb_id: Optional[str] = external_ids.get("imdb_id") tvdb_id: Optional[int] = external_ids.get("tvdb_id") # --- Images --- thumb_images: List[ImageInfo] = [] fanart_images: List[ImageInfo] = [] if tmdb_data.get("poster_path"): thumb_images.append(ImageInfo( url=get_image_url(tmdb_data["poster_path"], image_size), aspect="poster", )) if tmdb_data.get("backdrop_path"): fanart_images.append(ImageInfo( url=get_image_url(tmdb_data["backdrop_path"], image_size), )) logos: List[Dict[str, Any]] = tmdb_data.get("images", {}).get("logos", []) if logos: thumb_images.append(ImageInfo( url=get_image_url(logos[0]["file_path"], image_size), aspect="clearlogo", )) # --- Cast (top 10) --- actors: List[ActorInfo] = [] for member in tmdb_data.get("credits", {}).get("cast", [])[:10]: actor_thumb: Optional[str] = None if member.get("profile_path"): actor_thumb = get_image_url(member["profile_path"], "h632") actors.append(ActorInfo( name=member["name"], role=member.get("character"), thumb=actor_thumb, tmdbid=member["id"], )) # --- Named seasons --- named_seasons: List[NamedSeason] = [] for season_info in tmdb_data.get("seasons", []): season_name = season_info.get("name") season_number = season_info.get("season_number") if season_name and season_number is not None: named_seasons.append(NamedSeason( number=season_number, name=season_name, )) # --- Unique IDs --- unique_ids: List[UniqueID] = [] if tmdb_data.get("id"): unique_ids.append(UniqueID(type="tmdb", value=str(tmdb_data["id"]), default=False)) if imdb_id: unique_ids.append(UniqueID(type="imdb", value=imdb_id, default=False)) if tvdb_id: unique_ids.append(UniqueID(type="tvdb", value=str(tvdb_id), default=True)) # --- Content ratings --- fsk_rating: Optional[str] = _extract_fsk_rating(content_ratings) if content_ratings else None mpaa_rating: Optional[str] = ( _extract_rating_by_country(content_ratings, "US") if content_ratings else None ) # --- Country: prefer origin_country codes; fall back to production_countries names --- country_list: List[str] = list(tmdb_data.get("origin_country", [])) if not country_list: country_list = [c["name"] for c in tmdb_data.get("production_countries", [])] # --- Runtime --- runtime_list: List[int] = tmdb_data.get("episode_run_time", []) runtime: Optional[int] = runtime_list[0] if runtime_list else None return TVShowNFO( title=title, originaltitle=original_title, showtitle=title, sorttitle=title, year=year, plot=tmdb_data.get("overview") or None, outline=tmdb_data.get("overview") or None, tagline=tmdb_data.get("tagline") or None, runtime=runtime, premiered=first_air_date, status=tmdb_data.get("status"), genre=[g["name"] for g in tmdb_data.get("genres", [])], studio=[n["name"] for n in tmdb_data.get("networks", [])], country=country_list, ratings=ratings, fsk=fsk_rating, mpaa=mpaa_rating, tmdbid=tmdb_data.get("id"), imdbid=imdb_id, tvdbid=tvdb_id, uniqueid=unique_ids, thumb=thumb_images, fanart=fanart_images, actors=actors, namedseason=named_seasons, watched=False, dateadded=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), )