"""Image loading service for downloading series artwork from TMDB. This service downloads poster.jpg, fanart.jpg, and logo.png images for anime series using TMDB as the image source. Integrated with: - BackgroundLoaderService: triggered when adding new anime - SchedulerService: triggered during scheduled rescan """ from __future__ import annotations import asyncio from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import structlog from sqlalchemy.ext.asyncio import AsyncSession from src.server.database.service import AnimeSeriesService from src.server.nfo.tmdb_client import TMDBClient from src.server.utils.image_downloader import ImageDownloader from src.server.utils.media import FANART_FILENAME, LOGO_FILENAME, POSTER_FILENAME logger = structlog.get_logger(__name__) class ImageLoadingServiceError(Exception): """Exception raised for image loading failures.""" class ImageLoadingService: """Service for loading series images from TMDB. Downloads poster.jpg, fanart.jpg, and logo.png for anime series using TMDB as the image source. Images are saved to the series folder alongside tvshow.nfo. Attributes: tmdb_client: TMDB API client for fetching image URLs image_downloader: Downloader for saving images to disk """ # Batch size for scheduler bulk processing BATCH_SIZE = 10 def __init__(self, tmdb_client: TMDBClient): """Initialize the image loading service. Args: tmdb_client: TMDB API client for fetching image metadata """ self._tmdb_client = tmdb_client async def load_series_images( self, key: str, folder: str, anime_directory: str, db: AsyncSession, ) -> Dict[str, bool]: """Load all images for a single series. Downloads poster.jpg, fanart.jpg, and logo.png from TMDB if they don't already exist in the series folder. Args: key: Series unique identifier (provider key) folder: Series folder name (metadata, for path construction) anime_directory: Base anime directory path db: Database session Returns: Dict with download status for each image type: { "poster": bool, # True if poster.jpg exists/downloads succeeded "fanart": bool, # True if fanart.jpg exists/downloads succeeded "logo": bool # True if logo.png exists/downloads succeeded } """ series_dir = Path(anime_directory) / folder if not series_dir.exists(): logger.warning( "Series directory not found, cannot load images", key=key, folder=folder, path=str(series_dir), ) return {"poster": False, "fanart": False, "logo": False} # Get series from database to retrieve TMDB ID series = await AnimeSeriesService.get_by_key(db, key) if not series: logger.warning("Series not found in database key=%s", key) return {"poster": False, "fanart": False, "logo": False} if not series.tmdb_id: logger.warning( "Series has no TMDB ID, cannot load images key=%s name=%s", key, series.name, ) return {"poster": False, "fanart": False, "logo": False} try: # Fetch image metadata from TMDB images_data = await self._tmdb_client.get_tv_show_images(series.tmdb_id) poster_url, logo_url, fanart_url = self._select_best_images(images_data) # Download images results = await self._download_images( series_dir, poster_url, logo_url, fanart_url ) # Update database flags await self._update_series_flags(db, series, results) return results except Exception as e: logger.exception("Failed to load images for series: %s", key, exc_info=e) return {"poster": False, "fanart": False, "logo": False} async def load_series_images_batch( self, series_list: List[Dict[str, Any]], anime_directory: str, db: AsyncSession, ) -> Dict[str, int]: """Load images for multiple series in batches. Used by scheduler to process many series efficiently. Args: series_list: List of dicts with 'key' and 'folder' for each series anime_directory: Base anime directory path db: Database session Returns: Dict with counts: { "total": int, # Total series processed "success": int, # Series with all images loaded "partial": int, # Series with some images loaded "failed": int, # Series with no images loaded "poster": int, # Count of poster.jpg downloads "fanart": int, # Count of fanart.jpg downloads "logo": int, # Count of logo.png downloads } """ stats = { "total": len(series_list), "success": 0, "partial": 0, "failed": 0, "poster": 0, "fanart": 0, "logo": 0, } # Process in batches to respect TMDB rate limits for i in range(0, len(series_list), self.BATCH_SIZE): batch = series_list[i : i + self.BATCH_SIZE] # Process each series sequentially to avoid concurrent use of the # same AsyncSession (SQLAlchemy async sessions are not thread-safe # for concurrent operations). BATCH_SIZE still paces TMDB requests. results: List[Dict[str, Any] | Exception] = [] for series in batch: result = await self.load_series_images( key=series["key"], folder=series["folder"], anime_directory=anime_directory, db=db, ) results.append(result) for series, result in zip(batch, results): if isinstance(result, Exception): logger.warning( "Image loading failed for series: %s", series["key"], exc_info=result, ) stats["failed"] += 1 continue if result["poster"] and result["fanart"] and result["logo"]: stats["success"] += 1 elif result["poster"] or result["fanart"] or result["logo"]: stats["partial"] += 1 else: stats["failed"] += 1 if result["poster"]: stats["poster"] += 1 if result["fanart"]: stats["fanart"] += 1 if result["logo"]: stats["logo"] += 1 # Small delay between batches to avoid overwhelming TMDB if i + self.BATCH_SIZE < len(series_list): await asyncio.sleep(0.5) logger.info( "Batch image loading completed", total=stats["total"], success=stats["success"], partial=stats["partial"], failed=stats["failed"], ) return stats def _select_best_images( self, images_data: Dict[str, Any] ) -> Tuple[Optional[str], Optional[str], Optional[str]]: """Select the best available images from TMDB data. Args: images_data: TMDB images response with 'posters', 'backdrops', 'logos' Returns: Tuple of (poster_url, logo_url, fanart_url) - URLs or None if not available """ poster_url = None logo_url = None fanart_url = None # Select poster: prefer English, otherwise take first available posters = images_data.get("posters", []) for poster in posters: if poster.get("iso_639_1") == "en" or poster.get("iso_639_1") is None: poster_url = self._tmdb_client.get_image_url(poster["file_path"]) break if not poster_url and posters: poster_url = self._tmdb_client.get_image_url(posters[0]["file_path"]) # Select logo/clearlogo: prefer English with transparent background logos = images_data.get("logos", []) for logo in logos: if logo.get("iso_639_1") == "en": logo_url = self._tmdb_client.get_image_url(logo["file_path"]) break if not logo_url and logos: logo_url = self._tmdb_client.get_image_url(logos[0]["file_path"]) # Select fanart/backdrop: prefer English backdrops = images_data.get("backdrops", []) for backdrop in backdrops: if backdrop.get("iso_639_1") == "en": fanart_url = self._tmdb_client.get_image_url(backdrop["file_path"]) break if not fanart_url and backdrops: fanart_url = self._tmdb_client.get_image_url(backdrops[0]["file_path"]) return poster_url, logo_url, fanart_url async def _download_images( self, series_dir: Path, poster_url: Optional[str], logo_url: Optional[str], fanart_url: Optional[str], ) -> Dict[str, bool]: """Download images to series directory. Args: series_dir: Path to series folder poster_url: URL for poster.jpg logo_url: URL for logo.png fanart_url: URL for fanart.jpg Returns: Dict with download status for each image """ results = {"poster": False, "fanart": False, "logo": False} async with ImageDownloader() as downloader: tasks = [] if poster_url: tasks.append( self._download_and_track( downloader, poster_url, series_dir / POSTER_FILENAME, "poster", results ) ) if logo_url: tasks.append( self._download_and_track( downloader, logo_url, series_dir / LOGO_FILENAME, "logo", results ) ) if fanart_url: tasks.append( self._download_and_track( downloader, fanart_url, series_dir / FANART_FILENAME, "fanart", results ) ) if tasks: await asyncio.gather(*tasks, return_exceptions=True) return results async def _download_and_track( self, downloader: ImageDownloader, url: str, local_path: Path, key: str, results: Dict[str, bool], ) -> None: """Download single image and track result. Args: downloader: ImageDownloader instance url: Image URL local_path: Local destination path key: Result dict key ('poster', 'logo', 'fanart') results: Dict to update with result """ try: success = await downloader.download_image( url, local_path, skip_existing=True, validate=True ) results[key] = success except Exception as e: logger.warning("Failed to download %s: %s", key, e) results[key] = False async def _update_series_flags( self, db: AsyncSession, series: Any, results: Dict[str, bool], ) -> None: """Update database flags after image loading. Args: db: Database session series: AnimeSeries instance results: Dict with download status for each image """ try: series.images_loaded = results["poster"] and results["fanart"] series.logo_loaded = results["logo"] await db.commit() except Exception as e: logger.warning("Failed to update series image flags: %s", e) await db.rollback() # Singleton instance _image_loading_service: Optional[ImageLoadingService] = None def get_image_loading_service() -> ImageLoadingService: """Get the image loading service singleton. Returns: ImageLoadingService instance Raises: RuntimeError: If service not initialized """ if _image_loading_service is None: raise RuntimeError( "ImageLoadingService not initialized. " "Call init_image_loading_service() first." ) return _image_loading_service def init_image_loading_service(tmdb_client: TMDBClient) -> ImageLoadingService: """Initialize the image loading service singleton. Args: tmdb_client: TMDB API client Returns: ImageLoadingService instance """ global _image_loading_service _image_loading_service = ImageLoadingService(tmdb_client=tmdb_client) return _image_loading_service