"""NFO service for creating and managing tvshow.nfo files. This service orchestrates TMDB API calls, XML generation, and media downloads to create complete NFO metadata for TV series. Example: >>> nfo_service = NFOService(tmdb_api_key="key", anime_directory="/anime") >>> await nfo_service.create_tvshow_nfo("Attack on Titan", "/anime/aot", 2013) """ import logging import re from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from lxml import etree from src.core.services.tmdb_client import TMDBAPIError, TMDBClient from src.core.utils.image_downloader import ImageDownloader from src.core.utils.nfo_generator import generate_tvshow_nfo from src.core.utils.nfo_mapper import tmdb_to_nfo_model logger = logging.getLogger(__name__) class NFOService: """Service for creating and managing tvshow.nfo files. Attributes: tmdb_client: TMDB API client image_downloader: Image downloader utility anime_directory: Base directory for anime series """ def __init__( self, tmdb_api_key: str, anime_directory: str, image_size: str = "original", auto_create: bool = True ): """Initialize NFO service. Args: tmdb_api_key: TMDB API key anime_directory: Base anime directory path image_size: Image size to download (original, w500, etc.) auto_create: Whether to auto-create NFOs """ self.tmdb_client = TMDBClient(api_key=tmdb_api_key) self.image_downloader = ImageDownloader() self.anime_directory = Path(anime_directory) self.image_size = image_size self.auto_create = auto_create def has_nfo(self, serie_folder: str) -> bool: """Check if tvshow.nfo exists for a series. Args: serie_folder: Series folder name Returns: True if NFO file exists """ nfo_path = self.anime_directory / serie_folder / "tvshow.nfo" return nfo_path.exists() @staticmethod def _extract_year_from_name(serie_name: str) -> Tuple[str, Optional[int]]: """Extract year from series name if present in format 'Name (YYYY)'. Args: serie_name: Series name, possibly with year in parentheses Returns: Tuple of (clean_name, year) - clean_name: Series name without year - year: Extracted year or None Examples: >>> _extract_year_from_name("Attack on Titan (2013)") ("Attack on Titan", 2013) >>> _extract_year_from_name("Attack on Titan") ("Attack on Titan", None) """ # Match year in parentheses at the end: (YYYY) match = re.search(r'\((\d{4})\)\s*$', serie_name) if match: year = int(match.group(1)) clean_name = serie_name[:match.start()].strip() return clean_name, year return serie_name, None async def check_nfo_exists(self, serie_folder: str) -> bool: """Check if tvshow.nfo exists for a series. Args: serie_folder: Series folder name Returns: True if tvshow.nfo exists """ nfo_path = self.anime_directory / serie_folder / "tvshow.nfo" return nfo_path.exists() async def create_tvshow_nfo( self, serie_name: str, serie_folder: str, year: Optional[int] = None, download_poster: bool = True, download_logo: bool = True, download_fanart: bool = True ) -> Path: """Create tvshow.nfo by scraping TMDB. Args: serie_name: Name of the series to search (may include year in parentheses) serie_folder: Series folder name year: Release year (helps narrow search). If None and name contains year, year will be auto-extracted download_poster: Whether to download poster.jpg download_logo: Whether to download logo.png download_fanart: Whether to download fanart.jpg Returns: Path to created NFO file Raises: TMDBAPIError: If TMDB API fails FileNotFoundError: If series folder doesn't exist """ # Extract year from name if not provided clean_name, extracted_year = self._extract_year_from_name(serie_name) if year is None and extracted_year is not None: year = extracted_year logger.info(f"Extracted year {year} from series name") # Use clean name for search search_name = clean_name logger.info(f"Creating NFO for {search_name} (year: {year})") folder_path = self.anime_directory / serie_folder if not folder_path.exists(): logger.info(f"Creating series folder: {folder_path}") folder_path.mkdir(parents=True, exist_ok=True) async with self.tmdb_client: # Search for TV show with clean name (without year) logger.debug(f"Searching TMDB for: {search_name}") search_results = await self.tmdb_client.search_tv_show(search_name) if not search_results.get("results"): raise TMDBAPIError(f"No results found for: {search_name}") # Find best match (consider year if provided) tv_show = self._find_best_match(search_results["results"], search_name, year) tv_id = tv_show["id"] logger.info(f"Found match: {tv_show['name']} (ID: {tv_id})") # Get detailed information with multi-language image support details = await self.tmdb_client.get_tv_show_details( tv_id, append_to_response="credits,external_ids,images" ) # Get content ratings for FSK content_ratings = await self.tmdb_client.get_tv_show_content_ratings(tv_id) # Enrich with fallback languages for empty overview/tagline # Pass search result overview as last resort fallback search_overview = tv_show.get("overview") or None details = await self._enrich_details_with_fallback( details, search_overview=search_overview ) # Convert TMDB data to TVShowNFO model nfo_model = tmdb_to_nfo_model( details, content_ratings, self.tmdb_client.get_image_url, self.image_size, ) # Generate XML nfo_xml = generate_tvshow_nfo(nfo_model) # Save NFO file nfo_path = folder_path / "tvshow.nfo" nfo_path.write_text(nfo_xml, encoding="utf-8") logger.info(f"Created NFO: {nfo_path}") # Download media files await self._download_media_files( details, folder_path, download_poster=download_poster, download_logo=download_logo, download_fanart=download_fanart ) return nfo_path async def update_tvshow_nfo( self, serie_folder: str, download_media: bool = True ) -> Path: """Update existing tvshow.nfo with fresh data from TMDB. Args: serie_folder: Series folder name download_media: Whether to re-download media files Returns: Path to updated NFO file Raises: FileNotFoundError: If NFO file doesn't exist TMDBAPIError: If TMDB API fails or no TMDB ID found in NFO """ folder_path = self.anime_directory / serie_folder nfo_path = folder_path / "tvshow.nfo" if not nfo_path.exists(): raise FileNotFoundError(f"NFO file not found: {nfo_path}") logger.info(f"Updating NFO for {serie_folder}") # Parse existing NFO to extract TMDB ID try: tree = etree.parse(str(nfo_path)) root = tree.getroot() # Try to find TMDB ID from uniqueid elements tmdb_id = None for uniqueid in root.findall(".//uniqueid"): if uniqueid.get("type") == "tmdb": tmdb_id = int(uniqueid.text) break # Fallback: check for tmdbid element if tmdb_id is None: tmdbid_elem = root.find(".//tmdbid") if tmdbid_elem is not None and tmdbid_elem.text: tmdb_id = int(tmdbid_elem.text) if tmdb_id is None: raise TMDBAPIError( f"No TMDB ID found in existing NFO. " f"Delete the NFO and create a new one instead." ) logger.debug(f"Found TMDB ID: {tmdb_id}") except etree.XMLSyntaxError as e: raise TMDBAPIError(f"Invalid XML in NFO file: {e}") except ValueError as e: raise TMDBAPIError(f"Invalid TMDB ID format in NFO: {e}") # Fetch fresh data from TMDB async with self.tmdb_client: logger.debug(f"Fetching fresh data for TMDB ID: {tmdb_id}") details = await self.tmdb_client.get_tv_show_details( tmdb_id, append_to_response="credits,external_ids,images" ) # Get content ratings for FSK content_ratings = await self.tmdb_client.get_tv_show_content_ratings(tmdb_id) # Enrich with fallback languages for empty overview/tagline details = await self._enrich_details_with_fallback(details) # Convert TMDB data to TVShowNFO model nfo_model = tmdb_to_nfo_model( details, content_ratings, self.tmdb_client.get_image_url, self.image_size, ) # Generate XML nfo_xml = generate_tvshow_nfo(nfo_model) # Save updated NFO file nfo_path.write_text(nfo_xml, encoding="utf-8") logger.info(f"Updated NFO: {nfo_path}") # Re-download media files if requested if download_media: await self._download_media_files( details, folder_path, download_poster=True, download_logo=True, download_fanart=True ) return nfo_path def parse_nfo_ids(self, nfo_path: Path) -> Dict[str, Optional[int]]: """Parse TMDB ID and TVDB ID from an existing NFO file. Args: nfo_path: Path to tvshow.nfo file Returns: Dictionary with 'tmdb_id' and 'tvdb_id' keys. Values are integers if found, None otherwise. Example: >>> ids = nfo_service.parse_nfo_ids(Path("/anime/series/tvshow.nfo")) >>> print(ids) {'tmdb_id': 1429, 'tvdb_id': 79168} """ result = {"tmdb_id": None, "tvdb_id": None} if not nfo_path.exists(): logger.debug(f"NFO file not found: {nfo_path}") return result try: tree = etree.parse(str(nfo_path)) root = tree.getroot() # Try to find TMDB ID from uniqueid elements first for uniqueid in root.findall(".//uniqueid"): uid_type = uniqueid.get("type") uid_text = uniqueid.text if uid_type == "tmdb" and uid_text: try: result["tmdb_id"] = int(uid_text) except ValueError: logger.warning( f"Invalid TMDB ID format in NFO: {uid_text}" ) elif uid_type == "tvdb" and uid_text: try: result["tvdb_id"] = int(uid_text) except ValueError: logger.warning( f"Invalid TVDB ID format in NFO: {uid_text}" ) # Fallback: check for dedicated tmdbid/tvdbid elements if result["tmdb_id"] is None: tmdbid_elem = root.find(".//tmdbid") if tmdbid_elem is not None and tmdbid_elem.text: try: result["tmdb_id"] = int(tmdbid_elem.text) except ValueError: logger.warning( f"Invalid TMDB ID format in tmdbid element: " f"{tmdbid_elem.text}" ) if result["tvdb_id"] is None: tvdbid_elem = root.find(".//tvdbid") if tvdbid_elem is not None and tvdbid_elem.text: try: result["tvdb_id"] = int(tvdbid_elem.text) except ValueError: logger.warning( f"Invalid TVDB ID format in tvdbid element: " f"{tvdbid_elem.text}" ) logger.debug( f"Parsed IDs from NFO: {nfo_path.name} - " f"TMDB: {result['tmdb_id']}, TVDB: {result['tvdb_id']}" ) except etree.XMLSyntaxError as e: logger.error(f"Invalid XML in NFO file {nfo_path}: {e}") except Exception as e: # pylint: disable=broad-except logger.error(f"Error parsing NFO file {nfo_path}: {e}") return result async def _enrich_details_with_fallback( self, details: Dict[str, Any], search_overview: Optional[str] = None, ) -> Dict[str, Any]: """Enrich TMDB details with fallback languages for empty fields. When requesting details in ``de-DE``, some anime have an empty ``overview`` (and potentially other translatable fields). This method detects empty values and fills them from alternative languages (``en-US``, then ``ja-JP``) so that NFO files always contain a ``plot`` regardless of whether the German translation exists. As a last resort, the overview from the search result is used. Args: details: TMDB TV show details (language ``de-DE``). search_overview: Overview text from the TMDB search result, used as a final fallback if all language-specific requests fail or return empty overviews. Returns: The *same* dict, mutated in-place with fallback values where needed. """ overview = details.get("overview") or "" if overview: # Overview already populated – nothing to do. return details tmdb_id = details.get("id") fallback_languages = ["en-US", "ja-JP"] for lang in fallback_languages: if details.get("overview"): break logger.debug( "Trying %s fallback for TMDB ID %s", lang, tmdb_id, ) try: lang_details = await self.tmdb_client.get_tv_show_details( tmdb_id, language=lang, ) if not details.get("overview") and lang_details.get("overview"): details["overview"] = lang_details["overview"] logger.info( "Used %s overview fallback for TMDB ID %s", lang, tmdb_id, ) # Also fill tagline if missing if not details.get("tagline") and lang_details.get("tagline"): details["tagline"] = lang_details["tagline"] except Exception as exc: # pylint: disable=broad-except logger.warning( "Failed to fetch %s fallback for TMDB ID %s: %s", lang, tmdb_id, exc, ) # Last resort: use search result overview if not details.get("overview") and search_overview: details["overview"] = search_overview logger.info( "Used search result overview fallback for TMDB ID %s", tmdb_id, ) return details def _find_best_match( self, results: List[Dict[str, Any]], query: str, year: Optional[int] = None ) -> Dict[str, Any]: """Find best matching TV show from search results. Args: results: TMDB search results query: Original search query year: Expected release year Returns: Best matching TV show data """ if not results: raise TMDBAPIError("No search results to match") # If year is provided, try to find exact match if year: for result in results: first_air_date = result.get("first_air_date", "") if first_air_date.startswith(str(year)): logger.debug(f"Found year match: {result['name']} ({first_air_date})") return result # Return first result (usually best match) return results[0] async def _download_media_files( self, tmdb_data: Dict[str, Any], folder_path: Path, download_poster: bool = True, download_logo: bool = True, download_fanart: bool = True ) -> Dict[str, bool]: """Download media files (poster, logo, fanart). Args: tmdb_data: TMDB TV show details folder_path: Series folder path download_poster: Download poster.jpg download_logo: Download logo.png download_fanart: Download fanart.jpg Returns: Dictionary with download status for each file """ poster_url = None logo_url = None fanart_url = None # Get poster URL if download_poster and tmdb_data.get("poster_path"): poster_url = self.tmdb_client.get_image_url( tmdb_data["poster_path"], self.image_size ) # Get fanart URL if download_fanart and tmdb_data.get("backdrop_path"): fanart_url = self.tmdb_client.get_image_url( tmdb_data["backdrop_path"], "original" # Always use original for fanart ) # Get logo URL if download_logo: images_data = tmdb_data.get("images", {}) logos = images_data.get("logos", []) if logos: logo_url = self.tmdb_client.get_image_url( logos[0]["file_path"], "original" # Logos should be original size ) # Download all media concurrently results = await self.image_downloader.download_all_media( folder_path, poster_url=poster_url, logo_url=logo_url, fanart_url=fanart_url, skip_existing=True ) logger.info(f"Media download results: {results}") return results async def close(self): """Clean up resources.""" await self.tmdb_client.close()