#!/usr/bin/env python3 """ TV Show processing logic for NFOGuard """ import os import glob import re from pathlib import Path from typing import Optional, Dict, Any, List, Tuple from datetime import datetime, timezone, timedelta # Import core components from core.database import NFOGuardDatabase from core.nfo_manager import NFOManager from core.path_mapper import PathMapper from core.logging import _log from core.fs_cache import fs_cache, parse_episode_from_filename, extract_imdb_from_path from clients.sonarr_client import SonarrClient from clients.external_clients import ExternalClientManager from config.settings import config class TVProcessor: """Handles TV series processing""" def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager, path_mapper: PathMapper): self.db = db self.nfo_manager = nfo_manager self.path_mapper = path_mapper self.sonarr = SonarrClient( os.environ.get("SONARR_URL", ""), os.environ.get("SONARR_API_KEY", "") ) self.external_clients = ExternalClientManager() def find_series_path(self, series_title: str, imdb_id: str, sonarr_path: str = None) -> Optional[Path]: """Find series directory path""" # Try webhook path first if sonarr_path: container_path = self.path_mapper.sonarr_path_to_container_path(sonarr_path) path_obj = Path(container_path) if path_obj.exists(): return path_obj # Search by IMDb ID or title for media_path in config.tv_paths: if not media_path.exists(): continue # Search by IMDb ID if imdb_id: pattern = str(media_path / f"*[imdb-{imdb_id}]*") matches = glob.glob(pattern) if matches: return Path(matches[0]) # Search by title if series_title: title_clean = series_title.lower().replace(" ", "").replace("-", "") for item in media_path.iterdir(): if item.is_dir() and "[imdb-" in item.name.lower(): item_clean = item.name.lower().replace(" ", "").replace("-", "") if title_clean in item_clean: return item return None def process_series(self, series_path: Path) -> None: """Process a TV series directory""" imdb_id = self.nfo_manager.parse_imdb_from_path(series_path) if not imdb_id: _log("ERROR", f"No IMDb ID found in series path: {series_path}") return _log("INFO", f"Processing TV series: {series_path.name}") # Update database self.db.upsert_series(imdb_id, str(series_path)) # Find video files disk_episodes = self._find_disk_episodes(series_path) _log("INFO", f"Found {len(disk_episodes)} episodes on disk") # Get episode dates episode_dates = self._gather_episode_dates(series_path, imdb_id, disk_episodes) # Process episodes for (season, episode), (aired, dateadded, source) in episode_dates.items(): if (season, episode) in disk_episodes: # Create NFO if config.manage_nfo: season_dir = series_path / config.tv_season_dir_format.format(season=season) self.nfo_manager.create_episode_nfo( season_dir, season, episode, aired, dateadded, source, config.lock_metadata ) # Update file mtimes if config.fix_dir_mtimes and dateadded: video_files = disk_episodes[(season, episode)] for video_file in video_files: self.nfo_manager.set_file_mtime(video_file, dateadded) # Save to database self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True) # Create season/tvshow NFOs if config.manage_nfo: seasons_processed = set() for (season, episode) in disk_episodes.keys(): if season not in seasons_processed: season_dir = series_path / config.tv_season_dir_format.format(season=season) self.nfo_manager.create_season_nfo(season_dir, season) seasons_processed.add(season) # Get TVDB ID for better Emby compatibility tvdb_id = self.external_clients.get_tvdb_series_id(imdb_id) self.nfo_manager.create_tvshow_nfo(series_path, imdb_id, tvdb_id) _log("INFO", f"Completed processing TV series: {series_path.name}") def _find_disk_episodes(self, series_path: Path) -> Dict[Tuple[int, int], List[Path]]: """Find all episode video files on disk""" disk_episodes = {} video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v") # Use cached directory listing season_dirs = fs_cache.get_directory_contents(series_path) for season_dir in season_dirs: if not (season_dir.is_dir() and season_dir.name.lower().startswith(config.tv_season_dir_pattern)): continue try: # Extract season number from directory name # Handle formats like "Season 01", "S01", "Season01", etc. dir_name = season_dir.name.lower() if config.tv_season_dir_pattern in dir_name: # Extract everything after the pattern season_part = dir_name[len(config.tv_season_dir_pattern):].strip() else: continue season_num = int(season_part) except (ValueError, IndexError): continue # Use cached video file discovery video_files = fs_cache.find_video_files(season_dir) for video_file in video_files: # Use cached episode parsing episode_info = parse_episode_from_filename(video_file.name) if episode_info: file_season, file_episode = episode_info key = (season_num, file_episode) # Use directory season number if key not in disk_episodes: disk_episodes[key] = [] disk_episodes[key].append(video_file) return disk_episodes def _gather_episode_dates(self, series_path: Path, imdb_id: str, disk_episodes: Dict) -> Dict: """Gather episode dates from various sources""" episode_dates = {} # Check cache first cached_episodes = self.db.get_series_episodes(imdb_id, has_video_file_only=True) for ep in cached_episodes: key = (ep["season"], ep["episode"]) if key in disk_episodes: # Only use cached data for episodes we have episode_dates[key] = (ep["aired"], ep["dateadded"], ep["source"]) # Check for existing NFO files (including long-named ones) for migration nfo_episodes_found = 0 for (season_num, episode_num) in disk_episodes.keys(): if (season_num, episode_num) not in episode_dates: # Check if this episode has an existing NFO file that needs migration season_dir = series_path / config.tv_season_dir_format.format(season=season_num) existing_nfo = self.nfo_manager.find_existing_episode_nfo(season_dir, season_num, episode_num) if existing_nfo: # Force processing of this episode for NFO migration episode_dates[(season_num, episode_num)] = (None, None, "nfo_migration_required") nfo_episodes_found += 1 if nfo_episodes_found > 0: _log("INFO", f"Found {nfo_episodes_found} episodes with existing NFO files requiring migration") # Find missing episodes (not in cache and no existing NFO) cached_keys = set(episode_dates.keys()) missing_keys = set(disk_episodes.keys()) - cached_keys if not missing_keys: if nfo_episodes_found == 0: _log("INFO", "All episodes found in cache") return episode_dates _log("INFO", f"Querying APIs for {len(missing_keys)} missing episodes") # Query Sonarr for missing episodes if self.sonarr.enabled: series = self.sonarr.series_by_imdb(imdb_id) if series: series_id = series.get("id") if series_id: episodes = self.sonarr.episodes_for_series(series_id) for ep in episodes: season_num = ep.get("seasonNumber") episode_num = ep.get("episodeNumber") if not isinstance(season_num, int) or not isinstance(episode_num, int): continue key = (season_num, episode_num) if key not in missing_keys: continue # Get dates aired = self._parse_date_to_iso(ep.get("airDateUtc")) dateadded = None source = "sonarr:episode.airDateUtc" # Try to get import history episode_id = ep.get("id") if episode_id: import_date = self.sonarr.get_episode_import_history(episode_id) if import_date: dateadded = self._parse_date_to_iso(import_date) source = "sonarr:history.import" if not dateadded: dateadded = aired if aired or dateadded: episode_dates[key] = (aired, dateadded, source) # Fill remaining gaps with external APIs remaining_keys = missing_keys - set(episode_dates.keys()) if remaining_keys and self.external_clients.tmdb.enabled: tmdb_movie = self.external_clients.tmdb.find_by_imdb(imdb_id) if tmdb_movie: tv_id = tmdb_movie.get("id") if tv_id: seasons_needed = set(season for season, episode in remaining_keys) for season_num in seasons_needed: tmdb_episodes = self.external_clients.tmdb.get_tv_season_episodes(tv_id, season_num) for ep_num, air_date in tmdb_episodes.items(): key = (season_num, ep_num) if key in remaining_keys: aired = self._parse_date_to_iso(air_date) episode_dates[key] = (aired, aired, "tmdb:air_date") return episode_dates def _parse_date_to_iso(self, date_str: str) -> Optional[str]: """Parse date string to ISO format""" if not date_str: return None try: if len(date_str) == 10 and date_str[4] == "-": dt = datetime.fromisoformat(date_str).replace(tzinfo=timezone.utc) else: dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")).astimezone(timezone.utc) return dt.isoformat(timespec="seconds") except Exception: return None def process_webhook_episodes(self, series_path: Path, episodes_data: List[Dict]) -> None: """Process specific episodes from webhook data""" imdb_id = self.nfo_manager.parse_imdb_from_path(series_path) if not imdb_id: _log("ERROR", f"No IMDb ID found in series path: {series_path}") return _log("INFO", f"Processing {len(episodes_data)} episodes from webhook for series: {series_path.name}") # Update database self.db.upsert_series(imdb_id, str(series_path)) # Get enhanced metadata from Sonarr series_metadata = self._get_sonarr_series_metadata(imdb_id) for episode_data in episodes_data: season_num = episode_data.get("seasonNumber") episode_num = episode_data.get("episodeNumber") if not isinstance(season_num, int) or not isinstance(episode_num, int): continue _log("INFO", f"Processing webhook episode S{season_num:02d}E{episode_num:02d}") # Find episode file season_dir = series_path / config.tv_season_dir_format.format(season=season_num) if not season_dir.exists(): _log("WARNING", f"Season directory not found: {season_dir}") continue # Find video file for this episode episode_file = self._find_episode_video_file(season_dir, season_num, episode_num) if not episode_file: _log("WARNING", f"Video file not found for S{season_num:02d}E{episode_num:02d}") continue # Process this specific episode self.process_episode_file(series_path, season_dir, episode_file) def _find_episode_video_file(self, season_dir: Path, season_num: int, episode_num: int) -> Optional[Path]: """Find video file for specific episode""" video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v") episode_pattern = f"S{season_num:02d}E{episode_num:02d}" for video_file in season_dir.iterdir(): if (video_file.is_file() and video_file.suffix.lower() in video_exts and episode_pattern.upper() in video_file.name.upper()): return video_file return None def _get_sonarr_series_metadata(self, imdb_id: str) -> Optional[Dict]: """Get series metadata from Sonarr""" if not self.sonarr.enabled: return None series = self.sonarr.series_by_imdb(imdb_id) if not series: return None series_id = series.get("id") if not series_id: return None # Get episodes for this series episodes = self.sonarr.episodes_for_series(series_id) return { "series": series, "episodes": {(ep.get("seasonNumber"), ep.get("episodeNumber")): ep for ep in episodes} } def _get_episode_metadata(self, series_metadata: Optional[Dict], season_num: int, episode_num: int, season_path: Path) -> Dict: """Get enhanced episode metadata""" metadata = {} # Try to get title from Sonarr first if series_metadata and "episodes" in series_metadata: episode_data = series_metadata["episodes"].get((season_num, episode_num)) if episode_data and episode_data.get("title"): metadata["title"] = episode_data["title"] # Fallback to extracting title from filename if "title" not in metadata: filename_title = self._extract_title_from_filename(season_path, season_num, episode_num) if filename_title: metadata["title"] = filename_title return metadata def _extract_title_from_filename(self, season_path: Path, season_num: int, episode_num: int) -> Optional[str]: """Extract episode title from video filename""" episode_file = self._find_episode_video_file(season_path, season_num, episode_num) if not episode_file: return None filename = episode_file.stem # Pattern: Series.S01E01.Title.Here.RESOLUTION.etc # Look for everything after S##E## until common quality indicators episode_pattern = f"S{season_num:02d}E{episode_num:02d}" # Find the episode marker episode_index = filename.upper().find(episode_pattern.upper()) if episode_index == -1: return None # Extract everything after the episode marker after_episode = filename[episode_index + len(episode_pattern):] # Split by common separators parts = re.split(r'[.\-_\s]+', after_episode) # Stop at common quality/release indicators stop_words = { '720p', '1080p', '4k', '2160p', 'hdtv', 'webrip', 'bluray', 'dvdrip', 'web-dl', 'brrip', 'hdcam', 'hdts', 'cam', 'ts', 'tc', 'r5', 'dvdscr', 'x264', 'x265', 'h264', 'h265', 'xvid', 'divx', 'aac', 'mp3', 'ac3', 'dts', 'flac', 'atmos', 'truehd', 'dd', 'ddp', 'eac3' } title_parts = [] for part in parts: if not part: continue if part.lower() in stop_words: break # Stop at year patterns (e.g., 2023) if re.match(r'^\d{4}$', part): break title_parts.append(part) if title_parts: return ' '.join(title_parts) return None def process_episode_file(self, series_path: Path, season_path: Path, episode_file: Path) -> None: """Process a single TV episode file""" imdb_id = self.nfo_manager.parse_imdb_from_path(series_path) if not imdb_id: _log("ERROR", f"No IMDb ID found in series path: {series_path}") return # Parse episode info from filename episode_info = self._parse_episode_from_filename(episode_file.name) if not episode_info: _log("ERROR", f"Could not parse episode info from: {episode_file.name}") return season_num, episode_num = episode_info _log("INFO", f"Processing single episode S{season_num:02d}E{episode_num:02d}: {episode_file.name}") # Update database self.db.upsert_series(imdb_id, str(series_path)) # Get enhanced metadata from Sonarr series_metadata = self._get_sonarr_series_metadata(imdb_id) enhanced_metadata = self._get_episode_metadata(series_metadata, season_num, episode_num, season_path) if series_metadata else self._get_episode_metadata(None, season_num, episode_num, season_path) # Get episode date aired, dateadded, source = self._get_single_episode_date(imdb_id, season_num, episode_num, series_metadata) # Create NFO if config.manage_nfo and dateadded: self.nfo_manager.create_episode_nfo( season_path, season_num, episode_num, aired, dateadded, source, config.lock_metadata, enhanced_metadata ) # Update file mtime if config.fix_dir_mtimes and dateadded: self.nfo_manager.set_file_mtime(episode_file, dateadded) # Save to database self.db.upsert_episode_date(imdb_id, season_num, episode_num, aired, dateadded, source, True) _log("INFO", f"Completed processing episode S{season_num:02d}E{episode_num:02d}") def _parse_episode_from_filename(self, filename: str) -> Optional[Tuple[int, int]]: """Parse season and episode numbers from filename""" match = re.search(r"S(\d{1,2})E(\d{1,2})", filename, re.IGNORECASE) if match: return int(match.group(1)), int(match.group(2)) return None def _get_single_episode_date(self, imdb_id: str, season_num: int, episode_num: int, series_metadata: Optional[Dict]) -> Tuple[Optional[str], Optional[str], str]: """Get dates for a single episode""" # Check database first existing_episode = self.db.get_episode_date(imdb_id, season_num, episode_num) if existing_episode and existing_episode.get("dateadded"): return existing_episode.get("aired"), existing_episode["dateadded"], existing_episode["source"] # Try Sonarr metadata if series_metadata and "episodes" in series_metadata: episode_data = series_metadata["episodes"].get((season_num, episode_num)) if episode_data: aired = self._parse_date_to_iso(episode_data.get("airDateUtc")) dateadded = aired # Use air date as fallback source = "sonarr:episode.airDateUtc" # Try to get import history episode_id = episode_data.get("id") if episode_id and self.sonarr.enabled: import_date = self.sonarr.get_episode_import_history(episode_id) if import_date: dateadded = self._parse_date_to_iso(import_date) source = "sonarr:history.import" return aired, dateadded, source # Fallback to external APIs if self.external_clients.tmdb.enabled: tmdb_movie = self.external_clients.tmdb.find_by_imdb(imdb_id) if tmdb_movie: tv_id = tmdb_movie.get("id") if tv_id: tmdb_episodes = self.external_clients.tmdb.get_tv_season_episodes(tv_id, season_num) air_date = tmdb_episodes.get(episode_num) if air_date: aired = self._parse_date_to_iso(air_date) return aired, aired, "tmdb:air_date" return None, None, "no_valid_date_source"