sbcrumb/nfoguard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

retry for the AI
Local Docker Build (Dev) / build-dev (push) Successful in 16s
Details

Browse Source
This commit is contained in:
SBCrumb
2025-09-26 15:19:37 -04:00
parent c19d0f6290
commit 2c4dd274ff
3 changed files with 90 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files
VERSION
+1 -1
View File
@@ -1 +1 @@
2.0.6-debug-episode-detection
2.1.0-preserve-long-nfo
core/nfo_manager.py
+60 -210
View File
@@ -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 preserve filename")
print(f"🔍 Found existing episode NFO: {nfo_file.name}")
return nfo_file
except ValueError:
continue
@@ -607,78 +499,61 @@ class NFOManager:
return None
def find_video_file_for_episode(self, season_dir: Path, season_num: int, episode_num: int) -> Optional[Path]:
"""Find video file that matches the given season and episode"""
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 None
return f"S{season_num:02d}E{episode_num:02d}.nfo"
season_episode_pattern = f"S{season_num:02d}E{episode_num:02d}"
video_extensions = [".mkv", ".mp4", ".avi", ".mov", ".m4v"]
# 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"]
# Look for video files that contain the season/episode pattern
for video_file in season_dir.iterdir():
if (video_file.is_file() and
video_file.suffix.lower() in video_extensions and
season_episode_pattern in video_file.name):
print(f"🎬 Found matching video file: {video_file.name}")
return video_file
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 and long filenames"""
# First, check for existing long-named NFO files (PRESERVE these)
existing_long_nfo = self.find_existing_episode_nfo(season_dir, season_num, episode_num)
# Check if there's a short-named NFO that needs to be converted to long name
short_episode_filename = f"S{season_num:02d}E{episode_num:02d}.nfo"
short_nfo_path = season_dir / short_episode_filename
if existing_long_nfo:
# PRESERVE the long filename - update it in place
nfo_path = existing_long_nfo
print(f"🎯 Updating existing NFO (preserving name): {existing_long_nfo.name}")
elif short_nfo_path.exists():
# Convert short name to long name by finding matching video file
matching_video = self.find_video_file_for_episode(season_dir, season_num, episode_num)
if matching_video:
# Create long NFO name based on video filename
video_name_without_ext = matching_video.stem
long_nfo_name = f"{video_name_without_ext}.nfo"
long_nfo_path = season_dir / long_nfo_name
print(f"📝 Converting short NFO to long name: {short_episode_filename} -> {long_nfo_name}")
# Move/rename the short NFO to the long name
try:
short_nfo_path.rename(long_nfo_path)
nfo_path = long_nfo_path
print(f"✅ Successfully renamed NFO to match video file")
except Exception as e:
print(f"⚠️ Could not rename NFO file: {e}. Using short name.")
nfo_path = short_nfo_path
else:
# No matching video found, use existing short name
nfo_path = short_nfo_path
print(f"⚠️ No matching video file found for {short_episode_filename}, keeping short name")
else:
# No existing NFO, create new one with long name if video exists
matching_video = self.find_video_file_for_episode(season_dir, season_num, episode_num)
if matching_video:
video_name_without_ext = matching_video.stem
long_nfo_name = f"{video_name_without_ext}.nfo"
nfo_path = season_dir / long_nfo_name
print(f"🆕 Creating new NFO with long name: {long_nfo_name}")
else:
# Fallback to short name if no video file found
nfo_path = short_nfo_path
print(f"🆕 Creating new NFO with short name: {short_episode_filename}")
"""Create or update episode NFO file preserving existing content"""
# 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:
# Load existing NFO file if it exists
# Check for existing NFO file at target location
source_nfo_path = nfo_path if nfo_path.exists() else None
if source_nfo_path:
@@ -690,16 +565,7 @@ class NFOManager:
if episode.tag != "episodedetails":
raise ValueError("Root element is not <episodedetails>")
# Show what content fields are being preserved (for debugging)
if self.debug:
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" 📄 Content preserved: {', '.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 = {}
@@ -748,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
@@ -794,20 +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}")
# NFO file created/updated successfully
pass
except Exception as e:
print(f"❌ Error creating/updating episode NFO {nfo_path}: {e}")
@@ -847,4 +698,3 @@ class NFOManager:
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}")
nfoguard.py
+22 -171
View File
@@ -30,7 +30,6 @@ from core.path_mapper import PathMapper
from clients.radarr_client import RadarrClient
from clients.sonarr_client import SonarrClient
from clients.external_clients import ExternalClientManager
from processors.tv_series_processor import TVSeriesProcessor
# ---------------------------
# Configuration & Logging
@@ -441,33 +440,10 @@ 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 to ensure proper dates
nfo_episodes_found += 1
if nfo_episodes_found > 0:
_log("INFO", f"Found {nfo_episodes_found} episodes with existing NFO files")
# Find missing episodes (not in cache)
# Find missing episodes
cached_keys = set(episode_dates.keys())
missing_keys = set(disk_episodes.keys()) - cached_keys
# Add episodes with existing NFOs to missing_keys so they get proper date lookup
for season_num, episode_num in disk_episodes.keys():
if (season_num, episode_num) not in cached_keys:
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:
missing_keys.add((season_num, episode_num))
if not missing_keys:
_log("INFO", "All episodes found in cache")
return episode_dates
@@ -575,7 +551,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:
@@ -630,21 +606,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)
@@ -661,14 +622,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
@@ -683,7 +642,7 @@ 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)
@@ -738,7 +697,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:
@@ -763,97 +721,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({
return {
"title": episode.get("title"),
"overview": episode.get("overview"),
"runtime": episode.get("runtime"),
"ratings": episode.get("ratings", {})
})
break
}
# 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]]:
@@ -970,7 +851,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:
@@ -1569,13 +1450,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"""
@@ -1629,23 +1509,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':
@@ -1726,17 +1594,9 @@ start_time = datetime.now(timezone.utc)
db = NFOGuardDatabase(config.db_path)
nfo_manager = NFOManager(config.manager_brand, config.debug)
path_mapper = PathMapper(config) # FIXED: Pass config to PathMapper
# Initialize Sonarr client using environment variables
sonarr = SonarrClient(
base_url=os.environ.get("SONARR_URL", ""),
api_key=os.environ.get("SONARR_API_KEY", "")
)
# Use new clean TV processor
tv_processor = TVSeriesProcessor(db, nfo_manager, path_mapper, sonarr)
tv_processor = TVProcessor(db, nfo_manager, path_mapper)
movie_processor = MovieProcessor(db, nfo_manager, path_mapper)
batcher = WebhookBatcher(nfo_manager)
batcher = WebhookBatcher()
# ---------------------------
# Webhook Handlers
@@ -1849,18 +1709,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 = {
@@ -2208,11 +2059,11 @@ 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 with new clean processor
# 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_manual_scan(item)
tv_processor.process_series(item)
except Exception as e:
_log("ERROR", f"Failed processing TV series {item}: {e}")