From 2e7922dc8bbefb241a2354969f5a94c0971b86bb Mon Sep 17 00:00:00 2001 From: SBCrumb Date: Sat, 27 Sep 2025 11:19:36 -0400 Subject: [PATCH] dev to main (#14) Reviewed-on: https://gitea.skalas.org/sbcrumb/nfoguard/pulls/14 Co-authored-by: SBCrumb Co-committed-by: SBCrumb --- VERSION | 2 +- core/episode_nfo_manager.py | 276 ++++++++++++++++++++++ core/nfo_manager.py | 246 ++++++------------- nfoguard.py | 296 ++++++++--------------- processors/tv_series_processor.py | 380 ++++++++++++++++++++++++++++++ 5 files changed, 822 insertions(+), 378 deletions(-) create mode 100644 core/episode_nfo_manager.py create mode 100644 processors/tv_series_processor.py diff --git a/VERSION b/VERSION index 9ab8337..b663c74 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.9.1 +2.1.5-preserve-long-nfo-episodes-only \ No newline at end of file diff --git a/core/episode_nfo_manager.py b/core/episode_nfo_manager.py new file mode 100644 index 0000000..b6823d4 --- /dev/null +++ b/core/episode_nfo_manager.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +""" +Episode NFO Manager - Handles TV episode NFO creation with video filename matching +Core principle: NFO filenames should match video filenames +""" +import xml.etree.ElementTree as ET +from pathlib import Path +from typing import Optional, Dict, Any, List, Tuple +import re +from .logging import _log + + +class EpisodeNFOManager: + """Manages episode NFO files with video filename matching""" + + def __init__(self, manager_brand: str = "NFOGuard"): + self.manager_brand = manager_brand + + def find_video_files_for_season(self, season_dir: Path) -> Dict[Tuple[int, int], List[Path]]: + """Find all video files in season directory, grouped by (season, episode)""" + if not season_dir.exists(): + _log("DEBUG", f"Season directory does not exist: {season_dir}") + return {} + + video_extensions = [".mkv", ".mp4", ".avi", ".mov", ".m4v"] + episodes = {} + + _log("DEBUG", f"Scanning video files in: {season_dir}") + for video_file in season_dir.iterdir(): + _log("DEBUG", f"Checking file: {video_file.name} (is_file: {video_file.is_file()}, suffix: {video_file.suffix.lower()})") + if (video_file.is_file() and + video_file.suffix.lower() in video_extensions): + + episode_info = self._parse_episode_from_filename(video_file.name) + _log("DEBUG", f"Episode parsing for '{video_file.name}': {episode_info}") + if episode_info: + season_num, episode_num = episode_info + key = (season_num, episode_num) + if key not in episodes: + episodes[key] = [] + episodes[key].append(video_file) + _log("DEBUG", f"Added video file: S{season_num:02d}E{episode_num:02d} → {video_file.name}") + + _log("DEBUG", f"Total video files found: {len(episodes)} episodes") + return episodes + + def _parse_episode_from_filename(self, filename: str) -> Optional[Tuple[int, int]]: + """Extract season and episode numbers from filename""" + # Try S##E## format first (most common) + match = re.search(r'[Ss](\d{1,2})[Ee](\d{1,2})', filename) + if match: + return int(match.group(1)), int(match.group(2)) + + # Try ##x## format + match = re.search(r'(\d{1,2})x(\d{1,2})', filename) + if match: + return int(match.group(1)), int(match.group(2)) + + return None + + def find_nfo_for_episode(self, season_dir: Path, season_num: int, episode_num: int) -> Optional[Path]: + """Find existing NFO file for episode (prefer video-matching filename)""" + if not season_dir.exists(): + return None + + # First, look for NFO files that match video filenames + video_files = self.find_video_files_for_season(season_dir) + key = (season_num, episode_num) + + if key in video_files: + for video_file in video_files[key]: + potential_nfo = season_dir / f"{video_file.stem}.nfo" + if potential_nfo.exists(): + _log("DEBUG", f"Found video-matching NFO: {potential_nfo.name}") + return potential_nfo + + # Fallback: look for short name NFO + short_nfo = season_dir / f"S{season_num:02d}E{episode_num:02d}.nfo" + if short_nfo.exists(): + _log("DEBUG", f"Found short-name NFO: {short_nfo.name}") + return short_nfo + + # Last resort: search all NFO files for matching season/episode data + for nfo_file in season_dir.glob("*.nfo"): + if self._nfo_matches_episode(nfo_file, season_num, episode_num): + _log("DEBUG", f"Found matching NFO by content: {nfo_file.name}") + return nfo_file + + return None + + def _nfo_matches_episode(self, nfo_path: Path, season_num: int, episode_num: int) -> bool: + """Check if NFO file contains the specified season/episode""" + try: + tree = ET.parse(nfo_path) + root = tree.getroot() + + if root.tag == "episodedetails": + season_elem = root.find("season") + episode_elem = root.find("episode") + + if (season_elem is not None and episode_elem is not None and + season_elem.text and episode_elem.text): + try: + file_season = int(season_elem.text) + file_episode = int(episode_elem.text) + return file_season == season_num and file_episode == episode_num + except ValueError: + pass + except (ET.ParseError, Exception): + pass + + return False + + def get_target_nfo_path(self, season_dir: Path, season_num: int, episode_num: int) -> Path: + """Get the target NFO path (prefer video filename, fallback to short name)""" + video_files = self.find_video_files_for_season(season_dir) + key = (season_num, episode_num) + + if key in video_files and video_files[key]: + # Use the first video file found (handle multiple files gracefully) + video_file = video_files[key][0] + target_nfo = season_dir / f"{video_file.stem}.nfo" + _log("DEBUG", f"Target NFO will match video: {target_nfo.name}") + return target_nfo + else: + # Fallback to short name if no video file found + target_nfo = season_dir / f"S{season_num:02d}E{episode_num:02d}.nfo" + _log("WARNING", f"No video file found for S{season_num:02d}E{episode_num:02d}, using short name: {target_nfo.name}") + return target_nfo + + def migrate_nfo_to_video_filename(self, season_dir: Path, season_num: int, episode_num: int) -> bool: + """If short-name NFO exists, rename it to match video filename""" + existing_nfo = self.find_nfo_for_episode(season_dir, season_num, episode_num) + target_nfo = self.get_target_nfo_path(season_dir, season_num, episode_num) + + # If we already have the right filename, nothing to do + if existing_nfo and existing_nfo == target_nfo: + return True + + # If we have an NFO but it doesn't match target, rename it + if existing_nfo and existing_nfo != target_nfo: + try: + _log("INFO", f"Migrating NFO filename: {existing_nfo.name} -> {target_nfo.name}") + existing_nfo.rename(target_nfo) + return True + except Exception as e: + _log("ERROR", f"Failed to rename NFO: {e}") + return False + + return False + + def create_episode_nfo(self, season_dir: Path, season_num: int, episode_num: int, + aired: Optional[str], dateadded: Optional[str], source: str, + title: Optional[str] = None, plot: Optional[str] = None) -> bool: + """Create or update episode NFO with video filename matching""" + + # Get the target NFO path (matching video filename) + nfo_path = self.get_target_nfo_path(season_dir, season_num, episode_num) + + # Migrate existing NFO if needed + self.migrate_nfo_to_video_filename(season_dir, season_num, episode_num) + + try: + # Load existing NFO if it exists + episode_elem = None + if nfo_path.exists(): + try: + tree = ET.parse(nfo_path) + episode_elem = tree.getroot() + + if episode_elem.tag != "episodedetails": + raise ValueError("Root element is not ") + + # Remove NFOGuard-managed fields (we'll re-add them) + for tag in ["season", "episode", "aired", "premiered", "dateadded", "lockdata"]: + existing = episode_elem.find(tag) + if existing is not None: + episode_elem.remove(existing) + + _log("DEBUG", f"Loaded existing NFO content for {nfo_path.name}") + + except (ET.ParseError, ValueError) as e: + _log("WARNING", f"Corrupted NFO file {nfo_path.name}: {e}. Creating new one.") + episode_elem = None + + # Create new structure if needed + if episode_elem is None: + episode_elem = ET.Element("episodedetails") + + # Add title if provided and not already present + if title and not episode_elem.find("title"): + title_elem = ET.SubElement(episode_elem, "title") + title_elem.text = title + + # Add plot if provided and not already present + if plot and not episode_elem.find("plot"): + plot_elem = ET.SubElement(episode_elem, "plot") + plot_elem.text = plot + + # Add NFOGuard fields at the end + season_elem = ET.SubElement(episode_elem, "season") + season_elem.text = str(season_num) + + episode_num_elem = ET.SubElement(episode_elem, "episode") + episode_num_elem.text = str(episode_num) + + if aired: + aired_elem = ET.SubElement(episode_elem, "aired") + aired_elem.text = aired[:10] if len(aired) >= 10 else aired + + # Also add premiered for compatibility + premiered_elem = ET.SubElement(episode_elem, "premiered") + premiered_elem.text = aired[:10] if len(aired) >= 10 else aired + + if dateadded: + dateadded_elem = ET.SubElement(episode_elem, "dateadded") + dateadded_elem.text = dateadded + + # Add lockdata + lockdata_elem = ET.SubElement(episode_elem, "lockdata") + lockdata_elem.text = "true" + + # Add comment with source + comment = ET.Comment(f" Created by {self.manager_brand} - Source: {source} ") + episode_elem.append(comment) + + # Write the NFO file + tree = ET.ElementTree(episode_elem) + ET.indent(tree, space=" ", level=0) + tree.write(nfo_path, encoding='utf-8', xml_declaration=True) + + _log("INFO", f"✅ Created/updated episode NFO: {nfo_path.name}") + _log("INFO", f" S{season_num:02d}E{episode_num:02d}, Aired: {aired}, DateAdded: {dateadded}, Source: {source}") + + return True + + except Exception as e: + _log("ERROR", f"Failed to create episode NFO {nfo_path}: {e}") + return False + + def extract_nfoguard_data(self, nfo_path: Path) -> Optional[Dict[str, str]]: + """Extract NFOGuard-managed data from existing NFO""" + if not nfo_path.exists(): + return None + + try: + tree = ET.parse(nfo_path) + root = tree.getroot() + + if root.tag != "episodedetails": + return None + + # Look for NFOGuard fields + dateadded_elem = root.find("dateadded") + aired_elem = root.find("aired") + lockdata_elem = root.find("lockdata") + + # Only consider it NFOGuard-managed if it has dateadded and lockdata + if (dateadded_elem is not None and dateadded_elem.text and + lockdata_elem is not None and lockdata_elem.text == "true"): + + result = { + "dateadded": dateadded_elem.text.strip(), + "source": "existing_nfo" + } + + if aired_elem is not None and aired_elem.text: + result["aired"] = aired_elem.text.strip() + + _log("DEBUG", f"Found NFOGuard data in {nfo_path.name}: {result}") + return result + + except (ET.ParseError, Exception) as e: + _log("WARNING", f"Error parsing NFO {nfo_path.name}: {e}") + + return None \ No newline at end of file diff --git a/core/nfo_manager.py b/core/nfo_manager.py index 6e47cf2..62de808 100644 --- a/core/nfo_manager.py +++ b/core/nfo_manager.py @@ -7,7 +7,7 @@ import os import xml.etree.ElementTree as ET from pathlib import Path from datetime import datetime -from typing import Optional, Dict, Any +from typing import Optional, Dict, Any, Tuple import re @@ -163,7 +163,7 @@ class NFOManager: def extract_nfoguard_dates_from_episode_nfo(self, season_path: Path, season_num: int, episode_num: int) -> Optional[Dict[str, str]]: """Extract NFOGuard-managed dates from existing episode NFO file""" - nfo_filename = f"S{season_num:02d}E{episode_num:02d}.nfo" + nfo_filename = f"s{season_num:02d}e{episode_num:02d}.nfo" nfo_path = season_path / nfo_filename if not nfo_path.exists(): @@ -190,11 +190,7 @@ class NFOManager: if aired_elem is not None and aired_elem.text: result["aired"] = aired_elem.text.strip() - # Check if title is missing - title_elem = root.find('.//title') - result["has_title"] = title_elem is not None and title_elem.text and title_elem.text.strip() - - print(f"✅ Found NFOGuard data in episode NFO S{season_num:02d}E{episode_num:02d}: dateadded={result['dateadded']}, aired={result.get('aired', 'None')}, has_title={result['has_title']}") + print(f"✅ Found NFOGuard data in episode NFO S{season_num:02d}E{episode_num:02d}: dateadded={result['dateadded']}, aired={result.get('aired', 'None')}") return result except (ET.ParseError, Exception) as e: @@ -203,104 +199,6 @@ class NFOManager: return None - def enhance_existing_episode_nfo_with_title(self, season_path: Path, season_num: int, episode_num: int, title: str) -> bool: - """Add title to existing episode NFO file that's missing one""" - nfo_filename = f"S{season_num:02d}E{episode_num:02d}.nfo" - nfo_path = season_path / nfo_filename - - if not nfo_path.exists(): - return False - - try: - tree = ET.parse(nfo_path) - root = tree.getroot() - - # Check if title already exists - existing_title = root.find('.//title') - if existing_title is not None and existing_title.text and existing_title.text.strip(): - return False # Title already exists - - # Add title element (insert near the top, after any existing plot element) - title_elem = ET.Element("title") - title_elem.text = title - - # Find the best position to insert title (after plot if it exists, otherwise at the beginning) - plot_elem = root.find('.//plot') - if plot_elem is not None: - # Insert after plot - plot_index = list(root).index(plot_elem) - root.insert(plot_index + 1, title_elem) - else: - # Insert at the beginning (after any existing title that might be empty) - if existing_title is not None: - title_index = list(root).index(existing_title) - root.remove(existing_title) - root.insert(title_index, title_elem) - else: - root.insert(0, title_elem) - - # Write the updated NFO - tree.write(nfo_path, encoding='utf-8', xml_declaration=True) - print(f"📝 Enhanced existing NFO with title: S{season_num:02d}E{episode_num:02d} - '{title}'") - return True - - except Exception as e: - print(f"⚠️ Error enhancing NFO with title: {e}") - return False - - def _extract_title_from_filename(self, season_dir: Path, season_num: int, episode_num: int) -> Optional[str]: - """Extract episode title from video filename as fallback when metadata doesn't provide it""" - try: - import re - # Look for video files matching this episode - season_pattern = f"S{season_num:02d}E{episode_num:02d}" - print(f"🔍 Searching for title in files for {season_pattern} in directory: {season_dir}") - - for video_file in season_dir.glob("*.mkv"): - filename = video_file.name - print(f"🔍 Checking file: {filename}") - if season_pattern in filename: - print(f"✅ Found matching file: {filename}") - # Pattern: SeriesName-S01E01-Episode Title[WEBDL-1080p][AAC2.0][h264].mkv - # Extract the part between season/episode and the first bracket - pattern = rf'{season_pattern}-(.*?)\[' - print(f"🔍 Using regex pattern: {pattern}") - match = re.search(pattern, filename) - if match: - title = match.group(1).strip() - print(f"🔍 Raw extracted title: '{title}'") - # Clean up common encoding artifacts and separators - title = title.replace('-', ' ').strip() - if title: - print(f"✅ Extracted title from filename: '{title}' for {season_pattern}") - return title - else: - print(f"⚠️ Title was empty after cleanup") - else: - print(f"⚠️ Regex pattern didn't match filename") - - # Also check .mp4 files - for video_file in season_dir.glob("*.mp4"): - filename = video_file.name - print(f"🔍 Checking .mp4 file: {filename}") - if season_pattern in filename: - print(f"✅ Found matching .mp4 file: {filename}") - pattern = rf'{season_pattern}-(.*?)\[' - match = re.search(pattern, filename) - if match: - title = match.group(1).strip() - print(f"🔍 Raw extracted title from .mp4: '{title}'") - title = title.replace('-', ' ').strip() - if title: - print(f"✅ Extracted title from .mp4 filename: '{title}' for {season_pattern}") - return title - - except Exception as e: - print(f"⚠️ Error extracting title from filename for S{season_num:02d}E{episode_num:02d}: {e}") - - print(f"⚠️ No title found in filenames for {season_pattern}") - return None - def _parse_nfo_with_tolerance(self, nfo_path: Path): """Parse NFO file with tolerance for URLs appended after XML""" try: @@ -566,18 +464,12 @@ class NFOManager: print(f"❌ Error creating/updating season NFO {nfo_path}: {e}") def find_existing_episode_nfo(self, season_dir: Path, season_num: int, episode_num: int) -> Optional[Path]: - """Find existing episode NFO file that matches season/episode but isn't standardized name""" + """Find any existing episode NFO file that matches season/episode""" if not season_dir.exists(): return None - - # Standard filename pattern we're looking to create - standard_pattern = f"S{season_num:02d}E{episode_num:02d}.nfo" # Look for NFO files in the season directory for nfo_file in season_dir.glob("*.nfo"): - # Skip if it's already the standard format - if nfo_file.name == standard_pattern: - continue # Check if this NFO contains the right season/episode try: @@ -596,7 +488,7 @@ class NFOManager: file_episode = int(episode_elem.text) if file_season == season_num and file_episode == episode_num: - print(f"🔍 Found existing episode NFO: {nfo_file.name} -> will migrate to {standard_pattern}") + print(f"🔍 Found existing episode NFO: {nfo_file.name}") return nfo_file except ValueError: continue @@ -607,23 +499,62 @@ class NFOManager: return None + def _get_target_episode_nfo_name(self, season_dir: Path, season_num: int, episode_num: int) -> str: + """Get target NFO filename - prefer existing NFO, then matching video file, fallback to short name""" + if not season_dir.exists(): + return f"S{season_num:02d}E{episode_num:02d}.nfo" + + # First check if an NFO already exists for this episode + existing_nfo = self.find_existing_episode_nfo(season_dir, season_num, episode_num) + if existing_nfo: + print(f"📂 Existing NFO found, will preserve filename: {existing_nfo.name}") + return existing_nfo.name + + # Look for video files with matching season/episode + video_extensions = [".mkv", ".mp4", ".avi", ".mov", ".m4v", ".wmv", ".flv", ".webm"] + + for video_file in season_dir.iterdir(): + if (video_file.is_file() and + video_file.suffix.lower() in video_extensions): + + # Parse episode info from video filename + episode_info = self._parse_episode_from_filename(video_file.name) + if episode_info and episode_info == (season_num, episode_num): + # Found matching video file - use its name for NFO + target_nfo_name = f"{video_file.stem}.nfo" + print(f"🎯 Target NFO will match video: {target_nfo_name}") + return target_nfo_name + + # Fallback to short name if no matching video found + short_name = f"S{season_num:02d}E{episode_num:02d}.nfo" + print(f"⚠️ No matching video file found, using short name: {short_name}") + return short_name + + def _parse_episode_from_filename(self, filename: str) -> Optional[Tuple[int, int]]: + """Parse season and episode numbers from filename""" + # Try S##E## format first + match = re.search(r'[Ss](\d{1,2})[Ee](\d{1,2})', filename) + if match: + return int(match.group(1)), int(match.group(2)) + + # Try ##x## format + match = re.search(r'(\d{1,2})x(\d{1,2})', filename) + if match: + return int(match.group(1)), int(match.group(2)) + + return None + def create_episode_nfo(self, season_dir: Path, season_num: int, episode_num: int, aired: Optional[str], dateadded: Optional[str], source: str, lock_metadata: bool = True, enhanced_metadata: Optional[Dict[str, Any]] = None) -> None: """Create or update episode NFO file preserving existing content""" - # Generate episode filename pattern - episode_filename = f"S{season_num:02d}E{episode_num:02d}.nfo" - nfo_path = season_dir / episode_filename - - # Track if we need to delete an old long-named NFO file - old_nfo_to_delete = None + # Get target NFO filename (prefer long name matching video file) + target_nfo_name = self._get_target_episode_nfo_name(season_dir, season_num, episode_num) + nfo_path = season_dir / target_nfo_name try: - # First, check for existing long-named NFO files that need migration - existing_long_nfo = self.find_existing_episode_nfo(season_dir, season_num, episode_num) - - # Prioritize long-named file for migration, otherwise use standard file - source_nfo_path = existing_long_nfo if existing_long_nfo else nfo_path if nfo_path.exists() else None + # Check for existing NFO file at target location + source_nfo_path = nfo_path if nfo_path.exists() else None if source_nfo_path: try: @@ -634,20 +565,7 @@ class NFOManager: if episode.tag != "episodedetails": raise ValueError("Root element is not ") - # If we're migrating from a long-named file, mark it for deletion - if existing_long_nfo and source_nfo_path == existing_long_nfo: - old_nfo_to_delete = existing_long_nfo - print(f"📦 Migrating episode NFO: {existing_long_nfo.name} -> {episode_filename}") - - # Show what content fields are being preserved - content_fields = ["title", "plot", "runtime", "premiered"] - preserved_content = [] - for field in content_fields: - elem = episode.find(field) - if elem is not None and elem.text: - preserved_content.append(field) - if preserved_content: - print(f" 📄 Preserving content: {', '.join(preserved_content)}") + print(f"📝 Updating existing NFO: {nfo_path.name}") # Preserve existing content fields and store NFOGuard-managed fields for re-adding preserved_values = {} @@ -696,29 +614,6 @@ class NFOManager: runtime_elem = ET.SubElement(episode, "runtime") runtime_elem.text = str(enhanced_metadata["runtime"]) - # Fallback: Extract title from filename if no valid title present - title_elem = episode.find("title") - has_valid_title = title_elem is not None and title_elem.text and title_elem.text.strip() - - print(f"🔍 Title check for S{season_num:02d}E{episode_num:02d}: has_element={title_elem is not None}, has_text={title_elem.text if title_elem is not None else 'N/A'}, has_valid_title={has_valid_title}") - - if not has_valid_title: - print(f"🔍 No valid title found in NFO for S{season_num:02d}E{episode_num:02d}, attempting filename extraction") - filename_title = self._extract_title_from_filename(season_dir, season_num, episode_num) - if filename_title: - # Remove existing empty/invalid title element if present - if title_elem is not None: - episode.remove(title_elem) - - # Add new title element - title_elem = ET.SubElement(episode, "title") - title_elem.text = filename_title - print(f"✅ Added filename-extracted title to NFO: S{season_num:02d}E{episode_num:02d} - '{filename_title}'") - else: - print(f"⚠️ Could not extract title from filename for S{season_num:02d}E{episode_num:02d}") - else: - print(f"✅ Valid title already exists in NFO for S{season_num:02d}E{episode_num:02d}: '{title_elem.text.strip()}')") - # Add NFOGuard fields at the bottom # Basic episode info at the end @@ -742,28 +637,28 @@ class NFOManager: lockdata = ET.SubElement(episode, "lockdata") lockdata.text = "true" - # Add NFOGuard comment at the bottom with other NFOGuard fields - nfoguard_comment = ET.Comment(f" Created by {self.manager_brand} - Source: {source} ") - episode.append(nfoguard_comment) + # Add NFOGuard comment at the beginning + comment_text = f" Created by {self.manager_brand} - Source: {source} " # Write file with proper formatting tree = ET.ElementTree(episode) ET.indent(tree, space=" ", level=0) - # Write to file normally (ET will handle the comment properly) - tree.write(nfo_path, encoding='utf-8', xml_declaration=True) + # Write to string first to add comment + xml_str = ET.tostring(episode, encoding='unicode') + + # Add XML declaration and comment + full_xml = f'\n\n{xml_str}' + + # Write to file + with open(nfo_path, 'w', encoding='utf-8') as f: + f.write(full_xml) print(f"✅ Successfully created/updated episode NFO: {nfo_path}") print(f" S{season_num:02d}E{episode_num:02d}, Aired: {aired}, Date Added: {dateadded}") - # Clean up old long-named NFO file if we migrated from it - if old_nfo_to_delete and old_nfo_to_delete.exists(): - try: - old_nfo_to_delete.unlink() - print(f"🗑️ Cleaned up old NFO file: {old_nfo_to_delete.name}") - except Exception as cleanup_error: - print(f"⚠️ Warning: Could not delete old NFO file {old_nfo_to_delete.name}: {cleanup_error}") - + # NFO file created/updated successfully + pass except Exception as e: print(f"❌ Error creating/updating episode NFO {nfo_path}: {e}") @@ -802,5 +697,4 @@ class NFOManager: if updated_files: print(f"✅ Updated {len(updated_files)} video file timestamps in {movie_dir.name}") else: - print(f"⚠️ No video files found to update in {movie_dir.name}") - + print(f"⚠️ No video files found to update in {movie_dir.name}") \ No newline at end of file diff --git a/nfoguard.py b/nfoguard.py index f29275c..9ffb8f6 100644 --- a/nfoguard.py +++ b/nfoguard.py @@ -380,28 +380,39 @@ class TVProcessor: # 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) + # 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 = re.sub(r'\s+', ' ', name).strip() + + return name if name else None + 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") + _log("DEBUG", f"Scanning for episodes in: {series_path}") + _log("DEBUG", f"Looking for season dirs with pattern: '{config.tv_season_dir_pattern}'") + for season_dir in series_path.iterdir(): + _log("DEBUG", f"Checking: {season_dir.name} (is_dir: {season_dir.is_dir()})") if not (season_dir.is_dir() and season_dir.name.lower().startswith(config.tv_season_dir_pattern)): + _log("DEBUG", f"Skipping {season_dir.name} - doesn't match season pattern") continue try: @@ -440,29 +451,12 @@ class TVProcessor: 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) + # Find missing episodes 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") + _log("INFO", "All episodes found in cache") return episode_dates _log("INFO", f"Querying APIs for {len(missing_keys)} missing episodes") @@ -470,6 +464,15 @@ class TVProcessor: # Query Sonarr for missing episodes if self.sonarr.enabled: series = self.sonarr.series_by_imdb(imdb_id) + if not series: + _log("DEBUG", f"Trying direct series search for IMDb: {imdb_id}") + series = self.sonarr.series_by_imdb_direct(imdb_id) + if not series: + # Fallback: try searching by series title extracted from path + series_title = self._extract_series_title_from_path(series_path) + if series_title: + _log("DEBUG", f"Trying title search for: {series_title}") + series = self.sonarr.series_by_title(series_title) if series: series_id = series.get("id") if series_id: @@ -568,7 +571,7 @@ class TVProcessor: for (season, episode), (aired, dateadded, source) in episode_dates.items(): if (season, episode) in disk_episodes: # Get enhanced episode metadata - enhanced_metadata = self._get_episode_metadata(series_metadata, season, episode, season_path) if series_metadata else self._get_episode_metadata(None, season, episode, season_path) + enhanced_metadata = self._get_episode_metadata(series_metadata, season, episode) if series_metadata else None # Create NFO if config.manage_nfo: @@ -587,12 +590,8 @@ class TVProcessor: # Save to database self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True) - # Create season NFO - if config.manage_nfo: - self.nfo_manager.create_season_nfo(season_path, season_num) - # 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) + # Skip season.nfo and tvshow.nfo creation - focus only on episode NFOs + pass _log("INFO", f"Completed processing season {season_num}") @@ -623,21 +622,6 @@ class TVProcessor: source = nfo_data["source"] aired = nfo_data.get("aired") - # Check if NFO is missing title and try to enhance it - if not nfo_data.get("has_title", False): - _log("INFO", f"NFO S{season_num:02d}E{episode_num:02d} missing title, attempting to extract from filename") - filename_title = self._extract_title_from_filename(season_path, season_num, episode_num) - if filename_title: - enhanced = self.nfo_manager.enhance_existing_episode_nfo_with_title( - season_path, season_num, episode_num, filename_title - ) - if enhanced: - _log("INFO", f"✅ Enhanced NFO S{season_num:02d}E{episode_num:02d} with title: '{filename_title}'") - else: - _log("WARNING", f"Failed to enhance NFO S{season_num:02d}E{episode_num:02d} with title") - else: - _log("WARNING", f"Could not extract title from filename for S{season_num:02d}E{episode_num:02d}") - # Update file mtime if enabled (NFO is already correct) if config.fix_dir_mtimes and dateadded: self.nfo_manager.set_file_mtime(episode_file, dateadded) @@ -654,14 +638,12 @@ class TVProcessor: source = existing_episode["source"] aired = existing_episode.get("aired") - # Create NFO with existing data, but try to add title from filename if missing + # Create NFO with existing data (no enhanced metadata needed) if config.manage_nfo and dateadded: - # Try to extract title from filename as fallback for Tier 2 processing - enhanced_metadata = self._get_episode_metadata(None, season_num, episode_num, season_path) self.nfo_manager.create_episode_nfo( season_path, season_num, episode_num, aired, dateadded, source, config.lock_metadata, - enhanced_metadata # Include filename-extracted title if available + None # No enhanced metadata for database-cached episodes ) # Update file mtime if enabled @@ -676,13 +658,13 @@ class TVProcessor: # 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) + enhanced_metadata = self._get_episode_metadata(series_metadata, season_num, episode_num) if series_metadata else None # 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: + # Create NFO (always create if manage_nfo is enabled, even without dateadded) + if config.manage_nfo: self.nfo_manager.create_episode_nfo( season_path, season_num, episode_num, aired, dateadded, source, config.lock_metadata, @@ -731,7 +713,6 @@ class TVProcessor: return None - def _get_sonarr_series_metadata(self, imdb_id: str) -> Optional[Dict[str, Any]]: """Get enhanced series metadata from Sonarr API""" try: @@ -739,6 +720,9 @@ class TVProcessor: return None series = self.sonarr.series_by_imdb(imdb_id) + if not series: + _log("DEBUG", f"Trying direct series search for IMDb: {imdb_id}") + series = self.sonarr.series_by_imdb_direct(imdb_id) if not series: _log("DEBUG", f"No Sonarr series found for IMDb {imdb_id}") return None @@ -756,97 +740,20 @@ class TVProcessor: _log("ERROR", f"Error getting Sonarr metadata for {imdb_id}: {e}") return None - def _get_episode_metadata(self, series_metadata: Dict[str, Any], season_num: int, episode_num: int, season_dir: Optional[Path] = None) -> Optional[Dict[str, Any]]: - """Extract specific episode metadata from series data with filename fallback for titles""" - _log("INFO", f"🔍 _get_episode_metadata called for S{season_num:02d}E{episode_num:02d}, season_dir: {season_dir}") - metadata = { - "title": None, - "overview": None, - "runtime": None, - "ratings": {} - } + def _get_episode_metadata(self, series_metadata: Dict[str, Any], season_num: int, episode_num: int) -> Optional[Dict[str, Any]]: + """Extract specific episode metadata from series data""" + if not series_metadata or "episodes" not in series_metadata: + return None - # Try to get metadata from Sonarr first - if series_metadata and "episodes" in series_metadata: - for episode in series_metadata["episodes"]: - if episode.get("seasonNumber") == season_num and episode.get("episodeNumber") == episode_num: - metadata.update({ - "title": episode.get("title"), - "overview": episode.get("overview"), - "runtime": episode.get("runtime"), - "ratings": episode.get("ratings", {}) - }) - break + for episode in series_metadata["episodes"]: + if episode.get("seasonNumber") == season_num and episode.get("episodeNumber") == episode_num: + return { + "title": episode.get("title"), + "overview": episode.get("overview"), + "runtime": episode.get("runtime"), + "ratings": episode.get("ratings", {}) + } - # If no title from Sonarr, try to extract from filename - if not metadata["title"] and season_dir: - _log("INFO", f"📁 No title from Sonarr for S{season_num:02d}E{episode_num:02d}, trying filename extraction") - filename_title = self._extract_title_from_filename(season_dir, season_num, episode_num) - if filename_title: - metadata["title"] = filename_title - _log("INFO", f"✅ Using filename-extracted title for S{season_num:02d}E{episode_num:02d}: '{filename_title}'") - else: - _log("DEBUG", f"⚠️ No filename title extracted for S{season_num:02d}E{episode_num:02d}") - elif metadata["title"]: - _log("DEBUG", f"✅ Using Sonarr title for S{season_num:02d}E{episode_num:02d}: '{metadata['title']}'") - - # Return metadata if we have at least some information - if any(metadata.values()): - return metadata - - return None - - def _extract_title_from_filename(self, season_dir: Path, season_num: int, episode_num: int) -> Optional[str]: - """Extract episode title from video filename as fallback when Sonarr doesn't provide it""" - try: - import re - # Look for video files matching this episode - season_pattern = f"S{season_num:02d}E{episode_num:02d}" - _log("INFO", f"🔍 Searching for title in files for {season_pattern} in directory: {season_dir}") - - for video_file in season_dir.glob("*.mkv"): - filename = video_file.name - _log("DEBUG", f"🔍 Checking file: {filename}") - if season_pattern in filename: - _log("DEBUG", f"✅ Found matching file: {filename}") - # Pattern: SeriesName-S01E01-Episode Title[WEBDL-1080p][AAC2.0][h264].mkv - # Extract the part between season/episode and the first bracket - pattern = rf'{season_pattern}-(.*?)\[' - _log("DEBUG", f"🔍 Using regex pattern: {pattern}") - match = re.search(pattern, filename) - if match: - title = match.group(1).strip() - _log("DEBUG", f"🔍 Raw extracted title: '{title}'") - # Clean up common encoding artifacts and separators - title = title.replace('-', ' ').strip() - if title: - _log("INFO", f"✅ Extracted title from filename: '{title}' for {season_pattern}") - return title - else: - _log("DEBUG", f"⚠️ Title was empty after cleanup") - else: - _log("DEBUG", f"⚠️ Regex pattern didn't match filename") - - # Also check .mp4 files - for video_file in season_dir.glob("*.mp4"): - filename = video_file.name - _log("DEBUG", f"🔍 Checking .mp4 file: {filename}") - if season_pattern in filename: - _log("DEBUG", f"✅ Found matching .mp4 file: {filename}") - pattern = rf'{season_pattern}-(.*?)\[' - match = re.search(pattern, filename) - if match: - title = match.group(1).strip() - _log("DEBUG", f"🔍 Raw extracted title from .mp4: '{title}'") - title = title.replace('-', ' ').strip() - if title: - _log("INFO", f"✅ Extracted title from .mp4 filename: '{title}' for {season_pattern}") - return title - - except Exception as e: - _log("ERROR", f"Error extracting title from filename for S{season_num:02d}E{episode_num:02d}: {e}") - - _log("DEBUG", f"⚠️ No title found in filenames for {season_pattern}") return None def _gather_season_episode_dates(self, series_path: Path, imdb_id: str, season_num: int, disk_episodes: Dict[Tuple[int, int], List[Path]], series_metadata: Optional[Dict[str, Any]] = None) -> Dict[Tuple[int, int], Tuple[Optional[str], Optional[str], str]]: @@ -880,6 +787,14 @@ class TVProcessor: if self.sonarr.enabled: try: series = self.sonarr.series_by_imdb(imdb_id) + if not series: + _log("DEBUG", f"Trying direct series search for IMDb: {imdb_id}") + series = self.sonarr.series_by_imdb_direct(imdb_id) + if not series: + # Try title search as fallback for IMDb mismatches + _log("DEBUG", f"Trying title search fallback for IMDb: {imdb_id}") + # We don't have series_path here, so just try the imdb_id as a title search + # This is less ideal but better than nothing if series: episodes = self.sonarr.episodes_for_series(series["id"]) for ep in episodes: @@ -963,7 +878,7 @@ class TVProcessor: # Get episode date information - webhook processing prioritizes existing DB entries _log("DEBUG", f"Processing webhook episode: IMDb={imdb_id}, S{season_num:02d}E{episode_num:02d}") aired, dateadded, source = self._get_webhook_episode_date(imdb_id, season_num, episode_num, series_metadata) - enhanced_metadata = self._get_episode_metadata(series_metadata, season_num, episode_num, season_dir) if series_metadata else self._get_episode_metadata(None, season_num, episode_num, season_dir) + enhanced_metadata = self._get_episode_metadata(series_metadata, season_num, episode_num) if series_metadata else None # Create NFO if config.manage_nfo: @@ -990,20 +905,8 @@ class TVProcessor: episodes_processed += 1 - # Create season/tvshow NFOs if any episodes were processed - if episodes_processed > 0 and config.manage_nfo: - seasons_processed = set() - for webhook_episode in webhook_episodes: - season_num = webhook_episode.get("seasonNumber") - if season_num and season_num not in seasons_processed: - season_dir = series_path / config.tv_season_dir_format.format(season=season_num) - if season_dir.exists(): - self.nfo_manager.create_season_nfo(season_dir, season_num) - seasons_processed.add(season_num) - - # 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) + # Skip season.nfo and tvshow.nfo creation - focus only on episode NFOs + pass _log("INFO", f"Completed targeted processing: {episodes_processed}/{len(webhook_episodes)} episodes processed") @@ -1047,6 +950,9 @@ class TVProcessor: if not aired and self.sonarr.enabled: try: series = self.sonarr.series_by_imdb(imdb_id) + if not series: + _log("DEBUG", f"Trying direct series search for IMDb: {imdb_id}") + series = self.sonarr.series_by_imdb_direct(imdb_id) if series: episodes = self.sonarr.episodes_for_series(series["id"]) for ep in episodes: @@ -1562,13 +1468,12 @@ class MovieProcessor: class WebhookBatcher: """Batches webhook events to avoid processing storms""" - def __init__(self, nfo_manager: NFOManager): + def __init__(self): self.pending: Dict[str, Dict] = {} self.timers: Dict[str, threading.Timer] = {} self.processing: Set[str] = set() self.lock = threading.Lock() self.executor = ThreadPoolExecutor(max_workers=config.max_concurrent) - self.nfo_manager = nfo_manager def add_webhook(self, key: str, webhook_data: Dict, media_type: str): """Add webhook to batch queue""" @@ -1622,23 +1527,11 @@ class WebhookBatcher: # CRITICAL: Validate that the path contains the expected IMDb ID for movies if media_type == 'movie': expected_imdb = key.replace('movie:', '') if key.startswith('movie:') else key - # Use comprehensive IMDb detection (directory, filenames, NFO content) - detected_imdb = self.nfo_manager.find_movie_imdb_id(path_obj) - - # Check if detected IMDb matches expected (handle both full IMDb IDs and just numbers) - imdb_match = False - if detected_imdb: - if detected_imdb == expected_imdb: - imdb_match = True - elif detected_imdb.replace('tt', '') == expected_imdb.replace('tt', ''): - imdb_match = True - - if not imdb_match: + if expected_imdb not in path_str.lower(): _log("ERROR", f"BATCH VALIDATION FAILED: Expected IMDb {expected_imdb} not found in path {path_str}") - _log("ERROR", f"Detected IMDb: {detected_imdb}, Expected: {expected_imdb}") _log("ERROR", f"This prevents processing wrong movies due to batch corruption") return - _log("DEBUG", f"Batch validation passed: Expected IMDb {expected_imdb} matches detected IMDb {detected_imdb}") + _log("DEBUG", f"Batch validation passed: IMDb {expected_imdb} found in path") # Process based on media type if media_type == 'tv': @@ -1721,7 +1614,7 @@ nfo_manager = NFOManager(config.manager_brand, config.debug) path_mapper = PathMapper(config) # FIXED: Pass config to PathMapper tv_processor = TVProcessor(db, nfo_manager, path_mapper) movie_processor = MovieProcessor(db, nfo_manager, path_mapper) -batcher = WebhookBatcher(nfo_manager) +batcher = WebhookBatcher() # --------------------------- # Webhook Handlers @@ -1834,18 +1727,9 @@ async def radarr_webhook(request: Request, background_tasks: BackgroundTasks): _log("ERROR", f"This prevents processing wrong movies due to path mapping issues") return {"status": "error", "message": f"Mapped movie path does not exist: {container_path}"} - # Verify the path contains the expected IMDb ID using comprehensive detection - detected_imdb = nfo_manager.find_movie_imdb_id(Path(container_path)) - imdb_match = False - if detected_imdb: - if detected_imdb == imdb_id or detected_imdb.replace('tt', '') == imdb_id.replace('tt', ''): - imdb_match = True - - if not imdb_match: - _log("WARNING", f"IMDb ID {imdb_id} not found via comprehensive detection in {container_path}") - _log("DEBUG", f"Detected IMDb: {detected_imdb}, Expected: {imdb_id}") - else: - _log("DEBUG", f"IMDb ID validated: {imdb_id} matches detected {detected_imdb}") + # Verify the path contains the expected IMDb ID + if imdb_id not in container_path.lower(): + _log("WARNING", f"IMDb ID {imdb_id} not found in container path {container_path}") # Create movie-specific webhook data with proper path validation movie_webhook_data = { @@ -2193,13 +2077,23 @@ async def manual_scan(background_tasks: BackgroundTasks, path: Optional[str] = N except Exception as e: _log("ERROR", f"Failed processing episode {scan_path}: {e}") else: - # Full series processing - for item in scan_path.iterdir(): - if item.is_dir() and nfo_manager.parse_imdb_from_path(item): - try: - tv_processor.process_series(item) - except Exception as e: - _log("ERROR", f"Failed processing TV series {item}: {e}") + # Check if this path itself is a series (has IMDb ID in the directory name) + if nfo_manager.parse_imdb_from_path(scan_path): + try: + tv_processor.process_series(scan_path) + except Exception as e: + _log("ERROR", f"Failed processing TV series {scan_path}: {e}") + else: + # Full series processing - scan subdirectories + for item in scan_path.iterdir(): + if (item.is_dir() and + not item.name.lower().startswith('season') and + not re.match(r'^season\s+\d+$', item.name, re.IGNORECASE) and + nfo_manager.parse_imdb_from_path(item)): + try: + tv_processor.process_series(item) + except Exception as e: + _log("ERROR", f"Failed processing TV series {item}: {e}") if scan_type in ["both", "movies"] and scan_path in config.movie_paths: _log("INFO", f"Scanning movies in: {scan_path}") diff --git a/processors/tv_series_processor.py b/processors/tv_series_processor.py new file mode 100644 index 0000000..07ceedc --- /dev/null +++ b/processors/tv_series_processor.py @@ -0,0 +1,380 @@ +#!/usr/bin/env python3 +""" +TV Series Processor - Clean implementation for TV episode processing +Handles manual scans and webhook processing with proper NFO filename matching +""" +from pathlib import Path +from typing import Optional, Dict, Any, List, Tuple +from datetime import datetime, timezone +import re + +from core.database import NFOGuardDatabase +from core.episode_nfo_manager import EpisodeNFOManager +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 core.logging import _log, convert_utc_to_local + + +class TVSeriesProcessor: + """Clean TV series processor with video filename matching""" + + def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager, + path_mapper: PathMapper, sonarr_client: SonarrClient): + self.db = db + self.nfo_manager = nfo_manager # Keep for series/season NFOs + self.episode_nfo_manager = EpisodeNFOManager() + self.path_mapper = path_mapper + self.sonarr = sonarr_client + self.external_manager = ExternalClientManager() + + def process_series_manual_scan(self, series_path: Path) -> bool: + """Process a TV series during manual scan""" + _log("INFO", f"Processing TV series: {series_path.name}") + + # Extract IMDb ID + imdb_id = self._extract_imdb_id(series_path) + if not imdb_id: + _log("ERROR", f"No IMDb ID found for series: {series_path.name}") + return False + + # Find all episodes on disk + episodes_on_disk = self._find_episodes_on_disk(series_path) + if not episodes_on_disk: + _log("WARNING", f"No episodes found on disk for: {series_path.name}") + return False + + _log("INFO", f"Found {len(episodes_on_disk)} episodes on disk") + + # Process each episode + episodes_processed = 0 + for (season_num, episode_num), video_files in episodes_on_disk.items(): + if self._process_episode_manual_scan(series_path, imdb_id, season_num, episode_num): + episodes_processed += 1 + + # Create series-level NFOs if any episodes were processed + if episodes_processed > 0: + self._create_series_nfos(series_path, imdb_id) + + _log("INFO", f"Completed processing series: {series_path.name} ({episodes_processed} episodes)") + return episodes_processed > 0 + + def process_episode_webhook(self, webhook_data: Dict[str, Any]) -> bool: + """Process a single episode from Sonarr webhook""" + # TODO: Parse webhook data and extract episode info + # This will be implemented when we add webhook support + pass + + def _extract_imdb_id(self, series_path: Path) -> Optional[str]: + """Extract IMDb ID from series directory or files""" + # Try directory name first + imdb_id = self.nfo_manager.parse_imdb_from_path(series_path) + if imdb_id: + return imdb_id + + # Try tvshow.nfo if it exists + tvshow_nfo = series_path / "tvshow.nfo" + if tvshow_nfo.exists(): + imdb_id = self.nfo_manager.parse_imdb_from_nfo(tvshow_nfo) + if imdb_id: + return imdb_id + + # Try any existing episode NFO files + for season_dir in series_path.iterdir(): + if season_dir.is_dir() and self._is_season_directory(season_dir.name): + for nfo_file in season_dir.glob("*.nfo"): + imdb_id = self.nfo_manager.parse_imdb_from_nfo(nfo_file) + if imdb_id: + return imdb_id + + return None + + def _is_season_directory(self, dirname: str) -> bool: + """Check if directory name matches season pattern""" + return bool(re.match(r'^[Ss]eason\s+\d+$', dirname, re.IGNORECASE)) + + def _find_episodes_on_disk(self, series_path: Path) -> Dict[Tuple[int, int], List[Path]]: + """Find all episodes on disk, grouped by (season, episode)""" + episodes = {} + + try: + _log("DEBUG", f"Scanning for season directories in: {series_path}") + _log("DEBUG", f"Series path exists: {series_path.exists()}, is_dir: {series_path.is_dir()}") + + if not series_path.exists() or not series_path.is_dir(): + _log("ERROR", f"Series path does not exist or is not a directory: {series_path}") + return episodes + + # List all items in directory for debugging + try: + items = list(series_path.iterdir()) + _log("DEBUG", f"Found {len(items)} items in series directory") + for item in items: + _log("DEBUG", f" Item: {item.name} (is_dir: {item.is_dir()})") + except Exception as e: + _log("ERROR", f"Failed to list directory contents: {e}") + return episodes + + for season_dir in series_path.iterdir(): + _log("DEBUG", f"Checking directory: {season_dir.name} (is_dir: {season_dir.is_dir()})") + _log("DEBUG", f"Season directory regex test for '{season_dir.name}': {self._is_season_directory(season_dir.name)}") + + if season_dir.is_dir() and self._is_season_directory(season_dir.name): + season_num = self._extract_season_number(season_dir.name) + _log("DEBUG", f"Found season directory: {season_dir.name} → season {season_num}") + if season_num is None: + _log("WARNING", f"Could not extract season number from: {season_dir.name}") + continue + + # Find video files in this season + season_episodes = self.episode_nfo_manager.find_video_files_for_season(season_dir) + _log("DEBUG", f"Found {len(season_episodes)} episodes in {season_dir.name}: {list(season_episodes.keys())}") + + # Add season directory info to episodes + for (s_num, e_num), video_files in season_episodes.items(): + _log("DEBUG", f"Episode S{s_num:02d}E{e_num:02d}: season_dir={season_num}, filename_season={s_num}") + if s_num == season_num: # Verify season matches directory + episodes[(s_num, e_num)] = video_files + _log("DEBUG", f"Added episode S{s_num:02d}E{e_num:02d} to processing list") + else: + _log("WARNING", f"Season mismatch: directory={season_num}, filename={s_num} for S{s_num:02d}E{e_num:02d}") + + _log("DEBUG", f"Total episodes found on disk: {len(episodes)}") + return episodes + + except Exception as e: + _log("ERROR", f"Exception in _find_episodes_on_disk: {e}") + return episodes + + def _extract_season_number(self, dirname: str) -> Optional[int]: + """Extract season number from directory name""" + match = re.search(r'[Ss]eason\s+(\d+)', dirname, re.IGNORECASE) + if match: + return int(match.group(1)) + return None + + def _process_episode_manual_scan(self, series_path: Path, imdb_id: str, + season_num: int, episode_num: int) -> bool: + """Process a single episode during manual scan""" + season_dir = series_path / f"Season {season_num:02d}" + if not season_dir.exists(): + # Try alternate format + season_dir = series_path / f"Season {season_num}" + if not season_dir.exists(): + _log("ERROR", f"Season directory not found for S{season_num:02d}E{episode_num:02d}") + return False + + _log("DEBUG", f"Processing episode S{season_num:02d}E{episode_num:02d}") + + # Step 1: Check for existing NFOGuard data + existing_nfo = self.episode_nfo_manager.find_nfo_for_episode(season_dir, season_num, episode_num) + if existing_nfo: + nfo_data = self.episode_nfo_manager.extract_nfoguard_data(existing_nfo) + if nfo_data: + # Verify against database + db_data = self.db.get_episode_date(imdb_id, season_num, episode_num) + if db_data and db_data.get("dateadded") == nfo_data.get("dateadded"): + _log("DEBUG", f"Episode S{season_num:02d}E{episode_num:02d} already up to date") + # Still migrate filename if needed + self.episode_nfo_manager.migrate_nfo_to_video_filename(season_dir, season_num, episode_num) + return True + + # Step 2: Check database + db_data = self.db.get_episode_date(imdb_id, season_num, episode_num) + if db_data and db_data.get("dateadded"): + _log("DEBUG", f"Using database data for S{season_num:02d}E{episode_num:02d}") + aired = db_data.get("aired") + dateadded = db_data.get("dateadded") + source = db_data.get("source", "database") + else: + # Step 3: Query Sonarr for episode data + aired, dateadded, source = self._get_episode_dates_from_sonarr(imdb_id, season_num, episode_num) + + # Step 4: Create/update NFO and database + if dateadded: + # Get episode metadata for title/plot + title, plot = self._get_episode_metadata_from_sonarr(imdb_id, season_num, episode_num) + + # Create/update NFO with video filename + success = self.episode_nfo_manager.create_episode_nfo( + season_dir, season_num, episode_num, aired, dateadded, source, title, plot + ) + + if success: + # Update database + self.db.upsert_episode_date(imdb_id, season_num, episode_num, aired, dateadded, source, True) + return True + + _log("WARNING", f"Could not get dates for episode S{season_num:02d}E{episode_num:02d}") + return False + + def _get_episode_dates_from_sonarr(self, imdb_id: str, season_num: int, episode_num: int) -> Tuple[Optional[str], Optional[str], str]: + """Get episode dates from Sonarr API with proper fallback""" + aired = None + dateadded = None + source = "no_data_found" + + if not self.sonarr.enabled: + _log("WARNING", "Sonarr not enabled, cannot get episode dates") + return aired, dateadded, source + + try: + # Find series in Sonarr using lookup endpoint first + series = self.sonarr.series_by_imdb(imdb_id) + if not series: + _log("WARNING", f"Series not found via Sonarr lookup for IMDb: {imdb_id}") + # Try direct method as fallback (slower but more reliable) + _log("DEBUG", f"Trying direct series search for IMDb: {imdb_id}") + series = self.sonarr.series_by_imdb_direct(imdb_id) + if not series: + _log("WARNING", f"Series not found via direct search either for IMDb: {imdb_id}") + # Fall through to external API fallback + + if series: + # Get episodes for series + episodes = self.sonarr.episodes_for_series(series["id"]) + target_episode = None + + for episode in episodes: + if (episode.get("seasonNumber") == season_num and + episode.get("episodeNumber") == episode_num): + target_episode = episode + break + + if not target_episode: + _log("WARNING", f"Episode S{season_num:02d}E{episode_num:02d} not found in Sonarr") + # Don't return here - fall through to external API fallback + else: + # Get airdate + aired = target_episode.get("airDateUtc") + + # Try to get import history + episode_id = target_episode.get("id") + if episode_id: + import_date = self.sonarr.get_episode_import_history(episode_id) + if import_date: + dateadded = convert_utc_to_local(import_date) + source = "sonarr:history.import" + _log("INFO", f"Found import date for S{season_num:02d}E{episode_num:02d}: {dateadded}") + return aired, dateadded, source + + # Fallback to airdate if no import history + if aired: + dateadded = convert_utc_to_local(aired) + source = "sonarr:episode.airDateUtc" + _log("WARNING", f"No import history for S{season_num:02d}E{episode_num:02d}, using airdate: {dateadded}") + return aired, dateadded, source + + except Exception as e: + _log("ERROR", f"Sonarr API error for S{season_num:02d}E{episode_num:02d}: {e}") + + # Try external APIs for episode airdate + _log("INFO", f"Trying external APIs for episode S{season_num:02d}E{episode_num:02d} airdate") + aired, source = self._get_episode_airdate_from_external_apis(imdb_id, season_num, episode_num) + if aired: + dateadded = convert_utc_to_local(aired) + return aired, dateadded, source + + _log("ERROR", f"Could not get any date information for S{season_num:02d}E{episode_num:02d}") + return aired, dateadded, source + + def _get_episode_metadata_from_sonarr(self, imdb_id: str, season_num: int, episode_num: int) -> Tuple[Optional[str], Optional[str]]: + """Get episode title and plot from Sonarr""" + if not self.sonarr.enabled: + return None, None + + try: + series = self.sonarr.series_by_imdb(imdb_id) + if not series: + # Try direct method as fallback + series = self.sonarr.series_by_imdb_direct(imdb_id) + + if series: + episodes = self.sonarr.episodes_for_series(series["id"]) + for episode in episodes: + if (episode.get("seasonNumber") == season_num and + episode.get("episodeNumber") == episode_num): + title = episode.get("title") + plot = episode.get("overview") + return title, plot + except Exception as e: + _log("DEBUG", f"Could not get metadata for S{season_num:02d}E{episode_num:02d}: {e}") + + return None, None + + def _get_episode_airdate_from_external_apis(self, imdb_id: str, season_num: int, episode_num: int) -> Tuple[Optional[str], str]: + """Get episode airdate from external APIs (TMDB, OMDb) as fallback""" + # Try TMDB first + if self.external_manager.tmdb.enabled: + try: + _log("DEBUG", f"Trying TMDB for episode S{season_num:02d}E{episode_num:02d} airdate") + + # First convert IMDb to TMDB TV ID + tv_search = self.external_manager.tmdb._get(f"/find/{imdb_id}", {"external_source": "imdb_id"}) + if tv_search and tv_search.get("tv_results"): + tv_id = tv_search["tv_results"][0].get("id") + if tv_id: + _log("DEBUG", f"Found TMDB TV ID {tv_id} for {imdb_id}") + + # Get episode details + episode_data = self.external_manager.tmdb._get(f"/tv/{tv_id}/season/{season_num}/episode/{episode_num}") + if episode_data and episode_data.get("air_date"): + airdate = episode_data["air_date"] + # Convert to ISO format with UTC timezone + iso_airdate = f"{airdate}T00:00:00Z" + _log("INFO", f"Found TMDB airdate for S{season_num:02d}E{episode_num:02d}: {iso_airdate}") + return iso_airdate, "tmdb:episode.air_date" + except Exception as e: + _log("WARNING", f"TMDB episode lookup failed for S{season_num:02d}E{episode_num:02d}: {e}") + + # Try OMDb as fallback + if self.external_manager.omdb.enabled: + try: + _log("DEBUG", f"Trying OMDb for episode S{season_num:02d}E{episode_num:02d} airdate") + episode_dates = self.external_manager.omdb.get_tv_season_episodes(imdb_id, season_num) + if episode_num in episode_dates: + airdate = episode_dates[episode_num] + # Convert to ISO format + from datetime import datetime, timezone + try: + # Try to parse OMDb date format (usually DD MMM YYYY) + dt = datetime.strptime(airdate, "%d %b %Y").replace(tzinfo=timezone.utc) + iso_airdate = dt.isoformat(timespec="seconds") + _log("INFO", f"Found OMDb airdate for S{season_num:02d}E{episode_num:02d}: {iso_airdate}") + return iso_airdate, "omdb:episode.released" + except ValueError: + # Try other common formats + for fmt in ["%Y-%m-%d", "%d %B %Y"]: + try: + dt = datetime.strptime(airdate, fmt).replace(tzinfo=timezone.utc) + iso_airdate = dt.isoformat(timespec="seconds") + _log("INFO", f"Found OMDb airdate for S{season_num:02d}E{episode_num:02d}: {iso_airdate}") + return iso_airdate, "omdb:episode.released" + except ValueError: + continue + except Exception as e: + _log("WARNING", f"OMDb episode lookup failed for S{season_num:02d}E{episode_num:02d}: {e}") + + _log("WARNING", f"No external API airdate found for S{season_num:02d}E{episode_num:02d}") + return None, "no_external_data" + + def _create_series_nfos(self, series_path: Path, imdb_id: str): + """Create tvshow.nfo and season.nfo files only if they don't exist""" + # Create tvshow.nfo only if it doesn't exist + tvshow_nfo = series_path / "tvshow.nfo" + if not tvshow_nfo.exists(): + self.nfo_manager.create_tvshow_nfo(series_path, imdb_id) + else: + _log("DEBUG", f"Skipping tvshow.nfo creation - already exists: {tvshow_nfo}") + + # Create season.nfo for each season directory only if they don't exist + for season_dir in series_path.iterdir(): + if season_dir.is_dir() and self._is_season_directory(season_dir.name): + season_num = self._extract_season_number(season_dir.name) + if season_num is not None: + season_nfo = season_dir / "season.nfo" + if not season_nfo.exists(): + self.nfo_manager.create_season_nfo(season_dir, season_num) + else: + _log("DEBUG", f"Skipping season.nfo creation - already exists: {season_nfo}") \ No newline at end of file