""" TV Series Processor for NFOGuard Handles TV series processing and episode management """ import os import glob import re from pathlib import Path from typing import Optional, Dict, List, Set, Tuple, Any from datetime import datetime from core.database import NFOGuardDatabase from core.nfo_manager import NFOManager from core.path_mapper import PathMapper from clients.sonarr_client import SonarrClient from clients.external_clients import ExternalClientManager from config.settings import config from utils.logging import _log 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) # Skip season.nfo and tvshow.nfo creation - focus only on episode NFOs pass _log("INFO", f"Completed processing TV series: {series_path.name}") def _extract_series_title_from_path(self, series_path: Path) -> Optional[str]: """Extract series title from directory path, removing year and IMDb ID""" name = series_path.name # Remove IMDb ID part: [imdb-ttXXXXXX] or [ttXXXXXX] name = re.sub(r'\s*\[imdb-?tt\d+\]', '', name, flags=re.IGNORECASE) name = re.sub(r'\s*\[tt\d+\]', '', name, flags=re.IGNORECASE) # Remove year in parentheses: (YYYY) name = re.sub(r'\s*\(\d{4}\)', '', name) # Clean up extra spaces name = ' '.join(name.split()) return name.strip() if name.strip() else None def _find_disk_episodes(self, series_path: Path) -> Dict[Tuple[int, int], List[Path]]: """Find all episodes on disk and return mapping of (season, episode) -> [video_files]""" episodes = {} if not series_path.exists(): return episodes video_extensions = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.ts'} # Define regex pattern for episode files episode_pattern = re.compile( r'.*[sS](\d{1,2})[eE](\d{1,3}).*|.*(\d{1,2})x(\d{1,3}).*' ) for item in series_path.rglob('*'): if item.is_file() and item.suffix.lower() in video_extensions: # Try to extract season/episode from filename match = episode_pattern.match(item.name) if match: if match.group(1) and match.group(2): # SxxExx format season = int(match.group(1)) episode = int(match.group(2)) elif match.group(3) and match.group(4): # NxNN format season = int(match.group(3)) episode = int(match.group(4)) else: continue key = (season, episode) if key not in episodes: episodes[key] = [] episodes[key].append(item) return episodes def _gather_episode_dates(self, series_path: Path, imdb_id: str, disk_episodes: Dict[Tuple[int, int], List[Path]]) -> Dict[Tuple[int, int], Tuple[Optional[str], Optional[str], str]]: """Gather episode air dates and date added information""" episode_dates = {} # Get data from Sonarr first if available sonarr_episodes = self._get_sonarr_episodes(imdb_id) # Get data from external sources for missing information for (season, episode) in disk_episodes: aired = None dateadded = None source = "unknown" # Try Sonarr first if (season, episode) in sonarr_episodes: sonarr_data = sonarr_episodes[(season, episode)] aired = sonarr_data.get('airDate') dateadded = sonarr_data.get('dateAdded') if dateadded: source = "sonarr" # Fallback to external sources if needed if not aired: external_aired = self.external_clients.get_episode_air_date(imdb_id, season, episode) if external_aired: aired = external_aired if not dateadded: source = "external" # Use air date as fallback for dateadded if available if not dateadded and aired: dateadded = aired source = f"{source}_fallback" if source != "unknown" else "aired_fallback" episode_dates[(season, episode)] = (aired, dateadded, source) return episode_dates def _get_sonarr_episodes(self, imdb_id: str) -> Dict[Tuple[int, int], Dict[str, Any]]: """Get episode information from Sonarr""" try: series_data = self.sonarr.get_series_by_imdb(imdb_id) if not series_data: return {} series_id = series_data.get('id') if not series_id: return {} episodes = self.sonarr.get_episodes(series_id) episode_map = {} for episode in episodes: season = episode.get('seasonNumber', 0) episode_num = episode.get('episodeNumber', 0) if season > 0 and episode_num > 0: episode_map[(season, episode_num)] = { 'airDate': episode.get('airDate'), 'dateAdded': episode.get('episodeFile', {}).get('dateAdded') if episode.get('hasFile') else None } return episode_map except Exception as e: _log("ERROR", f"Failed to get Sonarr episodes for {imdb_id}: {e}") return {} def process_season(self, series_path: str, season_name: str) -> Dict[str, Any]: """Process a specific season""" series_path_obj = Path(series_path) if not series_path_obj.exists(): raise FileNotFoundError(f"Series path not found: {series_path}") season_path = series_path_obj / season_name if not season_path.exists(): raise FileNotFoundError(f"Season directory not found: {season_path}") # Extract season number from directory name season_match = re.search(r'(\d+)', season_name) if not season_match: raise ValueError(f"Could not extract season number from: {season_name}") season_num = int(season_match.group(1)) # Get series IMDb ID imdb_id = self.nfo_manager.parse_imdb_from_path(series_path_obj) if not imdb_id: raise ValueError(f"No IMDb ID found in series path: {series_path}") _log("INFO", f"Processing season {season_num} of series: {series_path_obj.name}") # Find episodes in this season disk_episodes = self._find_disk_episodes(series_path_obj) season_episodes = {k: v for k, v in disk_episodes.items() if k[0] == season_num} if not season_episodes: return {"status": "no_episodes", "season": season_num, "episodes_found": 0} # Get episode dates episode_dates = self._gather_episode_dates(series_path_obj, imdb_id, season_episodes) # Process episodes processed_count = 0 for (season, episode), (aired, dateadded, source) in episode_dates.items(): if (season, episode) in season_episodes: # Create NFO if config.manage_nfo: self.nfo_manager.create_episode_nfo( season_path, season, episode, aired, dateadded, source, config.lock_metadata ) # Update file mtimes if config.fix_dir_mtimes and dateadded: video_files = season_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) processed_count += 1 _log("INFO", f"Processed {processed_count} episodes in season {season_num}") return { "status": "success", "season": season_num, "episodes_found": len(season_episodes), "episodes_processed": processed_count } def process_episode(self, series_path: str, season_name: str, episode_name: str) -> Dict[str, Any]: """Process a specific episode""" series_path_obj = Path(series_path) if not series_path_obj.exists(): raise FileNotFoundError(f"Series path not found: {series_path}") season_path = series_path_obj / season_name if not season_path.exists(): raise FileNotFoundError(f"Season directory not found: {season_path}") episode_path = season_path / episode_name if not episode_path.exists(): raise FileNotFoundError(f"Episode file not found: {episode_path}") # Extract season and episode numbers season_match = re.search(r'(\d+)', season_name) episode_match = re.search(r'[sS](\d+)[eE](\d+)|(\d+)x(\d+)', episode_name) if not season_match: raise ValueError(f"Could not extract season number from: {season_name}") if not episode_match: raise ValueError(f"Could not extract episode number from: {episode_name}") season_num = int(season_match.group(1)) if episode_match.group(1) and episode_match.group(2): # SxxExx format episode_num = int(episode_match.group(2)) elif episode_match.group(3) and episode_match.group(4): # NxNN format episode_num = int(episode_match.group(4)) else: raise ValueError(f"Could not parse episode number from: {episode_name}") # Get series IMDb ID imdb_id = self.nfo_manager.parse_imdb_from_path(series_path_obj) if not imdb_id: raise ValueError(f"No IMDb ID found in series path: {series_path}") _log("INFO", f"Processing episode S{season_num:02d}E{episode_num:02d} of series: {series_path_obj.name}") # Get episode data disk_episodes = {(season_num, episode_num): [episode_path]} episode_dates = self._gather_episode_dates(series_path_obj, imdb_id, disk_episodes) if (season_num, episode_num) not in episode_dates: return {"status": "no_data", "season": season_num, "episode": episode_num} aired, dateadded, source = episode_dates[(season_num, episode_num)] # Create NFO if config.manage_nfo: self.nfo_manager.create_episode_nfo( season_path, season_num, episode_num, aired, dateadded, source, config.lock_metadata ) # Update file mtime if config.fix_dir_mtimes and dateadded: self.nfo_manager.set_file_mtime(episode_path, dateadded) # Save to database self.db.upsert_episode_date(imdb_id, season_num, episode_num, aired, dateadded, source, True) _log("INFO", f"Processed episode S{season_num:02d}E{episode_num:02d}") return { "status": "success", "season": season_num, "episode": episode_num, "aired": aired, "dateadded": dateadded, "source": source }