refactor: simplify NFO handling, remove legacy services
- Drop nfo_factory, nfo_repair_service, nfo_service, series_manager_service - Delete key_resolution_service, consolidate into folder_rename_service - Remove bulk of NFO-related tests (coverage via integration tests) - Streamline SeriesApp, background_loader, initialization services - Add folder_rename_service to scheduler Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -23,8 +23,6 @@ from src.core.entities.SerieList import SerieList
|
||||
from src.core.entities.series import Serie
|
||||
from src.core.providers.provider_factory import Loaders
|
||||
from src.core.SerieScanner import SerieScanner
|
||||
from src.core.services.nfo_service import NFOService
|
||||
from src.core.services.tmdb_client import TMDBAPIError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -178,23 +176,9 @@ class SeriesApp:
|
||||
# Initialize empty list - series loaded later via load_series_from_list()
|
||||
# No need to call _init_list_sync() anymore
|
||||
|
||||
# Initialize NFO service if a TMDB API key is configured
|
||||
self.nfo_service: Optional[NFOService] = None
|
||||
try:
|
||||
from src.core.services.nfo_factory import get_nfo_factory
|
||||
|
||||
factory = get_nfo_factory()
|
||||
self.nfo_service = factory.create()
|
||||
logger.info("NFO service initialized successfully")
|
||||
except ValueError:
|
||||
logger.info(
|
||||
"NFO service not available — TMDB API key not configured"
|
||||
)
|
||||
self.nfo_service = None
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
logger.warning("Failed to initialize NFO service: %s", str(e))
|
||||
self.nfo_service = None
|
||||
|
||||
# NFO service removed - metadata handling moved to server layer
|
||||
self.nfo_service = None
|
||||
|
||||
logger.info(
|
||||
"SeriesApp initialized for directory: %s",
|
||||
directory_to_search,
|
||||
@@ -356,95 +340,6 @@ class SeriesApp:
|
||||
)
|
||||
return False
|
||||
|
||||
# Check and create NFO files if needed
|
||||
if self.nfo_service and settings.nfo_auto_create:
|
||||
try:
|
||||
# Check if NFO exists
|
||||
nfo_exists = await self.nfo_service.check_nfo_exists(
|
||||
serie_folder
|
||||
)
|
||||
|
||||
if not nfo_exists:
|
||||
logger.info(
|
||||
"NFO not found for %s, creating metadata...",
|
||||
serie_folder
|
||||
)
|
||||
|
||||
# Fire NFO creation started event
|
||||
self._events.download_status(
|
||||
DownloadStatusEventArgs(
|
||||
serie_folder=serie_folder,
|
||||
key=key,
|
||||
season=season,
|
||||
episode=episode,
|
||||
status="nfo_creating",
|
||||
message="Creating NFO metadata...",
|
||||
item_id=item_id,
|
||||
)
|
||||
)
|
||||
|
||||
# Create NFO and download media files
|
||||
try:
|
||||
# Use folder name as series name
|
||||
await self.nfo_service.create_tvshow_nfo(
|
||||
serie_name=serie_folder,
|
||||
serie_folder=serie_folder,
|
||||
download_poster=settings.nfo_download_poster,
|
||||
download_logo=settings.nfo_download_logo,
|
||||
download_fanart=settings.nfo_download_fanart
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"NFO and media files created for %s",
|
||||
serie_folder
|
||||
)
|
||||
|
||||
# Fire NFO creation completed event
|
||||
self._events.download_status(
|
||||
DownloadStatusEventArgs(
|
||||
serie_folder=serie_folder,
|
||||
key=key,
|
||||
season=season,
|
||||
episode=episode,
|
||||
status="nfo_completed",
|
||||
message="NFO metadata created",
|
||||
item_id=item_id,
|
||||
)
|
||||
)
|
||||
|
||||
except TMDBAPIError as tmdb_error:
|
||||
logger.warning(
|
||||
"Failed to create NFO for %s: %s",
|
||||
serie_folder,
|
||||
str(tmdb_error)
|
||||
)
|
||||
# Fire failed event (but continue with download)
|
||||
self._events.download_status(
|
||||
DownloadStatusEventArgs(
|
||||
serie_folder=serie_folder,
|
||||
key=key,
|
||||
season=season,
|
||||
episode=episode,
|
||||
status="nfo_failed",
|
||||
message=(
|
||||
f"NFO creation failed: "
|
||||
f"{str(tmdb_error)}"
|
||||
),
|
||||
item_id=item_id,
|
||||
)
|
||||
)
|
||||
else:
|
||||
logger.debug("NFO already exists for %s", serie_folder)
|
||||
|
||||
except Exception as nfo_error: # pylint: disable=broad-except
|
||||
logger.error(
|
||||
"Error checking/creating NFO for %s: %s",
|
||||
serie_folder,
|
||||
str(nfo_error),
|
||||
exc_info=True
|
||||
)
|
||||
# Don't fail the download if NFO creation fails
|
||||
|
||||
try:
|
||||
def download_progress_handler(progress_info):
|
||||
"""Handle download progress events from loader."""
|
||||
|
||||
@@ -1,237 +0,0 @@
|
||||
"""NFO Service Factory Module.
|
||||
|
||||
This module provides a centralized factory for creating NFOService instances
|
||||
with consistent configuration and initialization logic.
|
||||
|
||||
The factory supports both direct instantiation and FastAPI dependency injection,
|
||||
while remaining testable through optional dependency overrides.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from src.config.settings import settings
|
||||
from src.core.services.nfo_service import NFOService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class NFOServiceFactory:
|
||||
"""Factory for creating NFOService instances with consistent configuration.
|
||||
|
||||
This factory centralizes NFO service initialization logic that was previously
|
||||
duplicated across multiple modules (SeriesApp, SeriesManagerService, API endpoints).
|
||||
|
||||
The factory follows these precedence rules for configuration:
|
||||
1. Explicit parameters (highest priority)
|
||||
2. Environment variables via settings
|
||||
3. config.json via ConfigService (fallback)
|
||||
4. Raise error if TMDB API key unavailable
|
||||
|
||||
Example:
|
||||
>>> factory = NFOServiceFactory()
|
||||
>>> nfo_service = factory.create()
|
||||
>>> # Or with custom settings:
|
||||
>>> nfo_service = factory.create(tmdb_api_key="custom_key")
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the NFO service factory."""
|
||||
self._config_service = None
|
||||
|
||||
def create(
|
||||
self,
|
||||
tmdb_api_key: Optional[str] = None,
|
||||
anime_directory: Optional[str] = None,
|
||||
image_size: Optional[str] = None,
|
||||
auto_create: Optional[bool] = None
|
||||
) -> NFOService:
|
||||
"""Create an NFOService instance with proper configuration.
|
||||
|
||||
This method implements the configuration precedence:
|
||||
1. Use explicit parameters if provided
|
||||
2. Fall back to settings (from ENV vars)
|
||||
3. Fall back to config.json (only if ENV not set)
|
||||
4. Raise ValueError if TMDB API key still unavailable
|
||||
|
||||
Args:
|
||||
tmdb_api_key: TMDB API key (optional, falls back to settings/config)
|
||||
anime_directory: Anime directory path (optional, defaults to settings)
|
||||
image_size: Image size for downloads (optional, defaults to settings)
|
||||
auto_create: Whether to auto-create NFO files (optional, defaults to settings)
|
||||
|
||||
Returns:
|
||||
NFOService: Configured NFO service instance
|
||||
|
||||
Raises:
|
||||
ValueError: If TMDB API key cannot be determined from any source
|
||||
|
||||
Example:
|
||||
>>> factory = NFOServiceFactory()
|
||||
>>> # Use all defaults from settings
|
||||
>>> service = factory.create()
|
||||
>>> # Override specific settings
|
||||
>>> service = factory.create(auto_create=False)
|
||||
"""
|
||||
# Step 1: Determine TMDB API key with fallback logic
|
||||
api_key = tmdb_api_key or settings.tmdb_api_key
|
||||
|
||||
# Step 2: If no API key in settings, try config.json as fallback
|
||||
if not api_key:
|
||||
api_key = self._get_api_key_from_config()
|
||||
|
||||
# Step 3: Validate API key is available
|
||||
if not api_key:
|
||||
raise ValueError(
|
||||
"TMDB API key not configured. Set TMDB_API_KEY environment "
|
||||
"variable or configure in config.json (nfo.tmdb_api_key)."
|
||||
)
|
||||
|
||||
# Step 4: Use provided values or fall back to settings
|
||||
directory = anime_directory or settings.anime_directory
|
||||
size = image_size or settings.nfo_image_size
|
||||
auto = auto_create if auto_create is not None else settings.nfo_auto_create
|
||||
|
||||
# Step 5: Create and return the service
|
||||
logger.debug(
|
||||
"Creating NFOService: directory=%s, size=%s, auto_create=%s",
|
||||
directory, size, auto
|
||||
)
|
||||
|
||||
return NFOService(
|
||||
tmdb_api_key=api_key,
|
||||
anime_directory=directory,
|
||||
image_size=size,
|
||||
auto_create=auto
|
||||
)
|
||||
|
||||
def create_optional(
|
||||
self,
|
||||
tmdb_api_key: Optional[str] = None,
|
||||
anime_directory: Optional[str] = None,
|
||||
image_size: Optional[str] = None,
|
||||
auto_create: Optional[bool] = None
|
||||
) -> Optional[NFOService]:
|
||||
"""Create an NFOService instance, returning None if configuration unavailable.
|
||||
|
||||
This is a convenience method for cases where NFO service is optional.
|
||||
Unlike create(), this returns None instead of raising ValueError when
|
||||
the TMDB API key is not configured.
|
||||
|
||||
Args:
|
||||
tmdb_api_key: TMDB API key (optional)
|
||||
anime_directory: Anime directory path (optional)
|
||||
image_size: Image size for downloads (optional)
|
||||
auto_create: Whether to auto-create NFO files (optional)
|
||||
|
||||
Returns:
|
||||
Optional[NFOService]: Configured service or None if key unavailable
|
||||
|
||||
Example:
|
||||
>>> factory = NFOServiceFactory()
|
||||
>>> service = factory.create_optional()
|
||||
>>> if service:
|
||||
... service.create_tvshow_nfo(...)
|
||||
"""
|
||||
try:
|
||||
return self.create(
|
||||
tmdb_api_key=tmdb_api_key,
|
||||
anime_directory=anime_directory,
|
||||
image_size=image_size,
|
||||
auto_create=auto_create
|
||||
)
|
||||
except ValueError as e:
|
||||
logger.debug("NFO service not available: %s", e)
|
||||
return None
|
||||
|
||||
def _get_api_key_from_config(self) -> Optional[str]:
|
||||
"""Get TMDB API key from config.json as fallback.
|
||||
|
||||
This method is only called when the API key is not in settings
|
||||
(i.e., not set via environment variable). It provides backward
|
||||
compatibility with config.json configuration.
|
||||
|
||||
Returns:
|
||||
Optional[str]: API key from config.json, or None if unavailable
|
||||
"""
|
||||
try:
|
||||
# Lazy import to avoid circular dependencies
|
||||
from src.server.services.config_service import get_config_service
|
||||
|
||||
if self._config_service is None:
|
||||
self._config_service = get_config_service()
|
||||
|
||||
config = self._config_service.load_config()
|
||||
|
||||
if config.nfo and config.nfo.tmdb_api_key:
|
||||
logger.debug("Using TMDB API key from config.json")
|
||||
return config.nfo.tmdb_api_key
|
||||
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
logger.debug("Could not load API key from config.json: %s", e)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# Global factory instance for convenience
|
||||
_factory_instance: Optional[NFOServiceFactory] = None
|
||||
|
||||
|
||||
def get_nfo_factory() -> NFOServiceFactory:
|
||||
"""Get the global NFO service factory instance.
|
||||
|
||||
This function provides a singleton factory instance for the application.
|
||||
The singleton pattern here is for the factory itself (which is stateless),
|
||||
not for the NFO service instances it creates.
|
||||
|
||||
Returns:
|
||||
NFOServiceFactory: The global factory instance
|
||||
|
||||
Example:
|
||||
>>> factory = get_nfo_factory()
|
||||
>>> service = factory.create()
|
||||
"""
|
||||
global _factory_instance
|
||||
|
||||
if _factory_instance is None:
|
||||
_factory_instance = NFOServiceFactory()
|
||||
|
||||
return _factory_instance
|
||||
|
||||
|
||||
def create_nfo_service(
|
||||
tmdb_api_key: Optional[str] = None,
|
||||
anime_directory: Optional[str] = None,
|
||||
image_size: Optional[str] = None,
|
||||
auto_create: Optional[bool] = None
|
||||
) -> NFOService:
|
||||
"""Convenience function to create an NFOService instance.
|
||||
|
||||
This is a shorthand for get_nfo_factory().create() that can be used
|
||||
when you need a quick NFO service instance without interacting with
|
||||
the factory directly.
|
||||
|
||||
Args:
|
||||
tmdb_api_key: TMDB API key (optional)
|
||||
anime_directory: Anime directory path (optional)
|
||||
image_size: Image size for downloads (optional)
|
||||
auto_create: Whether to auto-create NFO files (optional)
|
||||
|
||||
Returns:
|
||||
NFOService: Configured NFO service instance
|
||||
|
||||
Raises:
|
||||
ValueError: If TMDB API key cannot be determined
|
||||
|
||||
Example:
|
||||
>>> service = create_nfo_service()
|
||||
>>> # Or with custom settings:
|
||||
>>> service = create_nfo_service(auto_create=False)
|
||||
"""
|
||||
factory = get_nfo_factory()
|
||||
return factory.create(
|
||||
tmdb_api_key=tmdb_api_key,
|
||||
anime_directory=anime_directory,
|
||||
image_size=image_size,
|
||||
auto_create=auto_create
|
||||
)
|
||||
@@ -1,228 +0,0 @@
|
||||
"""NFO repair service for detecting and fixing incomplete tvshow.nfo files.
|
||||
|
||||
This module provides utilities to check whether an existing ``tvshow.nfo``
|
||||
contains all required tags and to trigger a repair (re-fetch from TMDB) when
|
||||
needed.
|
||||
|
||||
Example:
|
||||
>>> service = NfoRepairService(nfo_service)
|
||||
>>> repaired = await service.repair_series(Path("/anime/Attack on Titan"), "Attack on Titan")
|
||||
"""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
|
||||
from lxml import etree
|
||||
|
||||
from src.core.services.nfo_service import NFOService
|
||||
from src.core.services.tmdb_client import TMDBAPIError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# XPath relative to <tvshow> root → human-readable label
|
||||
REQUIRED_TAGS: Dict[str, str] = {
|
||||
"./title": "title",
|
||||
"./originaltitle": "originaltitle",
|
||||
"./year": "year",
|
||||
"./plot": "plot",
|
||||
"./runtime": "runtime",
|
||||
"./premiered": "premiered",
|
||||
"./status": "status",
|
||||
"./imdbid": "imdbid",
|
||||
"./genre": "genre",
|
||||
"./studio": "studio",
|
||||
"./country": "country",
|
||||
"./actor/name": "actor/name",
|
||||
"./watched": "watched",
|
||||
}
|
||||
|
||||
|
||||
def parse_nfo_tags(nfo_path: Path) -> Dict[str, List[str]]:
|
||||
"""Parse an existing tvshow.nfo and return present tag values.
|
||||
|
||||
Evaluates every XPath in :data:`REQUIRED_TAGS` against the document root
|
||||
and collects all non-empty text values.
|
||||
|
||||
Args:
|
||||
nfo_path: Absolute path to the ``tvshow.nfo`` file.
|
||||
|
||||
Returns:
|
||||
Mapping of XPath expression → list of non-empty text strings found in
|
||||
the document. Returns an empty dict on any error (missing file,
|
||||
invalid XML, permission error).
|
||||
|
||||
Example:
|
||||
>>> tags = parse_nfo_tags(Path("/anime/Attack on Titan/tvshow.nfo"))
|
||||
>>> tags.get("./title")
|
||||
['Attack on Titan']
|
||||
"""
|
||||
if not nfo_path.exists():
|
||||
logger.debug("NFO file not found: %s", nfo_path)
|
||||
return {}
|
||||
|
||||
try:
|
||||
tree = etree.parse(str(nfo_path))
|
||||
root = tree.getroot()
|
||||
|
||||
result: Dict[str, List[str]] = {}
|
||||
for xpath in REQUIRED_TAGS:
|
||||
elements = root.findall(xpath)
|
||||
result[xpath] = [e.text for e in elements if e.text]
|
||||
|
||||
return result
|
||||
|
||||
except etree.XMLSyntaxError as exc:
|
||||
logger.warning("Malformed XML in %s: %s", nfo_path, exc)
|
||||
return {}
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
logger.warning("Unexpected error parsing %s: %s", nfo_path, exc)
|
||||
return {}
|
||||
|
||||
|
||||
def find_missing_tags(nfo_path: Path) -> List[str]:
|
||||
"""Return tags that are absent or empty in the NFO.
|
||||
|
||||
Args:
|
||||
nfo_path: Absolute path to the ``tvshow.nfo`` file.
|
||||
|
||||
Returns:
|
||||
List of human-readable tag labels (values from :data:`REQUIRED_TAGS`)
|
||||
whose XPath matched no elements or only elements with empty text.
|
||||
An empty list means the NFO is complete.
|
||||
|
||||
Example:
|
||||
>>> missing = find_missing_tags(Path("/anime/series/tvshow.nfo"))
|
||||
>>> if missing:
|
||||
... print("Missing:", missing)
|
||||
"""
|
||||
parsed = parse_nfo_tags(nfo_path)
|
||||
missing: List[str] = []
|
||||
for xpath, label in REQUIRED_TAGS.items():
|
||||
if not parsed.get(xpath):
|
||||
missing.append(label)
|
||||
return missing
|
||||
|
||||
|
||||
def nfo_needs_repair(nfo_path: Path) -> bool:
|
||||
"""Return ``True`` if the NFO is missing any required tag.
|
||||
|
||||
Args:
|
||||
nfo_path: Absolute path to the ``tvshow.nfo`` file.
|
||||
|
||||
Returns:
|
||||
True if :func:`find_missing_tags` returns a non-empty list.
|
||||
|
||||
Example:
|
||||
>>> if nfo_needs_repair(Path("/anime/series/tvshow.nfo")):
|
||||
... await service.repair_series(series_path, series_name)
|
||||
"""
|
||||
return bool(find_missing_tags(nfo_path))
|
||||
|
||||
|
||||
def _read_tmdb_id(nfo_path: Path) -> int | None:
|
||||
"""Return the TMDB ID stored in an existing NFO, or ``None``.
|
||||
|
||||
Checks both ``<tmdbid>`` and ``<uniqueid type="tmdb">`` elements.
|
||||
|
||||
Args:
|
||||
nfo_path: Absolute path to the ``tvshow.nfo`` file.
|
||||
|
||||
Returns:
|
||||
Integer TMDB ID, or ``None`` if not found or not parseable.
|
||||
"""
|
||||
if not nfo_path.exists():
|
||||
return None
|
||||
try:
|
||||
root = etree.parse(str(nfo_path)).getroot()
|
||||
|
||||
for uniqueid in root.findall(".//uniqueid"):
|
||||
if uniqueid.get("type") == "tmdb" and uniqueid.text:
|
||||
return int(uniqueid.text)
|
||||
|
||||
tmdbid_elem = root.find(".//tmdbid")
|
||||
if tmdbid_elem is not None and tmdbid_elem.text:
|
||||
return int(tmdbid_elem.text)
|
||||
|
||||
except (etree.XMLSyntaxError, ValueError):
|
||||
pass
|
||||
except Exception: # pylint: disable=broad-except
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
class NfoRepairService:
|
||||
"""Service that detects and repairs incomplete tvshow.nfo files.
|
||||
|
||||
Wraps the module-level helpers with structured logging and delegates
|
||||
the actual TMDB re-fetch to an injected :class:`NFOService` instance.
|
||||
|
||||
Attributes:
|
||||
_nfo_service: The underlying NFOService used to update NFOs.
|
||||
"""
|
||||
|
||||
def __init__(self, nfo_service: NFOService) -> None:
|
||||
"""Initialise the repair service.
|
||||
|
||||
Args:
|
||||
nfo_service: Configured :class:`NFOService` instance.
|
||||
"""
|
||||
self._nfo_service = nfo_service
|
||||
|
||||
async def repair_series(self, series_path: Path, series_name: str) -> bool:
|
||||
"""Repair an NFO file if required tags are missing.
|
||||
|
||||
Checks ``{series_path}/tvshow.nfo`` for completeness. If tags are
|
||||
missing, logs them and calls
|
||||
``NFOService.update_tvshow_nfo(series_name)`` to re-fetch metadata
|
||||
from TMDB.
|
||||
|
||||
Args:
|
||||
series_path: Absolute path to the series folder.
|
||||
series_name: Series folder name used as the identifier for
|
||||
:meth:`NFOService.update_tvshow_nfo`.
|
||||
|
||||
Returns:
|
||||
``True`` if a repair was triggered, ``False`` if the NFO was
|
||||
already complete (or did not exist).
|
||||
"""
|
||||
nfo_path = series_path / "tvshow.nfo"
|
||||
missing = find_missing_tags(nfo_path)
|
||||
|
||||
if not missing:
|
||||
logger.info(
|
||||
"NFO repair skipped — complete: %s",
|
||||
series_name,
|
||||
)
|
||||
return False
|
||||
|
||||
logger.info(
|
||||
"NFO repair triggered for %s — missing tags: %s",
|
||||
series_name,
|
||||
", ".join(missing),
|
||||
)
|
||||
|
||||
try:
|
||||
await self._nfo_service.update_tvshow_nfo(
|
||||
series_name,
|
||||
download_media=False,
|
||||
)
|
||||
except TMDBAPIError as e:
|
||||
if "No TMDB ID found" in str(e):
|
||||
# No TMDB ID in existing NFO — create new one via search
|
||||
logger.info(
|
||||
"NFO has no TMDB ID, creating new NFO via TMDB search"
|
||||
)
|
||||
await self._nfo_service.create_tvshow_nfo(
|
||||
serie_name=series_name,
|
||||
serie_folder=series_name,
|
||||
download_poster=False,
|
||||
download_logo=False,
|
||||
download_fanart=False,
|
||||
)
|
||||
else:
|
||||
raise
|
||||
|
||||
logger.info("NFO repair completed: %s", series_name)
|
||||
return True
|
||||
@@ -1,891 +0,0 @@
|
||||
"""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
|
||||
import unicodedata
|
||||
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
|
||||
from src.core.entities.nfo_models import TVShowNFO
|
||||
|
||||
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
|
||||
|
||||
async def __aenter__(self) -> "NFOService":
|
||||
"""Enter async context manager."""
|
||||
await self.tmdb_client.__aenter__()
|
||||
await self.image_downloader.__aenter__()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Exit async context manager and cleanup resources."""
|
||||
await self.tmdb_client.close()
|
||||
await self.image_downloader.close()
|
||||
return False
|
||||
|
||||
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 the last year in parentheses at the end: (YYYY)
|
||||
match = re.search(r'\((\d{4})\)\s*$', serie_name)
|
||||
if match:
|
||||
year = int(match.group(1))
|
||||
# Strip ALL trailing year suffixes to get a fully clean name
|
||||
clean_name = re.sub(r'(\s*\(\d{4}\))+\s*$', '', serie_name).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,
|
||||
alt_titles: Optional[List[str]] = None
|
||||
) -> 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
|
||||
alt_titles: Alternative titles (e.g., Japanese title) for fallback search
|
||||
|
||||
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("Extracted year %s from series name", year)
|
||||
|
||||
# Use clean name for search
|
||||
search_name = clean_name
|
||||
|
||||
logger.info("Creating NFO for %s (year: %s)", search_name, year)
|
||||
|
||||
folder_path = self.anime_directory / serie_folder
|
||||
if not folder_path.exists():
|
||||
logger.info("Creating series folder: %s", folder_path)
|
||||
folder_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Check for existing NFO with TMDB ID to skip search
|
||||
nfo_path = folder_path / "tvshow.nfo"
|
||||
existing_ids = None
|
||||
if nfo_path.exists():
|
||||
try:
|
||||
existing_ids = self.parse_nfo_ids(nfo_path)
|
||||
if existing_ids.get("tmdb_id"):
|
||||
logger.info(
|
||||
"Found existing TMDB ID %s in NFO, using directly",
|
||||
existing_ids["tmdb_id"]
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug("Could not parse existing NFO IDs: %s", e)
|
||||
|
||||
try:
|
||||
await self.tmdb_client._ensure_session()
|
||||
|
||||
# Use existing TMDB ID if found, otherwise search
|
||||
if existing_ids and existing_ids.get("tmdb_id"):
|
||||
tv_id = existing_ids["tmdb_id"]
|
||||
logger.info("Fetching details directly for TMDB ID: %s", tv_id)
|
||||
details = await self.tmdb_client.get_tv_show_details(
|
||||
tv_id,
|
||||
append_to_response="credits,external_ids,images"
|
||||
)
|
||||
content_ratings = await self.tmdb_client.get_tv_show_content_ratings(tv_id)
|
||||
tv_show = {"id": tv_id, "name": details.get("name", serie_name)}
|
||||
search_source = "nfo_override"
|
||||
else:
|
||||
# Search for TV show - try multiple strategies
|
||||
tv_show, search_source = await self._search_with_fallback(
|
||||
search_name, year, alt_titles
|
||||
)
|
||||
tv_id = tv_show["id"]
|
||||
|
||||
logger.info("Found match: %s (ID: %s)", tv_show['name'], tv_id)
|
||||
|
||||
# Get detailed information with multi-language image support
|
||||
# Skip if we already fetched details via nfo_override
|
||||
if search_source != "nfo_override":
|
||||
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
|
||||
if not search_overview:
|
||||
try:
|
||||
logger.debug(
|
||||
"No overview in German search result, trying en-US search fallback for: %s",
|
||||
search_name,
|
||||
)
|
||||
en_search_results = await self.tmdb_client.search_tv_show(
|
||||
search_name,
|
||||
language="en-US",
|
||||
)
|
||||
if en_search_results.get("results"):
|
||||
en_match = self._find_best_match(
|
||||
en_search_results["results"], search_name, year
|
||||
)
|
||||
search_overview = en_match.get("overview") or None
|
||||
if search_overview:
|
||||
logger.info(
|
||||
"Using en-US search overview fallback for %s",
|
||||
search_name,
|
||||
)
|
||||
except (TMDBAPIError, Exception) as exc:
|
||||
logger.warning(
|
||||
"Failed en-US search fallback for overview: %s",
|
||||
exc,
|
||||
)
|
||||
|
||||
details = await self._enrich_details_with_fallback(
|
||||
details, search_overview=search_overview
|
||||
)
|
||||
else:
|
||||
# When using nfo_override, content_ratings already fetched
|
||||
pass
|
||||
|
||||
# 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("Created NFO: %s", 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
|
||||
finally:
|
||||
await self.tmdb_client.close()
|
||||
|
||||
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("Updating NFO for %s", 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("Found TMDB ID: %s", 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}")
|
||||
|
||||
try:
|
||||
await self.tmdb_client._ensure_session()
|
||||
logger.debug("Fetching fresh data for TMDB ID: %s", 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("Updated NFO: %s", 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
|
||||
finally:
|
||||
await self.tmdb_client.close()
|
||||
|
||||
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("NFO file not found: %s", 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("Invalid XML in NFO file %s: %s", nfo_path, e)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
logger.error("Error parsing NFO file %s: %s", nfo_path, e)
|
||||
|
||||
return result
|
||||
|
||||
def parse_nfo_year(self, nfo_path: Path) -> Optional[int]:
|
||||
"""Parse year from an existing NFO file.
|
||||
|
||||
Extracts year from <year> or <premiered> elements.
|
||||
|
||||
Args:
|
||||
nfo_path: Path to tvshow.nfo file
|
||||
|
||||
Returns:
|
||||
Year as integer if found, None otherwise.
|
||||
|
||||
Example:
|
||||
>>> year = nfo_service.parse_nfo_year(Path("/anime/series/tvshow.nfo"))
|
||||
>>> print(year)
|
||||
2013
|
||||
"""
|
||||
if not nfo_path.exists():
|
||||
logger.debug("NFO file not found: %s", nfo_path)
|
||||
return None
|
||||
|
||||
try:
|
||||
tree = etree.parse(str(nfo_path))
|
||||
root = tree.getroot()
|
||||
|
||||
# Try <year> element first
|
||||
year_elem = root.find(".//year")
|
||||
if year_elem is not None and year_elem.text:
|
||||
try:
|
||||
year = int(year_elem.text)
|
||||
if 1900 <= year <= 2100:
|
||||
logger.debug("Found year in NFO: %d", year)
|
||||
return year
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Fallback: try <premiered> element (format: YYYY-MM-DD)
|
||||
premiered_elem = root.find(".//premiered")
|
||||
if premiered_elem is not None and premiered_elem.text:
|
||||
if premiered_elem.text and len(premiered_elem.text) >= 4:
|
||||
try:
|
||||
year = int(premiered_elem.text[:4])
|
||||
if 1900 <= year <= 2100:
|
||||
logger.debug("Found year from premiered in NFO: %d", year)
|
||||
return year
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
logger.debug("No year found in NFO: %s", nfo_path)
|
||||
|
||||
except etree.XMLSyntaxError as e:
|
||||
logger.error("Invalid XML in NFO file %s: %s", nfo_path, e)
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
logger.error("Error parsing year from NFO file %s: %s", nfo_path, e)
|
||||
|
||||
return None
|
||||
|
||||
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("Found year match: %s (%s)", result['name'], first_air_date)
|
||||
return result
|
||||
|
||||
# Return first result (usually best match)
|
||||
return results[0]
|
||||
|
||||
async def _search_with_fallback(
|
||||
self,
|
||||
primary_query: str,
|
||||
year: Optional[int],
|
||||
alt_titles: Optional[List[str]] = None
|
||||
) -> Tuple[Dict[str, Any], str]:
|
||||
"""Search TMDB with fallback strategies.
|
||||
|
||||
Tries multiple search strategies in order:
|
||||
1. Primary query with year filter
|
||||
2. Alternative titles (e.g., Japanese name)
|
||||
3. Multi-language search (en-US)
|
||||
4. Search without year constraint
|
||||
5. Punctuation-normalized search
|
||||
|
||||
Args:
|
||||
primary_query: Primary search term
|
||||
year: Release year for filtering
|
||||
alt_titles: Alternative titles to try if primary fails
|
||||
|
||||
Returns:
|
||||
Tuple of (matched TV show dict, source description string)
|
||||
|
||||
Raises:
|
||||
TMDBAPIError: If all search strategies fail
|
||||
"""
|
||||
search_strategies = [
|
||||
# Strategy 1: Primary query as-is
|
||||
{"query": primary_query, "year": year, "lang": "de-DE", "desc": "primary"},
|
||||
]
|
||||
|
||||
# Strategy 2: Try alt titles (typically Japanese)
|
||||
if alt_titles:
|
||||
for alt in alt_titles:
|
||||
if alt != primary_query:
|
||||
search_strategies.append(
|
||||
{"query": alt, "year": year, "lang": "ja-JP", "desc": f"alt_title:{alt}"}
|
||||
)
|
||||
search_strategies.append(
|
||||
{"query": alt, "year": year, "lang": "en-US", "desc": f"alt_title:{alt}"}
|
||||
)
|
||||
|
||||
# Strategy 3: Try English search
|
||||
search_strategies.append(
|
||||
{"query": primary_query, "year": year, "lang": "en-US", "desc": "english"}
|
||||
)
|
||||
|
||||
# Strategy 4: Try without year constraint
|
||||
if year:
|
||||
search_strategies.append(
|
||||
{"query": primary_query, "year": None, "lang": "de-DE", "desc": "no_year"}
|
||||
)
|
||||
|
||||
# Strategy 5: Normalize punctuation
|
||||
normalized = self._normalize_query_for_search(primary_query)
|
||||
if normalized != primary_query:
|
||||
search_strategies.append(
|
||||
{"query": normalized, "year": year, "lang": "de-DE", "desc": f"normalized:{normalized}"}
|
||||
)
|
||||
|
||||
# Strategy 6: Try search/multi for series indexed as movies
|
||||
search_strategies.append(
|
||||
{"query": primary_query, "year": year, "lang": "en-US", "desc": "multi_search", "use_multi": True}
|
||||
)
|
||||
|
||||
last_error = None
|
||||
for strategy in search_strategies:
|
||||
query = strategy["query"]
|
||||
lang = strategy["lang"]
|
||||
desc = strategy["desc"]
|
||||
use_multi = strategy.get("use_multi", False)
|
||||
|
||||
try:
|
||||
logger.debug(
|
||||
"TMDB search attempt: query='%s', lang=%s, year=%s, strategy=%s",
|
||||
query, lang, strategy["year"], desc
|
||||
)
|
||||
|
||||
# Use search/multi for multi_search strategy
|
||||
if use_multi:
|
||||
search_results = await self.tmdb_client.search_multi(
|
||||
query,
|
||||
language=lang
|
||||
)
|
||||
# Filter for TV shows only
|
||||
if search_results.get("results"):
|
||||
tv_results = [
|
||||
r for r in search_results["results"]
|
||||
if r.get("media_type") == "tv"
|
||||
]
|
||||
if tv_results:
|
||||
search_results["results"] = tv_results
|
||||
else:
|
||||
search_results["results"] = []
|
||||
else:
|
||||
search_results = await self.tmdb_client.search_tv_show(
|
||||
query,
|
||||
language=lang
|
||||
)
|
||||
|
||||
if search_results.get("results"):
|
||||
# Apply year filter if we have one
|
||||
results = search_results["results"]
|
||||
if strategy["year"]:
|
||||
year_filtered = [
|
||||
r for r in results
|
||||
if r.get("first_air_date", "").startswith(str(strategy["year"]))
|
||||
]
|
||||
if year_filtered:
|
||||
match = year_filtered[0]
|
||||
else:
|
||||
# Year didn't match, still use first result but log it
|
||||
match = results[0]
|
||||
logger.debug(
|
||||
"Year %s not found in results for '%s', using: %s",
|
||||
strategy["year"], query, match["name"]
|
||||
)
|
||||
else:
|
||||
match = results[0]
|
||||
|
||||
logger.info(
|
||||
"TMDB search succeeded: '%s' found via strategy '%s' (ID: %s)",
|
||||
match["name"], desc, match["id"]
|
||||
)
|
||||
return match, desc
|
||||
else:
|
||||
logger.debug("No results for '%s' via %s", query, desc)
|
||||
|
||||
except TMDBAPIError as e:
|
||||
last_error = e
|
||||
logger.debug("Search strategy '%s' failed: %s", desc, e)
|
||||
continue
|
||||
|
||||
# All strategies exhausted
|
||||
raise TMDBAPIError(
|
||||
f"No results found for: {primary_query} (tried {len(search_strategies)} strategies)"
|
||||
)
|
||||
|
||||
def _normalize_query_for_search(self, query: str) -> str:
|
||||
"""Normalize query by removing punctuation and special chars.
|
||||
|
||||
Args:
|
||||
query: Original search query
|
||||
|
||||
Returns:
|
||||
Query with punctuation removed
|
||||
"""
|
||||
# Remove common punctuation but keep CJK characters
|
||||
normalized = unicodedata.normalize('NFKC', query)
|
||||
# Remove punctuation but not CJK
|
||||
normalized = re.sub(r'[^\w\s\u3000-\u9fff\u4e00-\u9faf]', '', normalized)
|
||||
# Collapse multiple spaces
|
||||
normalized = re.sub(r'\s+', ' ', normalized).strip()
|
||||
return normalized
|
||||
|
||||
|
||||
|
||||
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("Media download results: %s", results)
|
||||
return results
|
||||
|
||||
|
||||
|
||||
async def close(self):
|
||||
"""Clean up resources."""
|
||||
await self.tmdb_client.close()
|
||||
await self.image_downloader.close()
|
||||
|
||||
async def create_minimal_nfo(
|
||||
self,
|
||||
serie_name: str,
|
||||
serie_folder: str,
|
||||
year: Optional[int] = None
|
||||
) -> Path:
|
||||
"""Create minimal tvshow.nfo when TMDB lookup fails.
|
||||
|
||||
Creates a basic NFO with just the title (and year if available)
|
||||
so the series is tracked even without TMDB metadata.
|
||||
|
||||
Args:
|
||||
serie_name: Name of the series (may include year in parentheses)
|
||||
serie_folder: Series folder name
|
||||
year: Optional release year
|
||||
|
||||
Returns:
|
||||
Path to created NFO file
|
||||
|
||||
Raises:
|
||||
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
|
||||
|
||||
folder_path = self.anime_directory / serie_folder
|
||||
if not folder_path.exists():
|
||||
logger.info("Creating series folder: %s", folder_path)
|
||||
folder_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create minimal NFO model with just title and year
|
||||
nfo_model = TVShowNFO(
|
||||
title=clean_name,
|
||||
year=year,
|
||||
plot=f"No metadata available for {clean_name}. TMDB lookup failed."
|
||||
)
|
||||
|
||||
# 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("Created minimal NFO (no TMDB): %s", nfo_path)
|
||||
|
||||
return nfo_path
|
||||
@@ -1,309 +0,0 @@
|
||||
"""Service for managing series with NFO metadata support.
|
||||
|
||||
This service layer component orchestrates SerieList (core entity) with
|
||||
NFOService to provide automatic NFO creation and updates during series scans.
|
||||
|
||||
This follows clean architecture principles by keeping the core entities
|
||||
independent of external services like TMDB API.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from src.config.settings import settings
|
||||
from src.core.entities.SerieList import SerieList
|
||||
from src.core.services.nfo_service import NFOService
|
||||
from src.core.services.tmdb_client import TMDBAPIError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SeriesManagerService:
|
||||
"""Service for managing series with optional NFO metadata support.
|
||||
|
||||
This service wraps SerieList and adds NFO creation/update capabilities
|
||||
based on configuration settings. It maintains clean separation between
|
||||
core entities and external services.
|
||||
|
||||
Attributes:
|
||||
serie_list: SerieList instance for series management
|
||||
nfo_service: Optional NFOService for metadata management
|
||||
auto_create_nfo: Whether to auto-create NFO files
|
||||
update_on_scan: Whether to update existing NFO files
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
anime_directory: str,
|
||||
tmdb_api_key: Optional[str] = None,
|
||||
auto_create_nfo: bool = False,
|
||||
update_on_scan: bool = False,
|
||||
download_poster: bool = True,
|
||||
download_logo: bool = True,
|
||||
download_fanart: bool = True,
|
||||
image_size: str = "original"
|
||||
):
|
||||
"""Initialize series manager service.
|
||||
|
||||
Args:
|
||||
anime_directory: Base directory for anime series
|
||||
tmdb_api_key: TMDB API key (optional, required for NFO features)
|
||||
auto_create_nfo: Automatically create NFO files when scanning
|
||||
update_on_scan: Update existing NFO files when scanning
|
||||
download_poster: Download poster.jpg
|
||||
download_logo: Download logo.png
|
||||
download_fanart: Download fanart.jpg
|
||||
image_size: Image size to download
|
||||
"""
|
||||
self.anime_directory = anime_directory
|
||||
# Skip automatic folder scanning - we load from database instead
|
||||
self.serie_list = SerieList(anime_directory, skip_load=True)
|
||||
|
||||
# NFO configuration
|
||||
self.auto_create_nfo = auto_create_nfo
|
||||
self.update_on_scan = update_on_scan
|
||||
self.download_poster = download_poster
|
||||
self.download_logo = download_logo
|
||||
self.download_fanart = download_fanart
|
||||
|
||||
# Initialize NFO service if API key provided and NFO features enabled
|
||||
self.nfo_service: Optional[NFOService] = None
|
||||
if tmdb_api_key and (auto_create_nfo or update_on_scan):
|
||||
try:
|
||||
from src.core.services.nfo_factory import get_nfo_factory
|
||||
factory = get_nfo_factory()
|
||||
self.nfo_service = factory.create(
|
||||
tmdb_api_key=tmdb_api_key,
|
||||
anime_directory=anime_directory,
|
||||
image_size=image_size,
|
||||
auto_create=auto_create_nfo
|
||||
)
|
||||
logger.info("NFO service initialized (auto_create=%s, update=%s)",
|
||||
auto_create_nfo, update_on_scan)
|
||||
except (ValueError, Exception) as e: # pylint: disable=broad-except
|
||||
logger.warning(
|
||||
"Failed to initialize NFO service: %s", str(e)
|
||||
)
|
||||
self.nfo_service = None
|
||||
elif auto_create_nfo or update_on_scan:
|
||||
logger.warning(
|
||||
"NFO features requested but TMDB_API_KEY not provided. "
|
||||
"NFO creation/updates will be skipped."
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_settings(cls) -> "SeriesManagerService":
|
||||
"""Create SeriesManagerService from application settings.
|
||||
|
||||
Returns:
|
||||
Configured SeriesManagerService instance
|
||||
"""
|
||||
return cls(
|
||||
anime_directory=settings.anime_directory,
|
||||
tmdb_api_key=settings.tmdb_api_key,
|
||||
auto_create_nfo=settings.nfo_auto_create,
|
||||
update_on_scan=settings.nfo_update_on_scan,
|
||||
download_poster=settings.nfo_download_poster,
|
||||
download_logo=settings.nfo_download_logo,
|
||||
download_fanart=settings.nfo_download_fanart,
|
||||
image_size=settings.nfo_image_size
|
||||
)
|
||||
|
||||
async def process_nfo_for_series(
|
||||
self,
|
||||
serie_folder: str,
|
||||
serie_name: str,
|
||||
serie_key: str,
|
||||
year: Optional[int] = None
|
||||
):
|
||||
"""Process NFO file for a series (create or update).
|
||||
|
||||
Args:
|
||||
serie_folder: Series folder name
|
||||
serie_name: Series display name
|
||||
serie_key: Series unique identifier for database updates
|
||||
year: Release year (helps with TMDB matching)
|
||||
"""
|
||||
if not self.nfo_service:
|
||||
return
|
||||
|
||||
nfo_exists = False
|
||||
ids = {}
|
||||
|
||||
try:
|
||||
folder_path = Path(self.anime_directory) / serie_folder
|
||||
nfo_path = folder_path / "tvshow.nfo"
|
||||
nfo_exists = await self.nfo_service.check_nfo_exists(serie_folder)
|
||||
|
||||
# If NFO exists, parse IDs and update database
|
||||
if nfo_exists:
|
||||
logger.debug("Parsing IDs from existing NFO for '%s'", serie_name)
|
||||
ids = self.nfo_service.parse_nfo_ids(nfo_path)
|
||||
|
||||
if ids["tmdb_id"] or ids["tvdb_id"]:
|
||||
# Update database using service layer
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from src.server.database.connection import get_db_session
|
||||
from src.server.database.service import AnimeSeriesService
|
||||
|
||||
async with get_db_session() as db:
|
||||
series = await AnimeSeriesService.get_by_key(db, serie_key)
|
||||
|
||||
if series:
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# Prepare update fields
|
||||
update_fields = {
|
||||
"has_nfo": True,
|
||||
"nfo_updated_at": now,
|
||||
}
|
||||
|
||||
if series.nfo_created_at is None:
|
||||
update_fields["nfo_created_at"] = now
|
||||
|
||||
if ids["tmdb_id"] is not None:
|
||||
update_fields["tmdb_id"] = ids["tmdb_id"]
|
||||
logger.debug(
|
||||
f"Updated TMDB ID for '{serie_name}': "
|
||||
f"{ids['tmdb_id']}"
|
||||
)
|
||||
|
||||
if ids["tvdb_id"] is not None:
|
||||
update_fields["tvdb_id"] = ids["tvdb_id"]
|
||||
logger.debug(
|
||||
f"Updated TVDB ID for '{serie_name}': "
|
||||
f"{ids['tvdb_id']}"
|
||||
)
|
||||
|
||||
# Use service layer for update
|
||||
await AnimeSeriesService.update(db, series.id, **update_fields)
|
||||
await db.commit()
|
||||
|
||||
logger.info(
|
||||
f"Updated database with IDs from NFO for "
|
||||
f"'{serie_name}' - TMDB: {ids['tmdb_id']}, "
|
||||
f"TVDB: {ids['tvdb_id']}"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f"Series not found in database for NFO ID "
|
||||
f"update: {serie_key}"
|
||||
)
|
||||
|
||||
# Create NFO file only if it doesn't exist and auto_create enabled
|
||||
if not nfo_exists and self.auto_create_nfo:
|
||||
logger.info(
|
||||
f"Creating NFO for '{serie_name}' ({serie_folder})"
|
||||
)
|
||||
try:
|
||||
await self.nfo_service.create_tvshow_nfo(
|
||||
serie_name=serie_name,
|
||||
serie_folder=serie_folder,
|
||||
year=year,
|
||||
download_poster=self.download_poster,
|
||||
download_logo=self.download_logo,
|
||||
download_fanart=self.download_fanart
|
||||
)
|
||||
logger.info("Successfully created NFO for '%s'", serie_name)
|
||||
except TMDBAPIError as create_error:
|
||||
# TMDB lookup failed, create minimal NFO to track the series
|
||||
logger.warning(
|
||||
"TMDB lookup failed for '%s', creating minimal NFO: %s",
|
||||
serie_name, create_error
|
||||
)
|
||||
try:
|
||||
await self.nfo_service.create_minimal_nfo(
|
||||
serie_name=serie_name,
|
||||
serie_folder=serie_folder,
|
||||
year=year
|
||||
)
|
||||
logger.info("Created minimal NFO for '%s'", serie_name)
|
||||
except Exception as minimal_error:
|
||||
logger.error(
|
||||
"Failed to create minimal NFO for '%s': %s",
|
||||
serie_name, minimal_error
|
||||
)
|
||||
elif nfo_exists:
|
||||
logger.debug(
|
||||
f"NFO exists for '{serie_name}', skipping download"
|
||||
)
|
||||
|
||||
except TMDBAPIError as e:
|
||||
# Only log at ERROR if no NFO exists and we have no IDs
|
||||
# If NFO exists with IDs, this is just a lookup failure, log at DEBUG
|
||||
if nfo_exists and (ids.get("tmdb_id") or ids.get("tvdb_id")):
|
||||
logger.debug(
|
||||
"TMDB API lookup failed for '%s' (has NFO with IDs): %s",
|
||||
serie_name, e
|
||||
)
|
||||
else:
|
||||
logger.error("TMDB API error processing '%s': %s", serie_name, e)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Unexpected error processing NFO for '{serie_name}': {e}",
|
||||
exc_info=True
|
||||
)
|
||||
|
||||
async def scan_and_process_nfo(self):
|
||||
"""Scan all series and process NFO files based on configuration.
|
||||
|
||||
This method:
|
||||
1. Loads series from database (avoiding filesystem scan)
|
||||
2. For each series with existing NFO, reads TMDB/TVDB IDs
|
||||
and updates database
|
||||
3. For each series without NFO (if auto_create=True), creates one
|
||||
4. For each series with NFO (if update_on_scan=True), updates it
|
||||
5. Runs operations concurrently for better performance
|
||||
"""
|
||||
if not self.nfo_service:
|
||||
logger.info("NFO service not enabled, skipping NFO processing")
|
||||
return
|
||||
|
||||
# Import database dependencies
|
||||
from src.server.database.connection import get_db_session
|
||||
from src.server.database.service import AnimeSeriesService
|
||||
|
||||
# Load series from database (not from filesystem)
|
||||
async with get_db_session() as db:
|
||||
anime_series_list = await AnimeSeriesService.get_all(
|
||||
db, with_episodes=False
|
||||
)
|
||||
|
||||
if not anime_series_list:
|
||||
logger.info("No series found in database to process")
|
||||
return
|
||||
|
||||
logger.info("Processing NFO for %s series...", len(anime_series_list))
|
||||
|
||||
# Create tasks for concurrent processing
|
||||
# Each task creates its own database session
|
||||
tasks = []
|
||||
for anime_series in anime_series_list:
|
||||
# Extract year if available
|
||||
year = getattr(anime_series, 'year', None)
|
||||
|
||||
task = self.process_nfo_for_series(
|
||||
serie_folder=anime_series.folder,
|
||||
serie_name=anime_series.name,
|
||||
serie_key=anime_series.key,
|
||||
year=year
|
||||
)
|
||||
tasks.append(task)
|
||||
|
||||
# Process in batches to avoid overwhelming TMDB API
|
||||
batch_size = 5
|
||||
for i in range(0, len(tasks), batch_size):
|
||||
batch = tasks[i:i + batch_size]
|
||||
await asyncio.gather(*batch, return_exceptions=True)
|
||||
|
||||
# Small delay between batches to respect rate limits
|
||||
if i + batch_size < len(tasks):
|
||||
await asyncio.sleep(2)
|
||||
|
||||
async def close(self):
|
||||
"""Clean up resources."""
|
||||
if self.nfo_service:
|
||||
await self.nfo_service.close()
|
||||
Reference in New Issue
Block a user