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