dev to main (#14)
Local Docker Build (Main) / build (push) Successful in 20s
Local Docker Build (Main) / deploy (push) Successful in 0s

Reviewed-on: #14
Co-authored-by: SBCrumb <sbcrumb@kickthetree.com>
Co-committed-by: SBCrumb <sbcrumb@kickthetree.com>
This commit is contained in:
2025-09-27 11:19:36 -04:00
committed by sbcrumb
parent a4c333aa85
commit 2e7922dc8b
5 changed files with 822 additions and 378 deletions
+276
View File
@@ -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 <episodedetails>")
# 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
+70 -176
View File
@@ -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 <episodedetails>")
# 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'<?xml version="1.0" encoding="utf-8"?>\n<!--{comment_text}-->\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}")