diff --git a/src/server/core_utils_temp/nfo_generator.py b/src/server/core_utils_temp/nfo_generator.py index 50d674d..bd76eed 100644 --- a/src/server/core_utils_temp/nfo_generator.py +++ b/src/server/core_utils_temp/nfo_generator.py @@ -4,7 +4,7 @@ This module provides functions to generate tvshow.nfo XML files from TVShowNFO Pydantic models, adapted from the scraper project. Example: - >>> from src.server.entities.nfo_models import TVShowNFO + >>> from src.server.nfo.nfo_models import TVShowNFO >>> nfo = TVShowNFO(title="Test Show", year=2020, tmdbid=12345) >>> xml_string = generate_tvshow_nfo(nfo) """ @@ -15,7 +15,7 @@ from typing import Optional from lxml import etree from src.config.settings import settings -from src.server.entities.nfo_models import TVShowNFO +from src.server.nfo.nfo_models import TVShowNFO logger = logging.getLogger(__name__) diff --git a/src/server/core_utils_temp/nfo_mapper.py b/src/server/core_utils_temp/nfo_mapper.py index 09c7cd5..7721777 100644 --- a/src/server/core_utils_temp/nfo_mapper.py +++ b/src/server/core_utils_temp/nfo_mapper.py @@ -11,7 +11,7 @@ import logging from datetime import datetime from typing import Any, Callable, Dict, List, Optional -from src.server.entities.nfo_models import ( +from src.server.nfo.nfo_models import ( ActorInfo, ImageInfo, NamedSeason, diff --git a/src/server/nfo/__init__.py b/src/server/nfo/__init__.py new file mode 100644 index 0000000..fd2147f --- /dev/null +++ b/src/server/nfo/__init__.py @@ -0,0 +1,9 @@ +"""NFO package - TV show metadata generation for Kodi/XBMC. + +Re-exports the public API for the nfo package. +""" + +from src.server.nfo.nfo_models import TVShowNFO +from src.server.nfo.tmdb_client import TMDBClient, TMDBAPIError +from src.server.nfo.nfo_generator import generate_tvshow_nfo +from src.server.nfo.nfo_mapper import tmdb_to_nfo_model \ No newline at end of file diff --git a/src/server/nfo/nfo_generator.py b/src/server/nfo/nfo_generator.py new file mode 100644 index 0000000..bd76eed --- /dev/null +++ b/src/server/nfo/nfo_generator.py @@ -0,0 +1,213 @@ +"""NFO XML generator for Kodi/XBMC format. + +This module provides functions to generate tvshow.nfo XML files from +TVShowNFO Pydantic models, adapted from the scraper project. + +Example: + >>> from src.server.nfo.nfo_models import TVShowNFO + >>> nfo = TVShowNFO(title="Test Show", year=2020, tmdbid=12345) + >>> xml_string = generate_tvshow_nfo(nfo) +""" + +import logging +from typing import Optional + +from lxml import etree + +from src.config.settings import settings +from src.server.nfo.nfo_models import TVShowNFO + +logger = logging.getLogger(__name__) + + +def generate_tvshow_nfo(tvshow: TVShowNFO, pretty_print: bool = True) -> str: + """Generate tvshow.nfo XML content from TVShowNFO model. + + Args: + tvshow: TVShowNFO Pydantic model with metadata + pretty_print: Whether to format XML with indentation + + Returns: + XML string in Kodi/XBMC tvshow.nfo format + + Example: + >>> nfo = TVShowNFO(title="Attack on Titan", year=2013) + >>> xml = generate_tvshow_nfo(nfo) + """ + root = etree.Element("tvshow") + + # Basic information + _add_element(root, "title", tvshow.title) + _add_element(root, "originaltitle", tvshow.originaltitle) + _add_element(root, "showtitle", tvshow.showtitle) + _add_element(root, "sorttitle", tvshow.sorttitle) + _add_element(root, "year", str(tvshow.year) if tvshow.year else None) + + # Plot and description – always write even when empty so that + # all NFO files have a consistent set of tags regardless of whether they + # were produced by create or update. + _add_element(root, "plot", tvshow.plot, always_write=True) + _add_element(root, "outline", tvshow.outline) + _add_element(root, "tagline", tvshow.tagline) + + # Technical details + _add_element(root, "runtime", str(tvshow.runtime) if tvshow.runtime else None) + + # Content rating - prefer FSK if available and configured + if getattr(settings, 'nfo_prefer_fsk_rating', True) and tvshow.fsk: + _add_element(root, "mpaa", tvshow.fsk) + else: + _add_element(root, "mpaa", tvshow.mpaa) + + _add_element(root, "certification", tvshow.certification) + + # Status and dates + _add_element(root, "premiered", tvshow.premiered) + _add_element(root, "status", tvshow.status) + _add_element(root, "dateadded", tvshow.dateadded) + + # Ratings + if tvshow.ratings: + ratings_elem = etree.SubElement(root, "ratings") + for rating in tvshow.ratings: + rating_elem = etree.SubElement(ratings_elem, "rating") + if rating.name: + rating_elem.set("name", rating.name) + if rating.max_rating: + rating_elem.set("max", str(rating.max_rating)) + if rating.default: + rating_elem.set("default", "true") + + _add_element(rating_elem, "value", str(rating.value)) + if rating.votes is not None: + _add_element(rating_elem, "votes", str(rating.votes)) + + _add_element(root, "userrating", str(tvshow.userrating) if tvshow.userrating is not None else None) + + # IDs + _add_element(root, "tmdbid", str(tvshow.tmdbid) if tvshow.tmdbid else None) + _add_element(root, "imdbid", tvshow.imdbid) + _add_element(root, "tvdbid", str(tvshow.tvdbid) if tvshow.tvdbid else None) + + # Legacy ID fields for compatibility + _add_element(root, "id", str(tvshow.tvdbid) if tvshow.tvdbid else None) + _add_element(root, "imdb_id", tvshow.imdbid) + + # Unique IDs + for uid in tvshow.uniqueid: + uid_elem = etree.SubElement(root, "uniqueid") + uid_elem.set("type", uid.type) + if uid.default: + uid_elem.set("default", "true") + uid_elem.text = uid.value + + # Multi-value fields + for genre in tvshow.genre: + _add_element(root, "genre", genre) + + for studio in tvshow.studio: + _add_element(root, "studio", studio) + + for country in tvshow.country: + _add_element(root, "country", country) + + for tag in tvshow.tag: + _add_element(root, "tag", tag) + + # Thumbnails (posters, logos) + for thumb in tvshow.thumb: + thumb_elem = etree.SubElement(root, "thumb") + if thumb.aspect: + thumb_elem.set("aspect", thumb.aspect) + if thumb.season is not None: + thumb_elem.set("season", str(thumb.season)) + if thumb.type: + thumb_elem.set("type", thumb.type) + thumb_elem.text = str(thumb.url) + + # Fanart + if tvshow.fanart: + fanart_elem = etree.SubElement(root, "fanart") + for fanart in tvshow.fanart: + fanart_thumb = etree.SubElement(fanart_elem, "thumb") + fanart_thumb.text = str(fanart.url) + + # Named seasons + for named_season in tvshow.namedseason: + season_elem = etree.SubElement(root, "namedseason") + season_elem.set("number", str(named_season.number)) + season_elem.text = named_season.name + + # Actors + for actor in tvshow.actors: + actor_elem = etree.SubElement(root, "actor") + _add_element(actor_elem, "name", actor.name) + _add_element(actor_elem, "role", actor.role) + _add_element(actor_elem, "thumb", str(actor.thumb) if actor.thumb else None) + _add_element(actor_elem, "profile", str(actor.profile) if actor.profile else None) + _add_element(actor_elem, "tmdbid", str(actor.tmdbid) if actor.tmdbid else None) + + # Additional fields + _add_element(root, "trailer", str(tvshow.trailer) if tvshow.trailer else None) + _add_element(root, "watched", "true" if tvshow.watched else "false") + if tvshow.playcount is not None: + _add_element(root, "playcount", str(tvshow.playcount)) + + # Generate XML string + xml_str = etree.tostring( + root, + pretty_print=pretty_print, + encoding="unicode", + xml_declaration=False + ) + + # Add XML declaration + xml_declaration = '\n' + return xml_declaration + xml_str + + +def _add_element( + parent: etree.Element, + tag: str, + text: Optional[str], + always_write: bool = False, +) -> Optional[etree.Element]: + """Add a child element to parent if text is not None or empty. + + Args: + parent: Parent XML element + tag: Tag name for child element + text: Text content (None or empty strings are skipped + unless *always_write* is True) + always_write: When True the element is created even when + *text* is None/empty (the element will have + no text content). Useful for tags like + ```` that should always be present. + + Returns: + Created element or None if skipped + """ + if text is not None and text != "": + elem = etree.SubElement(parent, tag) + elem.text = text + return elem + if always_write: + return etree.SubElement(parent, tag) + return None + + +def validate_nfo_xml(xml_string: str) -> bool: + """Validate NFO XML structure. + + Args: + xml_string: XML content to validate + + Returns: + True if valid XML, False otherwise + """ + try: + etree.fromstring(xml_string.encode('utf-8')) + return True + except etree.XMLSyntaxError as e: + logger.error("Invalid NFO XML: %s", e) + return False diff --git a/src/server/nfo/nfo_mapper.py b/src/server/nfo/nfo_mapper.py new file mode 100644 index 0000000..7721777 --- /dev/null +++ b/src/server/nfo/nfo_mapper.py @@ -0,0 +1,234 @@ +"""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.server.nfo.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"), + ) diff --git a/src/server/nfo/nfo_models.py b/src/server/nfo/nfo_models.py new file mode 100644 index 0000000..25f1333 --- /dev/null +++ b/src/server/nfo/nfo_models.py @@ -0,0 +1,335 @@ +"""Pydantic models for NFO metadata based on Kodi/XBMC standard. + +This module provides data models for tvshow.nfo files that are compatible +with media center applications like Kodi, Plex, and Jellyfin. + +Example: + >>> nfo = TVShowNFO( + ... title="Attack on Titan", + ... year=2013, + ... tmdbid=1429 + ... ) + >>> nfo.premiered = "2013-04-07" +""" + +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, Field, HttpUrl, field_validator + + +class RatingInfo(BaseModel): + """Rating information from various sources. + + Attributes: + name: Source of the rating (e.g., 'themoviedb', 'imdb') + value: Rating value (typically 0-10) + votes: Number of votes + max_rating: Maximum possible rating (default: 10) + default: Whether this is the default rating to display + """ + + name: str = Field(..., description="Rating source name") + value: float = Field(..., ge=0, description="Rating value") + votes: Optional[int] = Field(None, ge=0, description="Number of votes") + max_rating: int = Field(10, ge=1, description="Maximum rating value") + default: bool = Field(False, description="Is this the default rating") + + @field_validator('value') + @classmethod + def validate_value(cls, v: float, info) -> float: + """Ensure rating value doesn't exceed max_rating.""" + # Note: max_rating is not available yet during validation, + # so we use a reasonable default check + if v > 10: + raise ValueError("Rating value cannot exceed 10") + return v + + +class ActorInfo(BaseModel): + """Actor/cast member information. + + Attributes: + name: Actor's name + role: Character name/role + thumb: URL to actor's photo + profile: URL to actor's profile page + tmdbid: TMDB ID for the actor + """ + + name: str = Field(..., description="Actor's name") + role: Optional[str] = Field(None, description="Character role") + thumb: Optional[HttpUrl] = Field(None, description="Actor photo URL") + profile: Optional[HttpUrl] = Field(None, description="Actor profile URL") + tmdbid: Optional[int] = Field(None, description="TMDB actor ID") + + +class ImageInfo(BaseModel): + """Image information for posters, fanart, and logos. + + Attributes: + url: URL to the image + aspect: Image aspect/type (e.g., 'poster', 'clearlogo', 'logo') + season: Season number for season-specific images + type: Image type (e.g., 'season') + """ + + url: HttpUrl = Field(..., description="Image URL") + aspect: Optional[str] = Field( + None, + description="Image aspect (poster, clearlogo, logo)" + ) + season: Optional[int] = Field(None, ge=-1, description="Season number") + type: Optional[str] = Field(None, description="Image type") + + +class NamedSeason(BaseModel): + """Named season information. + + Attributes: + number: Season number + name: Season name/title + """ + + number: int = Field(..., ge=0, description="Season number") + name: str = Field(..., description="Season name") + + +class UniqueID(BaseModel): + """Unique identifier from various sources. + + Attributes: + type: ID source type (tmdb, imdb, tvdb) + value: The ID value + default: Whether this is the default ID + """ + + type: str = Field(..., description="ID type (tmdb, imdb, tvdb)") + value: str = Field(..., description="ID value") + default: bool = Field(False, description="Is default ID") + + +class TVShowNFO(BaseModel): + """Main tvshow.nfo structure following Kodi/XBMC standard. + + This model represents the complete metadata for a TV show that can be + serialized to XML for use with media center applications. + + Attributes: + title: Main title of the show + originaltitle: Original title (e.g., in original language) + showtitle: Show title (often same as title) + sorttitle: Title used for sorting + year: Release year + plot: Full plot description + outline: Short plot summary + tagline: Show tagline/slogan + runtime: Episode runtime in minutes + mpaa: Content rating (e.g., TV-14, TV-MA) + certification: Additional certification info + premiered: Premiere date (YYYY-MM-DD format) + status: Show status (e.g., 'Continuing', 'Ended') + studio: List of production studios + genre: List of genres + country: List of countries + tag: List of tags/keywords + ratings: List of ratings from various sources + userrating: User's personal rating + watched: Whether the show has been watched + playcount: Number of times watched + tmdbid: TMDB ID + imdbid: IMDB ID + tvdbid: TVDB ID + uniqueid: List of unique IDs + thumb: List of thumbnail/poster images + fanart: List of fanart/backdrop images + actors: List of cast members + namedseason: List of named seasons + trailer: Trailer URL + dateadded: Date when added to library + """ + + # Required fields + title: str = Field(..., description="Show title", min_length=1) + + # Basic information (optional) + originaltitle: Optional[str] = Field(None, description="Original title") + showtitle: Optional[str] = Field(None, description="Show title") + sorttitle: Optional[str] = Field(None, description="Sort title") + year: Optional[int] = Field( + None, + ge=1900, + le=2100, + description="Release year" + ) + + # Plot and description + plot: Optional[str] = Field(None, description="Full plot description") + outline: Optional[str] = Field(None, description="Short plot summary") + tagline: Optional[str] = Field(None, description="Show tagline") + + # Technical details + runtime: Optional[int] = Field( + None, + ge=0, + description="Episode runtime in minutes" + ) + mpaa: Optional[str] = Field(None, description="Content rating") + fsk: Optional[str] = Field( + None, + description="German FSK rating (e.g., 'FSK 12', 'FSK 16')" + ) + certification: Optional[str] = Field( + None, + description="Certification info" + ) + + # Status and dates + premiered: Optional[str] = Field( + None, + description="Premiere date (YYYY-MM-DD)" + ) + status: Optional[str] = Field(None, description="Show status") + dateadded: Optional[str] = Field( + None, + description="Date added to library" + ) + + # Multi-value fields + studio: List[str] = Field( + default_factory=list, + description="Production studios" + ) + genre: List[str] = Field( + default_factory=list, + description="Genres" + ) + country: List[str] = Field( + default_factory=list, + description="Countries" + ) + tag: List[str] = Field( + default_factory=list, + description="Tags/keywords" + ) + + # IDs + tmdbid: Optional[int] = Field(None, description="TMDB ID") + imdbid: Optional[str] = Field(None, description="IMDB ID") + tvdbid: Optional[int] = Field(None, description="TVDB ID") + uniqueid: List[UniqueID] = Field( + default_factory=list, + description="Unique IDs" + ) + + # Ratings and viewing info + ratings: List[RatingInfo] = Field( + default_factory=list, + description="Ratings" + ) + userrating: Optional[float] = Field( + None, + ge=0, + le=10, + description="User rating" + ) + watched: bool = Field(False, description="Watched status") + playcount: Optional[int] = Field( + None, + ge=0, + description="Play count" + ) + + # Media + thumb: List[ImageInfo] = Field( + default_factory=list, + description="Thumbnail images" + ) + fanart: List[ImageInfo] = Field( + default_factory=list, + description="Fanart images" + ) + + # Cast and crew + actors: List[ActorInfo] = Field( + default_factory=list, + description="Cast members" + ) + + # Seasons + namedseason: List[NamedSeason] = Field( + default_factory=list, + description="Named seasons" + ) + + # Additional + trailer: Optional[HttpUrl] = Field(None, description="Trailer URL") + + @field_validator('premiered') + @classmethod + def validate_premiered_date(cls, v: Optional[str]) -> Optional[str]: + """Validate premiered date format (YYYY-MM-DD).""" + if v is None: + return v + + # Check format strictly: YYYY-MM-DD + if len(v) != 10 or v[4] != '-' or v[7] != '-': + raise ValueError( + "Premiered date must be in YYYY-MM-DD format" + ) + + try: + datetime.strptime(v, '%Y-%m-%d') + except ValueError as exc: + raise ValueError( + "Premiered date must be in YYYY-MM-DD format" + ) from exc + + return v + + @field_validator('dateadded') + @classmethod + def validate_dateadded(cls, v: Optional[str]) -> Optional[str]: + """Validate dateadded format (YYYY-MM-DD HH:MM:SS).""" + if v is None: + return v + + # Check format strictly: YYYY-MM-DD HH:MM:SS + if len(v) != 19 or v[4] != '-' or v[7] != '-' or v[10] != ' ' or v[13] != ':' or v[16] != ':': + raise ValueError( + "Dateadded must be in YYYY-MM-DD HH:MM:SS format" + ) + + try: + datetime.strptime(v, '%Y-%m-%d %H:%M:%S') + except ValueError as exc: + raise ValueError( + "Dateadded must be in YYYY-MM-DD HH:MM:SS format" + ) from exc + + return v + + @field_validator('imdbid') + @classmethod + def validate_imdbid(cls, v: Optional[str]) -> Optional[str]: + """Validate IMDB ID format (should start with 'tt').""" + if v is None: + return v + + if not v.startswith('tt'): + raise ValueError("IMDB ID must start with 'tt'") + + if not v[2:].isdigit(): + raise ValueError("IMDB ID must be 'tt' followed by digits") + + return v + + def model_post_init(self, __context) -> None: + """Set default values after initialization.""" + # Set showtitle to title if not provided + if self.showtitle is None: + self.showtitle = self.title + + # Set originaltitle to title if not provided + if self.originaltitle is None: + self.originaltitle = self.title diff --git a/src/server/nfo/tmdb_client.py b/src/server/nfo/tmdb_client.py new file mode 100644 index 0000000..aef3e61 --- /dev/null +++ b/src/server/nfo/tmdb_client.py @@ -0,0 +1,424 @@ +"""TMDB API client for fetching TV show metadata. + +This module provides an async client for The Movie Database (TMDB) API, +adapted from the scraper project to fit the AniworldMain architecture. + +Example: + >>> async with TMDBClient(api_key="your_key") as client: + ... results = await client.search_tv_show("Attack on Titan") + ... show_id = results["results"][0]["id"] + ... details = await client.get_tv_show_details(show_id) +""" + +import asyncio +import logging +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +import aiohttp + +logger = logging.getLogger(__name__) + + +class TMDBAPIError(Exception): + """Exception raised for TMDB API errors.""" + pass + + +class TMDBClient: + """Async TMDB API client for TV show metadata. + + Attributes: + api_key: TMDB API key for authentication + base_url: Base URL for TMDB API + image_base_url: Base URL for TMDB images + max_connections: Maximum concurrent connections + session: aiohttp ClientSession for requests + """ + + DEFAULT_BASE_URL = "https://api.themoviedb.org/3" + DEFAULT_IMAGE_BASE_URL = "https://image.tmdb.org/t/p" + NEGATIVE_CACHE_TTL = 86400 # 24 hours + + def __init__( + self, + api_key: str, + base_url: str = DEFAULT_BASE_URL, + image_base_url: str = DEFAULT_IMAGE_BASE_URL, + max_connections: int = 10 + ): + """Initialize TMDB client. + + Args: + api_key: TMDB API key + base_url: TMDB API base URL + image_base_url: TMDB image base URL + max_connections: Maximum concurrent connections + """ + if not api_key: + raise ValueError("TMDB API key is required") + + self.api_key = api_key + self.base_url = base_url.rstrip('/') + self.image_base_url = image_base_url.rstrip('/') + self.max_connections = max_connections + self.session: Optional[aiohttp.ClientSession] = None + self._cache: Dict[str, Any] = {} + self._negative_cache: Dict[str, float] = {} # query -> timestamp when cached + # TMDB allows ~40 req/s; use 30 concurrent + per-second throttle to stay safe + self._semaphore = asyncio.Semaphore(30) + self._rate_limit_lock = asyncio.Lock() + self._request_timestamps: List[float] = [] + self._max_requests_per_second = 35 # Stay under TMDB's ~40/s limit + + async def __aenter__(self): + """Async context manager entry.""" + await self._ensure_session() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """Async context manager exit.""" + await self.close() + + async def _ensure_session(self): + """Ensure aiohttp session is created.""" + if self.session is None or self.session.closed: + connector = aiohttp.TCPConnector(limit=self.max_connections) + self.session = aiohttp.ClientSession(connector=connector) + + async def _request( + self, + endpoint: str, + params: Optional[Dict[str, Any]] = None, + max_retries: int = 5 + ) -> Dict[str, Any]: + """Make an async request to TMDB API with retries. + + Args: + endpoint: API endpoint (e.g., 'search/tv') + params: Query parameters + max_retries: Maximum retry attempts + + Returns: + API response as dictionary + + Raises: + TMDBAPIError: If request fails after retries + """ + await self._ensure_session() + + url = f"{self.base_url}/{endpoint}" + params = params or {} + params["api_key"] = self.api_key + + # Cache key for deduplication + cache_key = f"{endpoint}:{str(sorted(params.items()))}" + if cache_key in self._cache: + logger.debug("Cache hit for %s", endpoint) + return self._cache[cache_key] + + # Check negative cache (cached empty results) + negative_cache_key = f"{endpoint}:{str(sorted(params.items()))}" + if negative_cache_key in self._negative_cache: + if time.monotonic() - self._negative_cache[negative_cache_key] < self.NEGATIVE_CACHE_TTL: + logger.debug("Negative cache hit for %s (cached empty result)", endpoint) + return {"results": []} + else: + # Expired negative cache entry + del self._negative_cache[negative_cache_key] + + delay = 1 + last_error = None + + # Rate limiting: ensure we don't exceed ~35 requests/second + async with self._rate_limit_lock: + now = time.monotonic() + # Remove timestamps older than 1 second + self._request_timestamps = [ + ts for ts in self._request_timestamps if now - ts < 1.0 + ] + if len(self._request_timestamps) >= self._max_requests_per_second: + sleep_time = 1.0 - (now - self._request_timestamps[0]) + if sleep_time > 0: + logger.debug("Rate throttling: waiting %.2fs", sleep_time) + await asyncio.sleep(sleep_time) + self._request_timestamps.append(time.monotonic()) + + async with self._semaphore: + for attempt in range(max_retries): + try: + # Re-ensure session before each attempt in case it was closed + await self._ensure_session() + + if self.session is None: + raise TMDBAPIError("Session is not available") + + logger.debug("TMDB API request: %s (attempt %s)", endpoint, attempt + 1) + async with self.session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=60)) as resp: + if resp.status == 401: + raise TMDBAPIError("Invalid TMDB API key") + elif resp.status == 404: + raise TMDBAPIError(f"Resource not found: {endpoint}") + elif resp.status == 429: + # Rate limit - wait longer with exponential backoff + retry_after = int(resp.headers.get('Retry-After', max(delay * 2, 2))) + logger.warning("Rate limited, waiting %ss", retry_after) + await asyncio.sleep(retry_after) + continue + + resp.raise_for_status() + data = await resp.json() + self._cache[cache_key] = data + # Cache negative result if empty + if endpoint.startswith("search/") and not data.get("results"): + self._negative_cache[negative_cache_key] = time.monotonic() + logger.debug("Cached negative result for %s", endpoint) + return data + + except asyncio.TimeoutError as e: + last_error = e + if attempt < max_retries - 1: + logger.warning("Request timeout (attempt %s), retrying in %ss", attempt + 1, delay) + await asyncio.sleep(delay) + delay *= 2 + else: + logger.error("Request timed out after %s attempts", max_retries) + + except (aiohttp.ClientError, AttributeError) as e: + last_error = e + # If connector/session was closed, try to recreate it + if "Connector is closed" in str(e) or isinstance(e, AttributeError): + logger.warning( + "Session issue detected, recreating session: %s", + e, + exc_info=True, + ) + self.session = None + await self._ensure_session() + + # DNS / host-unreachable errors are not transient — abort immediately + error_str = str(e) + if "name resolution" in error_str.lower() or ( + isinstance(e, aiohttp.ClientConnectorError) and + "Cannot connect to host" in error_str + ): + logger.error("Non-transient connection error, aborting retries: %s", e) + raise TMDBAPIError(f"Request failed after {attempt + 1} attempts: {e}") from e + + if attempt < max_retries - 1: + logger.warning("Request failed (attempt %s): %s, retrying in %ss", attempt + 1, e, delay) + await asyncio.sleep(delay) + delay *= 2 + else: + logger.error("Request failed after %s attempts: %s", max_retries, e) + + raise TMDBAPIError(f"Request failed after {max_retries} attempts: {last_error}") + + async def search_tv_show( + self, + query: str, + language: str = "de-DE", + page: int = 1 + ) -> Dict[str, Any]: + """Search for TV shows by name. + + Args: + query: Search query (show name) + language: Language for results (default: German) + page: Page number for pagination + + Returns: + Search results with list of shows + + Example: + >>> results = await client.search_tv_show("Attack on Titan") + >>> shows = results["results"] + """ + return await self._request( + "search/tv", + {"query": query, "language": language, "page": page} + ) + + async def search_multi( + self, + query: str, + language: str = "en-US", + page: int = 1 + ) -> Dict[str, Any]: + """Search for movies and TV shows by name using TMDB multi search. + + Multi search returns both movies and TV shows, useful for anime + that might be indexed as movies on TMDB. + + Args: + query: Search query (show name) + language: Language for results (default: English) + page: Page number for pagination + + Returns: + Search results with list of movies and TV shows + + Example: + >>> results = await client.search_multi("Suzume no Tojimari") + >>> shows = [r for r in results["results"] if r["media_type"] == "tv"] + """ + return await self._request( + "search/multi", + {"query": query, "language": language, "page": page} + ) + + async def get_tv_show_details( + self, + tv_id: int, + language: str = "de-DE", + append_to_response: Optional[str] = None + ) -> Dict[str, Any]: + """Get detailed information about a TV show. + + Args: + tv_id: TMDB TV show ID + language: Language for metadata + append_to_response: Additional data to include (e.g., "credits,images") + + Returns: + TV show details including metadata, cast, etc. + """ + params = {"language": language} + if append_to_response: + params["append_to_response"] = append_to_response + + return await self._request(f"tv/{tv_id}", params) + + async def get_tv_show_content_ratings(self, tv_id: int) -> Dict[str, Any]: + """Get content ratings for a TV show. + + Args: + tv_id: TMDB TV show ID + + Returns: + Content ratings by country + """ + return await self._request(f"tv/{tv_id}/content_ratings") + + async def get_tv_show_external_ids(self, tv_id: int) -> Dict[str, Any]: + """Get external IDs (IMDB, TVDB) for a TV show. + + Args: + tv_id: TMDB TV show ID + + Returns: + Dictionary with external IDs (imdb_id, tvdb_id, etc.) + """ + return await self._request(f"tv/{tv_id}/external_ids") + + async def get_tv_show_images( + self, + tv_id: int, + language: Optional[str] = None + ) -> Dict[str, Any]: + """Get images (posters, backdrops, logos) for a TV show. + + Args: + tv_id: TMDB TV show ID + language: Language filter for images (None = all languages) + + Returns: + Dictionary with poster, backdrop, and logo lists + """ + params = {} + if language: + params["language"] = language + + return await self._request(f"tv/{tv_id}/images", params) + + async def download_image( + self, + image_path: str, + local_path: Path, + size: str = "original" + ) -> None: + """Download an image from TMDB. + + Args: + image_path: Image path from TMDB API (e.g., "/abc123.jpg") + local_path: Local file path to save image + size: Image size (w500, original, etc.) + + Raises: + TMDBAPIError: If download fails + """ + await self._ensure_session() + + url = f"{self.image_base_url}/{size}{image_path}" + + try: + logger.debug("Downloading image from %s", url) + async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=60)) as resp: + resp.raise_for_status() + + # Ensure parent directory exists + local_path.parent.mkdir(parents=True, exist_ok=True) + + # Write image data + with open(local_path, "wb") as f: + f.write(await resp.read()) + + logger.info("Downloaded image to %s", local_path) + + except aiohttp.ClientError as e: + raise TMDBAPIError(f"Failed to download image: {e}") + + def get_image_url(self, image_path: str, size: str = "original") -> str: + """Get full URL for an image. + + Args: + image_path: Image path from TMDB API + size: Image size (w500, original, etc.) + + Returns: + Full image URL + """ + return f"{self.image_base_url}/{size}{image_path}" + + async def close(self): + """Close the aiohttp session and clean up resources.""" + if self.session and not self.session.closed: + await self.session.close() + self.session = None + logger.debug("TMDB client session closed") + + def __del__(self): + """Warn if session is unclosed during garbage collection.""" + if self.session is not None and not self.session.closed: + logger.warning( + "TMDBClient: unclosed session detected. " + "Use 'async with TMDBClient(...)' or call close() explicitly." + ) + + def clear_cache(self): + """Clear the request cache.""" + self._cache.clear() + logger.debug("TMDB client cache cleared") + + def clear_negative_cache(self): + """Clear the negative result cache.""" + self._negative_cache.clear() + logger.debug("TMDB negative cache cleared") + + def cleanup_expired_negative_cache(self) -> int: + """Remove expired entries from negative cache. + + Returns: + Number of entries removed + """ + now = time.monotonic() + expired_keys = [ + key for key, timestamp in self._negative_cache.items() + if now - timestamp >= self.NEGATIVE_CACHE_TTL + ] + for key in expired_keys: + del self._negative_cache[key] + if expired_keys: + logger.debug("Removed %d expired negative cache entries", len(expired_keys)) + return len(expired_keys) diff --git a/tests/api/test_nfo_diagnostics_repair.py b/tests/api/test_nfo_diagnostics_repair.py index 453c17b..3b4350d 100644 --- a/tests/api/test_nfo_diagnostics_repair.py +++ b/tests/api/test_nfo_diagnostics_repair.py @@ -289,7 +289,7 @@ class TestNfoRepair: self, authenticated_client, override_dependencies ): """Test repair handles TMDB API failure gracefully.""" - from src.server.services_nfo_temp.tmdb_client import TMDBAPIError + from src.server.nfo.tmdb_client import TMDBAPIError with patch("src.server.api.nfo.Path") as MockPath: mock_path = Mock() diff --git a/tests/integration/test_tmdb_resilience.py b/tests/integration/test_tmdb_resilience.py index d68056b..823e0de 100644 --- a/tests/integration/test_tmdb_resilience.py +++ b/tests/integration/test_tmdb_resilience.py @@ -9,7 +9,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import aiohttp import pytest -from src.server.services_nfo_temp.tmdb_client import TMDBAPIError, TMDBClient +from src.server.nfo.tmdb_client import TMDBAPIError, TMDBClient def _make_ctx(response): diff --git a/tests/performance/test_nfo_batch_performance.py b/tests/performance/test_nfo_batch_performance.py index 753f62d..4c9ef40 100644 --- a/tests/performance/test_nfo_batch_performance.py +++ b/tests/performance/test_nfo_batch_performance.py @@ -297,7 +297,7 @@ class TestTMDBAPIBatchingOptimization: # Simulate rate limit on 5th call if call_count == 5: - from src.server.services_nfo_temp.tmdb_client import TMDBAPIError + from src.server.nfo.tmdb_client import TMDBAPIError raise TMDBAPIError("Rate limit exceeded") await asyncio.sleep(0.01) diff --git a/tests/unit/test_tmdb_client.py b/tests/unit/test_tmdb_client.py index fe78189..0ae7134 100644 --- a/tests/unit/test_tmdb_client.py +++ b/tests/unit/test_tmdb_client.py @@ -6,7 +6,7 @@ import aiohttp import pytest from aiohttp import ClientResponseError, ClientSession -from src.server.services_nfo_temp.tmdb_client import TMDBAPIError, TMDBClient +from src.server.nfo.tmdb_client import TMDBAPIError, TMDBClient @pytest.fixture diff --git a/tests/unit/test_tmdb_rate_limiting.py b/tests/unit/test_tmdb_rate_limiting.py index c9723f7..8fd49a6 100644 --- a/tests/unit/test_tmdb_rate_limiting.py +++ b/tests/unit/test_tmdb_rate_limiting.py @@ -9,7 +9,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import aiohttp import pytest -from src.server.services_nfo_temp.tmdb_client import TMDBAPIError, TMDBClient +from src.server.nfo.tmdb_client import TMDBAPIError, TMDBClient def _make_ctx(response):