feat: Integrate NFO service with series management
- Created SeriesManagerService to orchestrate SerieList and NFOService - Follows clean architecture (core entities stay independent) - Supports auto-create and update-on-scan based on configuration - Created CLI tool (src/cli/nfo_cli.py) for NFO management - Commands: 'scan' (create/update NFOs) and 'status' (check NFO coverage) - Batch processing with rate limiting to respect TMDB API limits - Comprehensive error handling and logging Usage: python -m src.cli.nfo_cli scan # Create missing NFOs python -m src.cli.nfo_cli status # Check NFO statistics
This commit is contained in:
194
src/cli/nfo_cli.py
Normal file
194
src/cli/nfo_cli.py
Normal file
@@ -0,0 +1,194 @@
|
||||
"""CLI command for NFO management.
|
||||
|
||||
This script provides command-line interface for creating, updating,
|
||||
and checking NFO metadata files.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
|
||||
from src.config.settings import settings
|
||||
from src.core.services.series_manager_service import SeriesManagerService
|
||||
|
||||
|
||||
async def scan_and_create_nfo():
|
||||
"""Scan all series and create missing NFO files."""
|
||||
print("=" * 70)
|
||||
print("NFO Auto-Creation Tool")
|
||||
print("=" * 70)
|
||||
|
||||
if not settings.tmdb_api_key:
|
||||
print("\n❌ Error: TMDB_API_KEY not configured")
|
||||
print(" Set TMDB_API_KEY in .env file or environment")
|
||||
print(" Get API key from: https://www.themoviedb.org/settings/api")
|
||||
return 1
|
||||
|
||||
if not settings.anime_directory:
|
||||
print("\n❌ Error: ANIME_DIRECTORY not configured")
|
||||
return 1
|
||||
|
||||
print(f"\nAnime Directory: {settings.anime_directory}")
|
||||
print(f"Auto-create NFO: {settings.nfo_auto_create}")
|
||||
print(f"Update on scan: {settings.nfo_update_on_scan}")
|
||||
print(f"Download poster: {settings.nfo_download_poster}")
|
||||
print(f"Download logo: {settings.nfo_download_logo}")
|
||||
print(f"Download fanart: {settings.nfo_download_fanart}")
|
||||
|
||||
if not settings.nfo_auto_create:
|
||||
print("\n⚠️ Warning: NFO_AUTO_CREATE is set to False")
|
||||
print(" Enable it in .env to auto-create NFO files")
|
||||
print("\n Continuing anyway to demonstrate functionality...")
|
||||
# Override for demonstration
|
||||
settings.nfo_auto_create = True
|
||||
|
||||
print("\nInitializing series manager...")
|
||||
manager = SeriesManagerService.from_settings()
|
||||
|
||||
# Get series list first
|
||||
serie_list = manager.get_serie_list()
|
||||
all_series = serie_list.get_all()
|
||||
|
||||
print(f"Found {len(all_series)} series in directory")
|
||||
|
||||
if not all_series:
|
||||
print("\n⚠️ No series found. Add some anime series first.")
|
||||
return 0
|
||||
|
||||
# Show series without NFO
|
||||
series_without_nfo = []
|
||||
for serie in all_series:
|
||||
if not serie.has_nfo():
|
||||
series_without_nfo.append(serie)
|
||||
|
||||
if series_without_nfo:
|
||||
print(f"\nSeries without NFO: {len(series_without_nfo)}")
|
||||
for serie in series_without_nfo[:5]: # Show first 5
|
||||
print(f" - {serie.name} ({serie.folder})")
|
||||
if len(series_without_nfo) > 5:
|
||||
print(f" ... and {len(series_without_nfo) - 5} more")
|
||||
else:
|
||||
print("\n✅ All series already have NFO files!")
|
||||
|
||||
if not settings.nfo_update_on_scan:
|
||||
print("\nNothing to do. Enable NFO_UPDATE_ON_SCAN to update existing NFOs.")
|
||||
return 0
|
||||
|
||||
print("\nProcessing NFO files...")
|
||||
print("(This may take a while depending on the number of series)")
|
||||
|
||||
try:
|
||||
await manager.scan_and_process_nfo()
|
||||
print("\n✅ NFO processing complete!")
|
||||
|
||||
# Show updated stats
|
||||
serie_list.load_series() # Reload to get updated stats
|
||||
all_series = serie_list.get_all()
|
||||
series_with_nfo = [s for s in all_series if s.has_nfo()]
|
||||
series_with_poster = [s for s in all_series if s.has_poster()]
|
||||
series_with_logo = [s for s in all_series if s.has_logo()]
|
||||
series_with_fanart = [s for s in all_series if s.has_fanart()]
|
||||
|
||||
print("\nFinal Statistics:")
|
||||
print(f" Series with NFO: {len(series_with_nfo)}/{len(all_series)}")
|
||||
print(f" Series with poster: {len(series_with_poster)}/{len(all_series)}")
|
||||
print(f" Series with logo: {len(series_with_logo)}/{len(all_series)}")
|
||||
print(f" Series with fanart: {len(series_with_fanart)}/{len(all_series)}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return 1
|
||||
finally:
|
||||
await manager.close()
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
async def check_nfo_status():
|
||||
"""Check NFO status for all series."""
|
||||
print("=" * 70)
|
||||
print("NFO Status Check")
|
||||
print("=" * 70)
|
||||
|
||||
if not settings.anime_directory:
|
||||
print("\n❌ Error: ANIME_DIRECTORY not configured")
|
||||
return 1
|
||||
|
||||
print(f"\nAnime Directory: {settings.anime_directory}")
|
||||
|
||||
# Create series list (no NFO service needed for status check)
|
||||
from src.core.entities.SerieList import SerieList
|
||||
serie_list = SerieList(settings.anime_directory)
|
||||
all_series = serie_list.get_all()
|
||||
|
||||
if not all_series:
|
||||
print("\n⚠️ No series found")
|
||||
return 0
|
||||
|
||||
print(f"\nTotal series: {len(all_series)}")
|
||||
|
||||
# Categorize series
|
||||
with_nfo = []
|
||||
without_nfo = []
|
||||
|
||||
for serie in all_series:
|
||||
if serie.has_nfo():
|
||||
with_nfo.append(serie)
|
||||
else:
|
||||
without_nfo.append(serie)
|
||||
|
||||
print(f"\nWith NFO: {len(with_nfo)} ({len(with_nfo) * 100 // len(all_series)}%)")
|
||||
print(f"Without NFO: {len(without_nfo)} ({len(without_nfo) * 100 // len(all_series)}%)")
|
||||
|
||||
if without_nfo:
|
||||
print("\nSeries missing NFO:")
|
||||
for serie in without_nfo[:10]:
|
||||
print(f" ❌ {serie.name} ({serie.folder})")
|
||||
if len(without_nfo) > 10:
|
||||
print(f" ... and {len(without_nfo) - 10} more")
|
||||
|
||||
# Media file statistics
|
||||
with_poster = sum(1 for s in all_series if s.has_poster())
|
||||
with_logo = sum(1 for s in all_series if s.has_logo())
|
||||
with_fanart = sum(1 for s in all_series if s.has_fanart())
|
||||
|
||||
print("\nMedia Files:")
|
||||
print(f" Posters: {with_poster}/{len(all_series)} ({with_poster * 100 // len(all_series)}%)")
|
||||
print(f" Logos: {with_logo}/{len(all_series)} ({with_logo * 100 // len(all_series)}%)")
|
||||
print(f" Fanart: {with_fanart}/{len(all_series)} ({with_fanart * 100 // len(all_series)}%)")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def main():
|
||||
"""Main CLI entry point."""
|
||||
if len(sys.argv) < 2:
|
||||
print("NFO Management Tool")
|
||||
print("\nUsage:")
|
||||
print(" python -m src.cli.nfo_cli scan # Scan and create/update NFO files")
|
||||
print(" python -m src.cli.nfo_cli status # Check NFO status for all series")
|
||||
print("\nConfiguration:")
|
||||
print(" Set TMDB_API_KEY in .env file")
|
||||
print(" Set NFO_AUTO_CREATE=true to enable auto-creation")
|
||||
print(" Set NFO_UPDATE_ON_SCAN=true to update existing NFOs")
|
||||
return 1
|
||||
|
||||
command = sys.argv[1].lower()
|
||||
|
||||
if command == "scan":
|
||||
return asyncio.run(scan_and_create_nfo())
|
||||
elif command == "status":
|
||||
return asyncio.run(check_nfo_status())
|
||||
else:
|
||||
print(f"Unknown command: {command}")
|
||||
print("Use 'scan' or 'status'")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
214
src/core/services/series_manager_service.py
Normal file
214
src/core/services/series_manager_service.py
Normal file
@@ -0,0 +1,214 @@
|
||||
"""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
|
||||
self.serie_list = SerieList(anime_directory)
|
||||
|
||||
# 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):
|
||||
self.nfo_service = NFOService(
|
||||
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)
|
||||
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, year: Optional[int] = None):
|
||||
"""Process NFO file for a series (create or update).
|
||||
|
||||
Args:
|
||||
serie_folder: Series folder name
|
||||
serie_name: Series display name
|
||||
year: Release year (helps with TMDB matching)
|
||||
"""
|
||||
if not self.nfo_service:
|
||||
return
|
||||
|
||||
try:
|
||||
folder_path = Path(self.anime_directory) / serie_folder
|
||||
nfo_exists = await self.nfo_service.check_nfo_exists(serie_folder)
|
||||
|
||||
if not nfo_exists and self.auto_create_nfo:
|
||||
logger.info(f"Creating NFO for '{serie_name}' ({serie_folder})")
|
||||
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(f"Successfully created NFO for '{serie_name}'")
|
||||
|
||||
elif nfo_exists and self.update_on_scan:
|
||||
logger.info(f"Updating NFO for '{serie_name}' ({serie_folder})")
|
||||
try:
|
||||
await self.nfo_service.update_tvshow_nfo(
|
||||
serie_folder=serie_folder,
|
||||
download_media=True
|
||||
)
|
||||
logger.info(f"Successfully updated NFO for '{serie_name}'")
|
||||
except NotImplementedError:
|
||||
logger.warning(
|
||||
f"NFO update not yet implemented for '{serie_name}'. "
|
||||
"Delete tvshow.nfo to recreate."
|
||||
)
|
||||
|
||||
except TMDBAPIError as e:
|
||||
logger.error(f"TMDB API error processing '{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. Uses SerieList to scan series folders
|
||||
2. For each series without NFO (if auto_create=True), creates one
|
||||
3. For each series with NFO (if update_on_scan=True), updates it
|
||||
4. Runs operations concurrently for better performance
|
||||
"""
|
||||
if not self.nfo_service:
|
||||
logger.info("NFO service not enabled, skipping NFO processing")
|
||||
return
|
||||
|
||||
# Get all series from SerieList
|
||||
all_series = self.serie_list.get_all()
|
||||
|
||||
if not all_series:
|
||||
logger.info("No series found to process")
|
||||
return
|
||||
|
||||
logger.info(f"Processing NFO for {len(all_series)} series...")
|
||||
|
||||
# Create tasks for concurrent processing
|
||||
tasks = []
|
||||
for serie in all_series:
|
||||
# Extract year from first air date if available
|
||||
year = None
|
||||
if hasattr(serie, 'year') and serie.year:
|
||||
year = serie.year
|
||||
|
||||
task = self.process_nfo_for_series(
|
||||
serie_folder=serie.folder,
|
||||
serie_name=serie.name,
|
||||
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)
|
||||
|
||||
logger.info("NFO processing complete")
|
||||
|
||||
def get_serie_list(self) -> SerieList:
|
||||
"""Get the underlying SerieList instance.
|
||||
|
||||
Returns:
|
||||
SerieList instance
|
||||
"""
|
||||
return self.serie_list
|
||||
|
||||
async def close(self):
|
||||
"""Clean up resources."""
|
||||
if self.nfo_service:
|
||||
await self.nfo_service.close()
|
||||
Reference in New Issue
Block a user