refactor: update and remove anything to do with NFO files
Local Docker Build (Dev) / build-dev (push) Successful in 35s
Local Docker Build (Dev) / build-dev (push) Successful in 35s
moving to an all DB approach since radarr overwrites .nfo files
This commit is contained in:
@@ -1,473 +0,0 @@
|
||||
"""
|
||||
Async NFO Manager for NFOGuard
|
||||
High-performance async NFO file operations with concurrent processing
|
||||
"""
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from typing import Optional, List, Dict, Any, Tuple
|
||||
from datetime import datetime
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
from utils.logging import _log
|
||||
from utils.async_file_utils import (
|
||||
async_read_nfo_file,
|
||||
async_write_nfo_file,
|
||||
async_file_exists,
|
||||
async_set_file_mtime,
|
||||
async_batch_nfo_operations,
|
||||
async_concurrent_episode_processing,
|
||||
aiofiles
|
||||
)
|
||||
from utils.nfo_patterns import (
|
||||
create_basic_nfo_structure,
|
||||
extract_imdb_from_nfo_content,
|
||||
extract_dates_from_nfo,
|
||||
extract_imdb_id_from_text
|
||||
)
|
||||
from utils.validation import validate_date_string
|
||||
from utils.file_utils import VIDEO_EXTENSIONS
|
||||
|
||||
|
||||
class AsyncNFOManager:
|
||||
"""Async NFO file manager with concurrent processing capabilities"""
|
||||
|
||||
def __init__(self, manager_brand: str = "NFOGuard", debug: bool = False):
|
||||
self.manager_brand = manager_brand
|
||||
self.debug = debug
|
||||
|
||||
async def _async_get_target_nfo_path(self, season_dir: Path, season: int, episode: int) -> Path:
|
||||
"""
|
||||
Get the target NFO path using video filename matching to prevent concatenation
|
||||
|
||||
Args:
|
||||
season_dir: Path to season directory
|
||||
season: Season number
|
||||
episode: Episode number
|
||||
|
||||
Returns:
|
||||
Path to target NFO file (video-matching preferred, generic fallback)
|
||||
"""
|
||||
try:
|
||||
# Find video files in the season directory
|
||||
video_files = []
|
||||
if await async_file_exists(season_dir):
|
||||
import re
|
||||
try:
|
||||
entries = await aiofiles.os.listdir(season_dir)
|
||||
for entry in entries:
|
||||
entry_path = season_dir / entry
|
||||
if await aiofiles.os.path.isfile(entry_path):
|
||||
if entry_path.suffix.lower() in VIDEO_EXTENSIONS:
|
||||
# Parse episode info from filename
|
||||
match = re.search(r'[Ss](\d{1,2})[Ee](\d{1,2})', entry)
|
||||
if match:
|
||||
s, e = int(match.group(1)), int(match.group(2))
|
||||
if s == season and e == episode:
|
||||
# Found matching video file - use its name for NFO
|
||||
target_nfo = season_dir / f"{entry_path.stem}.nfo"
|
||||
if self.debug:
|
||||
_log("DEBUG", f"Video-matching NFO path: {target_nfo.name}")
|
||||
return target_nfo
|
||||
except Exception as e:
|
||||
if self.debug:
|
||||
_log("WARNING", f"Failed to scan season directory {season_dir}: {e}")
|
||||
|
||||
# Fallback to generic filename if no matching video found
|
||||
target_nfo = season_dir / f"S{season:02d}E{episode:02d}.nfo"
|
||||
if self.debug:
|
||||
_log("WARNING", f"No video file found for S{season:02d}E{episode:02d}, using generic: {target_nfo.name}")
|
||||
return target_nfo
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Failed to determine NFO path for S{season:02d}E{episode:02d}: {e}")
|
||||
# Emergency fallback
|
||||
return season_dir / f"S{season:02d}E{episode:02d}.nfo"
|
||||
|
||||
async def async_parse_imdb_from_path(self, path: Path) -> Optional[str]:
|
||||
"""
|
||||
Async extract IMDb ID from directory path or filename
|
||||
|
||||
Args:
|
||||
path: Path to examine
|
||||
|
||||
Returns:
|
||||
IMDb ID if found, None otherwise
|
||||
"""
|
||||
# Use the sync version since it's just string processing
|
||||
return extract_imdb_id_from_text(str(path))
|
||||
|
||||
async def async_parse_imdb_from_nfo(self, nfo_path: Path) -> Optional[str]:
|
||||
"""
|
||||
Async extract IMDb ID from NFO file content
|
||||
|
||||
Args:
|
||||
nfo_path: Path to NFO file
|
||||
|
||||
Returns:
|
||||
IMDb ID if found, None otherwise
|
||||
"""
|
||||
root = await async_read_nfo_file(nfo_path)
|
||||
if root is None:
|
||||
return None
|
||||
|
||||
return extract_imdb_from_nfo_content(root)
|
||||
|
||||
async def async_find_movie_imdb_id(self, movie_dir: Path) -> Optional[str]:
|
||||
"""
|
||||
Async find IMDb ID from directory name, filenames, or NFO file
|
||||
|
||||
Args:
|
||||
movie_dir: Path to movie directory
|
||||
|
||||
Returns:
|
||||
IMDb ID if found, None otherwise
|
||||
"""
|
||||
if self.debug:
|
||||
_log("DEBUG", f"Async searching for IMDb ID in: {movie_dir.name}")
|
||||
|
||||
# First try directory name
|
||||
imdb_id = await self.async_parse_imdb_from_path(movie_dir)
|
||||
if imdb_id:
|
||||
if self.debug:
|
||||
_log("DEBUG", f"Found IMDb ID in directory name: {imdb_id}")
|
||||
return imdb_id
|
||||
|
||||
# Try all files in the directory concurrently
|
||||
if await async_file_exists(movie_dir):
|
||||
try:
|
||||
from utils.async_file_utils import aiofiles
|
||||
entries = await aiofiles.os.listdir(movie_dir)
|
||||
|
||||
# Create tasks to check all files
|
||||
async def check_file(filename: str) -> Optional[str]:
|
||||
file_path = movie_dir / filename
|
||||
if await aiofiles.os.path.isfile(file_path):
|
||||
return await self.async_parse_imdb_from_path(file_path)
|
||||
return None
|
||||
|
||||
# Check all files concurrently
|
||||
results = await asyncio.gather(
|
||||
*[check_file(filename) for filename in entries],
|
||||
return_exceptions=True
|
||||
)
|
||||
|
||||
# Find first valid IMDb ID
|
||||
for result in results:
|
||||
if isinstance(result, str) and result:
|
||||
if self.debug:
|
||||
_log("DEBUG", f"Found IMDb ID in filename: {result}")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Failed to scan directory {movie_dir}: {e}")
|
||||
|
||||
# Finally, try NFO file content
|
||||
nfo_path = movie_dir / "movie.nfo"
|
||||
imdb_id = await self.async_parse_imdb_from_nfo(nfo_path)
|
||||
if imdb_id:
|
||||
if self.debug:
|
||||
_log("DEBUG", f"Found IMDb ID in NFO file: {imdb_id}")
|
||||
return imdb_id
|
||||
|
||||
if self.debug:
|
||||
_log("DEBUG", f"No IMDb ID found for: {movie_dir.name}")
|
||||
return None
|
||||
|
||||
async def async_create_movie_nfo(
|
||||
self,
|
||||
movie_dir: Path,
|
||||
imdb_id: str,
|
||||
dateadded: str,
|
||||
premiered: Optional[str] = None,
|
||||
lock_metadata: bool = True
|
||||
) -> bool:
|
||||
"""
|
||||
Async create movie NFO file
|
||||
|
||||
Args:
|
||||
movie_dir: Path to movie directory
|
||||
imdb_id: IMDb ID
|
||||
dateadded: Date added
|
||||
premiered: Optional premiere date
|
||||
lock_metadata: Whether to lock metadata
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
nfo_path = movie_dir / "movie.nfo"
|
||||
|
||||
# Prepare dates
|
||||
dates = {"dateadded": dateadded}
|
||||
if premiered and validate_date_string(premiered):
|
||||
dates["premiered"] = premiered
|
||||
|
||||
# Create NFO structure
|
||||
root = create_basic_nfo_structure(
|
||||
media_type="movie",
|
||||
title=movie_dir.name,
|
||||
imdb_id=imdb_id,
|
||||
dates=dates,
|
||||
additional_fields={"source": self.manager_brand}
|
||||
)
|
||||
|
||||
# Write NFO file asynchronously
|
||||
success = await async_write_nfo_file(nfo_path, root, lock_metadata)
|
||||
|
||||
if success and self.debug:
|
||||
_log("DEBUG", f"Created movie NFO: {nfo_path}")
|
||||
|
||||
return success
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Failed to create movie NFO for {movie_dir}: {e}")
|
||||
return False
|
||||
|
||||
async def async_create_episode_nfo(
|
||||
self,
|
||||
season_dir: Path,
|
||||
season: int,
|
||||
episode: int,
|
||||
aired: Optional[str] = None,
|
||||
dateadded: Optional[str] = None,
|
||||
source: str = "unknown",
|
||||
lock_metadata: bool = True
|
||||
) -> bool:
|
||||
"""
|
||||
Async create episode NFO file with proper video filename matching
|
||||
|
||||
Args:
|
||||
season_dir: Path to season directory
|
||||
season: Season number
|
||||
episode: Episode number
|
||||
aired: Optional air date
|
||||
dateadded: Optional date added
|
||||
source: Source of the data
|
||||
lock_metadata: Whether to lock metadata
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
# Use proper video filename matching to prevent NFO concatenation
|
||||
nfo_path = await self._async_get_target_nfo_path(season_dir, season, episode)
|
||||
|
||||
# Prepare dates
|
||||
dates = {}
|
||||
if aired and validate_date_string(aired):
|
||||
dates["aired"] = aired
|
||||
if dateadded and validate_date_string(dateadded):
|
||||
dates["dateadded"] = dateadded
|
||||
|
||||
# Create NFO structure
|
||||
root = create_basic_nfo_structure(
|
||||
media_type="episodedetails",
|
||||
title=f"S{season:02d}E{episode:02d}",
|
||||
dates=dates,
|
||||
additional_fields={
|
||||
"season": str(season),
|
||||
"episode": str(episode),
|
||||
"source": source
|
||||
}
|
||||
)
|
||||
|
||||
# Write NFO file asynchronously
|
||||
success = await async_write_nfo_file(nfo_path, root, lock_metadata)
|
||||
|
||||
if success and self.debug:
|
||||
_log("DEBUG", f"Created episode NFO: {nfo_path}")
|
||||
|
||||
return success
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Failed to create episode NFO S{season:02d}E{episode:02d}: {e}")
|
||||
return False
|
||||
|
||||
async def async_batch_create_episode_nfos(
|
||||
self,
|
||||
episode_data_list: List[Dict[str, Any]],
|
||||
max_concurrent: int = 5
|
||||
) -> List[bool]:
|
||||
"""
|
||||
Batch create multiple episode NFOs concurrently
|
||||
|
||||
Args:
|
||||
episode_data_list: List of episode data dictionaries
|
||||
max_concurrent: Maximum concurrent NFO operations
|
||||
|
||||
Returns:
|
||||
List of success/failure results
|
||||
"""
|
||||
async def _create_episode_nfo(episode_data: Dict[str, Any]) -> bool:
|
||||
return await self.async_create_episode_nfo(
|
||||
season_dir=episode_data.get('season_dir'),
|
||||
season=episode_data.get('season'),
|
||||
episode=episode_data.get('episode'),
|
||||
aired=episode_data.get('aired'),
|
||||
dateadded=episode_data.get('dateadded'),
|
||||
source=episode_data.get('source', 'unknown'),
|
||||
lock_metadata=episode_data.get('lock_metadata', True)
|
||||
)
|
||||
|
||||
return await async_concurrent_episode_processing(
|
||||
episode_data_list,
|
||||
_create_episode_nfo,
|
||||
max_concurrent
|
||||
)
|
||||
|
||||
async def async_set_file_mtime(self, file_path: Path, date_str: str) -> bool:
|
||||
"""
|
||||
Async set file modification time from date string
|
||||
|
||||
Args:
|
||||
file_path: Path to file
|
||||
date_str: Date string in ISO format
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
if not validate_date_string(date_str):
|
||||
_log("WARNING", f"Invalid date format for mtime: {date_str}")
|
||||
return False
|
||||
|
||||
# Parse date string to timestamp
|
||||
dt = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
|
||||
timestamp = dt.timestamp()
|
||||
|
||||
# Set file mtime asynchronously
|
||||
success = await async_set_file_mtime(file_path, timestamp)
|
||||
|
||||
if success and self.debug:
|
||||
_log("DEBUG", f"Set mtime for {file_path}: {date_str}")
|
||||
|
||||
return success
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Failed to set mtime for {file_path}: {e}")
|
||||
return False
|
||||
|
||||
async def async_batch_set_file_mtimes(
|
||||
self,
|
||||
file_mtime_pairs: List[Tuple[Path, str]],
|
||||
max_concurrent: int = 10
|
||||
) -> List[bool]:
|
||||
"""
|
||||
Batch set file modification times concurrently
|
||||
|
||||
Args:
|
||||
file_mtime_pairs: List of (file_path, date_str) tuples
|
||||
max_concurrent: Maximum concurrent operations
|
||||
|
||||
Returns:
|
||||
List of success/failure results
|
||||
"""
|
||||
async def _set_single_mtime(file_path: Path, date_str: str) -> bool:
|
||||
return await self.async_set_file_mtime(file_path, date_str)
|
||||
|
||||
# Create tasks for all mtime operations
|
||||
semaphore = asyncio.Semaphore(max_concurrent)
|
||||
|
||||
async def _set_mtime_with_semaphore(file_path: Path, date_str: str) -> bool:
|
||||
async with semaphore:
|
||||
return await _set_single_mtime(file_path, date_str)
|
||||
|
||||
tasks = [
|
||||
_set_mtime_with_semaphore(file_path, date_str)
|
||||
for file_path, date_str in file_mtime_pairs
|
||||
]
|
||||
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
return [result if isinstance(result, bool) else False for result in results]
|
||||
|
||||
async def async_validate_nfo_integrity(
|
||||
self,
|
||||
nfo_paths: List[Path],
|
||||
max_concurrent: int = 10
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Async validate integrity of multiple NFO files
|
||||
|
||||
Args:
|
||||
nfo_paths: List of NFO file paths to validate
|
||||
max_concurrent: Maximum concurrent validations
|
||||
|
||||
Returns:
|
||||
Dictionary with validation results and statistics
|
||||
"""
|
||||
results = {
|
||||
'total_files': len(nfo_paths),
|
||||
'valid_files': 0,
|
||||
'invalid_files': 0,
|
||||
'missing_files': 0,
|
||||
'validation_errors': [],
|
||||
'file_results': {}
|
||||
}
|
||||
|
||||
async def _validate_single_nfo(nfo_path: Path) -> Dict[str, Any]:
|
||||
file_result = {
|
||||
'path': str(nfo_path),
|
||||
'exists': False,
|
||||
'valid_xml': False,
|
||||
'has_imdb_id': False,
|
||||
'has_dates': False,
|
||||
'error': None
|
||||
}
|
||||
|
||||
try:
|
||||
# Check if file exists
|
||||
if not await async_file_exists(nfo_path):
|
||||
file_result['error'] = 'File does not exist'
|
||||
return file_result
|
||||
|
||||
file_result['exists'] = True
|
||||
|
||||
# Try to parse NFO
|
||||
root = await async_read_nfo_file(nfo_path)
|
||||
if root is None:
|
||||
file_result['error'] = 'Failed to parse XML'
|
||||
return file_result
|
||||
|
||||
file_result['valid_xml'] = True
|
||||
|
||||
# Check for IMDb ID
|
||||
imdb_id = extract_imdb_from_nfo_content(root)
|
||||
file_result['has_imdb_id'] = bool(imdb_id)
|
||||
|
||||
# Check for dates
|
||||
dates = extract_dates_from_nfo(root)
|
||||
file_result['has_dates'] = any(dates.values())
|
||||
|
||||
except Exception as e:
|
||||
file_result['error'] = str(e)
|
||||
|
||||
return file_result
|
||||
|
||||
# Validate all files concurrently
|
||||
semaphore = asyncio.Semaphore(max_concurrent)
|
||||
|
||||
async def _validate_with_semaphore(nfo_path: Path) -> Dict[str, Any]:
|
||||
async with semaphore:
|
||||
return await _validate_single_nfo(nfo_path)
|
||||
|
||||
tasks = [_validate_with_semaphore(nfo_path) for nfo_path in nfo_paths]
|
||||
file_results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
# Process results
|
||||
for i, result in enumerate(file_results):
|
||||
if isinstance(result, dict):
|
||||
path_str = str(nfo_paths[i])
|
||||
results['file_results'][path_str] = result
|
||||
|
||||
if not result['exists']:
|
||||
results['missing_files'] += 1
|
||||
elif result['valid_xml']:
|
||||
results['valid_files'] += 1
|
||||
else:
|
||||
results['invalid_files'] += 1
|
||||
|
||||
if result.get('error'):
|
||||
results['validation_errors'].append(f"{path_str}: {result['error']}")
|
||||
|
||||
return results
|
||||
@@ -0,0 +1,267 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Database Populator for NFOGuard
|
||||
Bulk populates the NFOGuard database from Radarr/Sonarr
|
||||
Phase 4: Replace NFO-based initial population with direct DB/API queries
|
||||
"""
|
||||
import time
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
from datetime import datetime
|
||||
|
||||
from core.database import NFOGuardDatabase
|
||||
from clients.radarr_client import RadarrClient
|
||||
from clients.sonarr_client import SonarrClient
|
||||
from utils.logging import _log
|
||||
|
||||
|
||||
class DatabasePopulator:
|
||||
"""Populates NFOGuard database from Radarr/Sonarr sources"""
|
||||
|
||||
def __init__(self, db: NFOGuardDatabase, radarr_client: RadarrClient, sonarr_client: SonarrClient):
|
||||
self.db = db
|
||||
self.radarr = radarr_client
|
||||
self.sonarr = sonarr_client
|
||||
|
||||
def populate_movies(self) -> Dict[str, any]:
|
||||
"""
|
||||
Populate movies from Radarr database/API
|
||||
|
||||
Returns:
|
||||
Dictionary with statistics: {
|
||||
'total': int,
|
||||
'added': int,
|
||||
'updated': int,
|
||||
'skipped': int,
|
||||
'errors': int,
|
||||
'duration': float
|
||||
}
|
||||
"""
|
||||
_log("INFO", "Starting movie population from Radarr")
|
||||
start_time = time.time()
|
||||
|
||||
stats = {
|
||||
'total': 0,
|
||||
'added': 0,
|
||||
'updated': 0,
|
||||
'skipped': 0,
|
||||
'errors': 0,
|
||||
'duration': 0.0
|
||||
}
|
||||
|
||||
try:
|
||||
# Get all movies from Radarr
|
||||
movies = self.radarr.get_all_movies()
|
||||
if not movies:
|
||||
_log("WARNING", "No movies found in Radarr")
|
||||
return stats
|
||||
|
||||
stats['total'] = len(movies)
|
||||
_log("INFO", f"Found {stats['total']} movies in Radarr")
|
||||
|
||||
# Process each movie
|
||||
for movie in movies:
|
||||
try:
|
||||
imdb_id = movie.get('imdbId')
|
||||
if not imdb_id:
|
||||
_log("DEBUG", f"Skipping movie without IMDb ID: {movie.get('title')}")
|
||||
stats['skipped'] += 1
|
||||
continue
|
||||
|
||||
# Check if movie already exists in database
|
||||
existing = self.db.get_movie_dates(imdb_id)
|
||||
if existing and existing.get('dateadded'):
|
||||
_log("DEBUG", f"Movie {imdb_id} already in database, skipping")
|
||||
stats['skipped'] += 1
|
||||
continue
|
||||
|
||||
# Get movie path
|
||||
path = movie.get('path', '')
|
||||
|
||||
# Get release date
|
||||
released = None
|
||||
if movie.get('digitalRelease'):
|
||||
released = movie.get('digitalRelease')
|
||||
source_type = 'radarr:digital'
|
||||
elif movie.get('physicalRelease'):
|
||||
released = movie.get('physicalRelease')
|
||||
source_type = 'radarr:physical'
|
||||
elif movie.get('inCinemas'):
|
||||
released = movie.get('inCinemas')
|
||||
source_type = 'radarr:theatrical'
|
||||
else:
|
||||
source_type = 'radarr:unknown'
|
||||
|
||||
# Get import date from Radarr history
|
||||
import_date = self.radarr.get_movie_import_date(imdb_id)
|
||||
if import_date:
|
||||
dateadded = import_date
|
||||
source = 'radarr:import_history'
|
||||
elif released:
|
||||
# Use release date as fallback
|
||||
dateadded = released
|
||||
source = f'{source_type}_fallback'
|
||||
else:
|
||||
_log("DEBUG", f"No date available for movie {imdb_id}, skipping")
|
||||
stats['skipped'] += 1
|
||||
continue
|
||||
|
||||
# Insert into database
|
||||
self.db.upsert_movie_dates(imdb_id, released, dateadded, source, has_video_file=True)
|
||||
stats['added'] += 1
|
||||
_log("DEBUG", f"Added movie {imdb_id}: {movie.get('title')} (source: {source})")
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Error processing movie {movie.get('title', 'unknown')}: {e}")
|
||||
stats['errors'] += 1
|
||||
continue
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Error during movie population: {e}")
|
||||
stats['errors'] += 1
|
||||
|
||||
stats['duration'] = time.time() - start_time
|
||||
_log("INFO", f"Movie population complete: {stats['added']} added, {stats['skipped']} skipped, {stats['errors']} errors in {stats['duration']:.2f}s")
|
||||
return stats
|
||||
|
||||
def populate_tv_episodes(self) -> Dict[str, any]:
|
||||
"""
|
||||
Populate TV episodes from Sonarr API
|
||||
|
||||
Returns:
|
||||
Dictionary with statistics: {
|
||||
'total_series': int,
|
||||
'total_episodes': int,
|
||||
'added': int,
|
||||
'updated': int,
|
||||
'skipped': int,
|
||||
'errors': int,
|
||||
'duration': float
|
||||
}
|
||||
"""
|
||||
_log("INFO", "Starting TV episode population from Sonarr")
|
||||
start_time = time.time()
|
||||
|
||||
stats = {
|
||||
'total_series': 0,
|
||||
'total_episodes': 0,
|
||||
'added': 0,
|
||||
'updated': 0,
|
||||
'skipped': 0,
|
||||
'errors': 0,
|
||||
'duration': 0.0
|
||||
}
|
||||
|
||||
try:
|
||||
# Get all series from Sonarr
|
||||
all_series = self.sonarr.get_all_series()
|
||||
if not all_series:
|
||||
_log("WARNING", "No series found in Sonarr")
|
||||
return stats
|
||||
|
||||
stats['total_series'] = len(all_series)
|
||||
_log("INFO", f"Found {stats['total_series']} series in Sonarr")
|
||||
|
||||
# Process each series
|
||||
for series in all_series:
|
||||
try:
|
||||
imdb_id = series.get('imdbId')
|
||||
if not imdb_id:
|
||||
_log("DEBUG", f"Skipping series without IMDb ID: {series.get('title')}")
|
||||
continue
|
||||
|
||||
series_id = series.get('id')
|
||||
series_path = series.get('path', '')
|
||||
series_title = series.get('title', 'Unknown')
|
||||
|
||||
# Update series record
|
||||
self.db.upsert_series(imdb_id, series_path)
|
||||
|
||||
# Get all episodes for this series
|
||||
episodes = self.sonarr.episodes_for_series(series_id)
|
||||
if not episodes:
|
||||
continue
|
||||
|
||||
_log("DEBUG", f"Processing {len(episodes)} episodes for {series_title}")
|
||||
|
||||
# Process each episode
|
||||
for episode in episodes:
|
||||
try:
|
||||
season_num = episode.get('seasonNumber', 0)
|
||||
episode_num = episode.get('episodeNumber', 0)
|
||||
|
||||
if season_num < 0 or episode_num <= 0:
|
||||
continue
|
||||
|
||||
stats['total_episodes'] += 1
|
||||
|
||||
# Check if episode already exists
|
||||
existing = self.db.get_episode_date(imdb_id, season_num, episode_num)
|
||||
if existing and existing.get('dateadded'):
|
||||
stats['skipped'] += 1
|
||||
continue
|
||||
|
||||
# Get air date
|
||||
aired = episode.get('airDate')
|
||||
|
||||
# Get import date from history
|
||||
dateadded = None
|
||||
episode_id = episode.get('id')
|
||||
if episode_id and episode.get('hasFile'):
|
||||
import_date = self.sonarr.get_episode_import_history(episode_id)
|
||||
if import_date:
|
||||
dateadded = import_date
|
||||
source = 'sonarr:import_history'
|
||||
|
||||
# Fallback to air date if no import date
|
||||
if not dateadded and aired:
|
||||
dateadded = aired
|
||||
source = 'sonarr:aired_fallback'
|
||||
elif not dateadded:
|
||||
# No date available
|
||||
stats['skipped'] += 1
|
||||
continue
|
||||
|
||||
# Insert into database
|
||||
has_file = episode.get('hasFile', False)
|
||||
self.db.upsert_episode_date(imdb_id, season_num, episode_num, aired, dateadded, source, has_file)
|
||||
stats['added'] += 1
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Error processing episode S{season_num:02d}E{episode_num:02d} of {series_title}: {e}")
|
||||
stats['errors'] += 1
|
||||
continue
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Error processing series {series.get('title', 'unknown')}: {e}")
|
||||
stats['errors'] += 1
|
||||
continue
|
||||
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Error during TV episode population: {e}")
|
||||
stats['errors'] += 1
|
||||
|
||||
stats['duration'] = time.time() - start_time
|
||||
_log("INFO", f"TV episode population complete: {stats['added']} added, {stats['skipped']} skipped, {stats['errors']} errors in {stats['duration']:.2f}s")
|
||||
return stats
|
||||
|
||||
def populate_all(self) -> Dict[str, any]:
|
||||
"""
|
||||
Populate both movies and TV episodes
|
||||
|
||||
Returns:
|
||||
Combined statistics dictionary
|
||||
"""
|
||||
_log("INFO", "Starting full database population")
|
||||
start_time = time.time()
|
||||
|
||||
movie_stats = self.populate_movies()
|
||||
tv_stats = self.populate_tv_episodes()
|
||||
|
||||
combined_stats = {
|
||||
'movies': movie_stats,
|
||||
'tv': tv_stats,
|
||||
'total_duration': time.time() - start_time
|
||||
}
|
||||
|
||||
_log("INFO", f"Full database population complete in {combined_stats['total_duration']:.2f}s")
|
||||
return combined_stats
|
||||
@@ -1,276 +0,0 @@
|
||||
#!/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 at the bottom
|
||||
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
|
||||
@@ -1,890 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
NFO Manager for creating and managing metadata files
|
||||
Handles NFO creation for movies, TV shows, seasons, and episodes
|
||||
"""
|
||||
import os
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from typing import Optional, Dict, Any, Tuple
|
||||
import re
|
||||
|
||||
|
||||
class NFOManager:
|
||||
"""Manages NFO file creation and updates"""
|
||||
|
||||
def __init__(self, manager_brand: str = "NFOGuard", debug: bool = False):
|
||||
self.manager_brand = manager_brand
|
||||
self.debug = debug
|
||||
|
||||
def parse_imdb_from_path(self, path: Path) -> Optional[str]:
|
||||
"""Extract IMDb ID from directory path or filename"""
|
||||
# Look for various IMDb patterns in both directory and file names
|
||||
path_str = str(path).lower()
|
||||
|
||||
# Try [imdb-ttXXXXXXX] format first (most explicit)
|
||||
match = re.search(r'\[imdb-?(tt\d+)\]', path_str)
|
||||
if match:
|
||||
return match.group(1)
|
||||
|
||||
# Try standalone [ttXXXXXXX] format in brackets
|
||||
match = re.search(r'\[(tt\d+)\]', path_str)
|
||||
if match:
|
||||
return match.group(1)
|
||||
|
||||
# Try {imdb-ttXXXXXXX} format with curly braces
|
||||
match = re.search(r'\{imdb-?(tt\d+)\}', path_str)
|
||||
if match:
|
||||
return match.group(1)
|
||||
|
||||
# Try (imdb-ttXXXXXXX) format with parentheses
|
||||
match = re.search(r'\(imdb-?(tt\d+)\)', path_str)
|
||||
if match:
|
||||
return match.group(1)
|
||||
|
||||
# Try ttXXXXXXX at end of filename/dirname (common pattern)
|
||||
match = re.search(r'[-_\s](tt\d+)$', path_str)
|
||||
if match:
|
||||
return match.group(1)
|
||||
|
||||
return None
|
||||
|
||||
def parse_imdb_from_path_with_nfo_fallback(self, path: Path, sonarr_client=None, shutdown_event=None) -> Optional[str]:
|
||||
"""
|
||||
Enhanced IMDb detection that fallback to Sonarr ID lookup from NFO files
|
||||
|
||||
1. First try to parse IMDb ID from directory path/name
|
||||
2. If not found, scan NFO files in directory for Sonarr series ID
|
||||
3. Use Sonarr API to lookup series and extract IMDb ID
|
||||
"""
|
||||
# Primary: Try directory name first
|
||||
imdb_id = self.parse_imdb_from_path(path)
|
||||
if imdb_id:
|
||||
return imdb_id
|
||||
|
||||
# Check for shutdown signal before expensive operations
|
||||
if shutdown_event and shutdown_event.is_set():
|
||||
return None
|
||||
|
||||
# Fallback: Check NFO files for Sonarr series ID
|
||||
if not sonarr_client or not sonarr_client.enabled:
|
||||
return None
|
||||
|
||||
try:
|
||||
# Look for episode NFO files in the directory (and subdirectories)
|
||||
nfo_files = []
|
||||
if path.is_dir():
|
||||
# Check current directory and season subdirectories
|
||||
nfo_files.extend(list(path.glob("*.nfo")))
|
||||
for subdir in path.iterdir():
|
||||
if subdir.is_dir() and subdir.name.lower().startswith('season'):
|
||||
nfo_files.extend(list(subdir.glob("*.nfo")))
|
||||
|
||||
# Extract Sonarr series ID from any NFO file
|
||||
for nfo_file in nfo_files:
|
||||
# Check for shutdown signal during NFO processing
|
||||
if shutdown_event and shutdown_event.is_set():
|
||||
return None
|
||||
|
||||
try:
|
||||
tree = ET.parse(nfo_file)
|
||||
root = tree.getroot()
|
||||
|
||||
# Look for Sonarr series ID
|
||||
for uniqueid in root.findall('.//uniqueid[@type="sonarr"]'):
|
||||
sonarr_id = uniqueid.text
|
||||
if sonarr_id and sonarr_id.isdigit():
|
||||
# Check for shutdown signal before API call
|
||||
if shutdown_event and shutdown_event.is_set():
|
||||
return None
|
||||
|
||||
# Look up series in Sonarr to get IMDb ID
|
||||
series_data = sonarr_client.get_series_by_id(int(sonarr_id))
|
||||
if series_data:
|
||||
imdb_id = series_data.get('imdbId')
|
||||
if imdb_id:
|
||||
return imdb_id.replace('tt', '') if imdb_id.startswith('tt') else imdb_id
|
||||
|
||||
except Exception as e:
|
||||
continue # Skip this NFO file and try the next one
|
||||
|
||||
except Exception as e:
|
||||
pass # Fallback failed, return None
|
||||
|
||||
return None
|
||||
|
||||
def parse_imdb_from_nfo(self, nfo_path: Path) -> Optional[str]:
|
||||
"""Extract IMDb ID from NFO file content"""
|
||||
if not nfo_path.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
root = self._parse_nfo_with_tolerance(nfo_path)
|
||||
|
||||
# Check for <uniqueid type="imdb">ttXXXXXX</uniqueid>
|
||||
imdb_uniqueid = root.find('.//uniqueid[@type="imdb"]')
|
||||
if imdb_uniqueid is not None and imdb_uniqueid.text:
|
||||
imdb_id = imdb_uniqueid.text.strip()
|
||||
if imdb_id.startswith('tt'):
|
||||
return imdb_id
|
||||
|
||||
# Check for legacy <imdbid>ttXXXXXX</imdbid>
|
||||
imdbid_elem = root.find('.//imdbid')
|
||||
if imdbid_elem is not None and imdbid_elem.text:
|
||||
imdb_id = imdbid_elem.text.strip()
|
||||
if imdb_id.startswith('tt'):
|
||||
return imdb_id
|
||||
|
||||
# Check for legacy <imdb>ttXXXXXX</imdb>
|
||||
imdb_elem = root.find('.//imdb')
|
||||
if imdb_elem is not None and imdb_elem.text:
|
||||
imdb_id = imdb_elem.text.strip()
|
||||
if imdb_id.startswith('tt'):
|
||||
return imdb_id
|
||||
|
||||
# Last resort: Check for TMDB ID as fallback identifier
|
||||
# This handles movies that only have TMDB IDs in NFO files
|
||||
tmdb_uniqueid = root.find('.//uniqueid[@type="tmdb"]')
|
||||
if tmdb_uniqueid is not None and tmdb_uniqueid.text:
|
||||
tmdb_id = tmdb_uniqueid.text.strip()
|
||||
if tmdb_id.isdigit():
|
||||
print(f"⚠️ Found TMDB ID {tmdb_id} but no IMDb ID - using TMDB ID as fallback")
|
||||
# Return TMDB ID with prefix to distinguish from IMDb IDs
|
||||
return f"tmdb-{tmdb_id}"
|
||||
|
||||
except (ET.ParseError, Exception):
|
||||
# Skip corrupted or non-XML files
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
def find_movie_imdb_id(self, movie_dir: Path) -> Optional[str]:
|
||||
"""Find IMDb ID from directory name, filenames, or NFO file"""
|
||||
# First try directory name
|
||||
imdb_id = self.parse_imdb_from_path(movie_dir)
|
||||
if imdb_id:
|
||||
return imdb_id
|
||||
|
||||
# Try all files in the directory for IMDb ID patterns
|
||||
for file_path in movie_dir.iterdir():
|
||||
if file_path.is_file():
|
||||
imdb_id = self.parse_imdb_from_path(file_path)
|
||||
if imdb_id:
|
||||
return imdb_id
|
||||
|
||||
# Finally, try NFO file content (including TMDB fallback)
|
||||
nfo_path = movie_dir / "movie.nfo"
|
||||
imdb_id = self.parse_imdb_from_nfo(nfo_path)
|
||||
if imdb_id:
|
||||
return imdb_id
|
||||
|
||||
return None
|
||||
|
||||
def find_series_imdb_id(self, series_dir: Path) -> Optional[str]:
|
||||
"""Find IMDb ID from TV series directory name, filenames, or tvshow.nfo file"""
|
||||
# First try directory name
|
||||
imdb_id = self.parse_imdb_from_path(series_dir)
|
||||
if imdb_id:
|
||||
return imdb_id
|
||||
|
||||
# Try all files in the directory for IMDb ID patterns
|
||||
for file_path in series_dir.iterdir():
|
||||
if file_path.is_file():
|
||||
imdb_id = self.parse_imdb_from_path(file_path)
|
||||
if imdb_id:
|
||||
return imdb_id
|
||||
|
||||
# Finally, try tvshow.nfo file content
|
||||
nfo_path = series_dir / "tvshow.nfo"
|
||||
imdb_id = self.parse_imdb_from_nfo(nfo_path)
|
||||
if imdb_id:
|
||||
return imdb_id
|
||||
|
||||
return None
|
||||
|
||||
def extract_nfoguard_dates_from_nfo(self, nfo_path: Path) -> Optional[Dict[str, str]]:
|
||||
"""Extract NFOGuard-managed dates from existing NFO file"""
|
||||
if not nfo_path.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
tree = ET.parse(nfo_path)
|
||||
root = tree.getroot()
|
||||
|
||||
# Look for NFOGuard fields
|
||||
dateadded_elem = root.find('.//dateadded')
|
||||
premiered_elem = root.find('.//premiered')
|
||||
aired_elem = root.find('.//aired') # For TV episodes
|
||||
lockdata_elem = root.find('.//lockdata')
|
||||
|
||||
# Consider it NFOGuard-managed ONLY if it has BOTH lockdata=true AND dateadded
|
||||
# This prevents incomplete NFO files from being marked as "complete"
|
||||
if (lockdata_elem is not None and lockdata_elem.text == "true" and
|
||||
dateadded_elem is not None and dateadded_elem.text):
|
||||
# Extract original source from NFOGuard comment, default to nfo_file_existing
|
||||
source = "nfo_file_existing"
|
||||
|
||||
# Parse XML content to find NFOGuard comment with source
|
||||
nfo_content = nfo_path.read_text(encoding='utf-8')
|
||||
import re
|
||||
source_match = re.search(r'<!--\s*NFOGuard\s*-\s*Source:\s*([^-]+?)\s*-->', nfo_content)
|
||||
if source_match:
|
||||
source = source_match.group(1).strip()
|
||||
print(f"🔍 Extracted original source from NFO comment: {source}")
|
||||
|
||||
result = {
|
||||
"source": source
|
||||
}
|
||||
|
||||
if dateadded_elem is not None and dateadded_elem.text:
|
||||
result["dateadded"] = dateadded_elem.text.strip()
|
||||
|
||||
if premiered_elem is not None and premiered_elem.text:
|
||||
result["released"] = premiered_elem.text.strip()
|
||||
|
||||
if aired_elem is not None and aired_elem.text:
|
||||
result["aired"] = aired_elem.text.strip()
|
||||
|
||||
print(f"✅ Found NFOGuard data in NFO: dateadded={result.get('dateadded', 'None')}, source={source}, released={result.get('released', 'None')}, aired={result.get('aired', 'None')}")
|
||||
return result
|
||||
|
||||
except ET.ParseError as e:
|
||||
# Handle malformed XML files gracefully (common with external NFO files)
|
||||
print(f"🔍 NFO XML parse error (will try text fallback): {nfo_path.name}")
|
||||
# Try text-based fallback for extracting NFOGuard data
|
||||
return self._extract_nfoguard_data_text_fallback(nfo_path)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Unexpected error reading NFO file {nfo_path.name}: {e}")
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
def _extract_nfoguard_data_text_fallback(self, nfo_path: Path) -> Optional[Dict[str, str]]:
|
||||
"""Text-based fallback for extracting NFOGuard data from malformed XML files"""
|
||||
try:
|
||||
content = nfo_path.read_text(encoding='utf-8', errors='ignore')
|
||||
|
||||
# Look for NFOGuard elements using regex
|
||||
import re
|
||||
|
||||
# Check for lockdata=true first (NFOGuard marker)
|
||||
if not re.search(r'<lockdata>true</lockdata>', content, re.IGNORECASE):
|
||||
return None
|
||||
|
||||
# Extract dateadded
|
||||
dateadded_match = re.search(r'<dateadded>([^<]+)</dateadded>', content, re.IGNORECASE)
|
||||
if not dateadded_match:
|
||||
return None
|
||||
|
||||
dateadded = dateadded_match.group(1).strip()
|
||||
if not dateadded:
|
||||
return None
|
||||
|
||||
result = {
|
||||
"dateadded": dateadded,
|
||||
"source": "nfo_file_existing" # Default source
|
||||
}
|
||||
|
||||
# Extract optional fields
|
||||
premiered_match = re.search(r'<premiered>([^<]+)</premiered>', content, re.IGNORECASE)
|
||||
if premiered_match:
|
||||
result["released"] = premiered_match.group(1).strip()
|
||||
|
||||
aired_match = re.search(r'<aired>([^<]+)</aired>', content, re.IGNORECASE)
|
||||
if aired_match:
|
||||
result["aired"] = aired_match.group(1).strip()
|
||||
|
||||
# Extract source from NFOGuard comment
|
||||
source_match = re.search(r'<!--\s*NFOGuard\s*-\s*Source:\s*([^-]+?)\s*-->', content)
|
||||
if source_match:
|
||||
result["source"] = source_match.group(1).strip()
|
||||
|
||||
print(f"✅ Extracted NFOGuard data via text fallback: dateadded={result.get('dateadded', 'None')}, source={result['source']}")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Text fallback extraction failed for {nfo_path.name}: {e}")
|
||||
return None
|
||||
|
||||
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_path = season_path / nfo_filename
|
||||
|
||||
if not nfo_path.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
tree = ET.parse(nfo_path)
|
||||
root = tree.getroot()
|
||||
|
||||
# Look for NFOGuard fields in episode NFO
|
||||
dateadded_elem = root.find('.//dateadded')
|
||||
aired_elem = root.find('.//aired')
|
||||
lockdata_elem = root.find('.//lockdata')
|
||||
|
||||
# Consider it NFOGuard-managed ONLY if it has BOTH lockdata=true AND dateadded
|
||||
# This prevents incomplete episode NFO files from being marked as "complete"
|
||||
if (lockdata_elem is not None and lockdata_elem.text == "true" and
|
||||
dateadded_elem is not None and dateadded_elem.text):
|
||||
# Extract original source from NFOGuard comment, default to episode_nfo_existing
|
||||
source = "episode_nfo_existing"
|
||||
|
||||
# Parse XML content to find NFOGuard comment with source
|
||||
nfo_content = nfo_path.read_text(encoding='utf-8')
|
||||
import re
|
||||
source_match = re.search(r'<!--\s*NFOGuard\s*-\s*Source:\s*([^-]+?)\s*-->', nfo_content)
|
||||
if source_match:
|
||||
source = source_match.group(1).strip()
|
||||
print(f"🔍 Extracted original source from episode NFO comment: {source}")
|
||||
|
||||
result = {
|
||||
"source": source
|
||||
}
|
||||
|
||||
if dateadded_elem is not None and dateadded_elem.text:
|
||||
result["dateadded"] = dateadded_elem.text.strip()
|
||||
|
||||
if aired_elem is not None and aired_elem.text:
|
||||
result["aired"] = aired_elem.text.strip()
|
||||
|
||||
print(f"✅ Found NFOGuard data in episode NFO S{season_num:02d}E{episode_num:02d}: dateadded={result.get('dateadded', 'None')}, source={source}, aired={result.get('aired', 'None')}")
|
||||
return result
|
||||
|
||||
except (ET.ParseError, Exception) as e:
|
||||
print(f"⚠️ Error parsing episode NFO for NFOGuard data: {e}")
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
def _parse_nfo_with_tolerance(self, nfo_path: Path):
|
||||
"""Parse NFO file with tolerance for URLs appended after XML"""
|
||||
try:
|
||||
# First try normal parsing
|
||||
tree = ET.parse(nfo_path)
|
||||
return tree.getroot()
|
||||
except ET.ParseError as e:
|
||||
# If parsing fails, try to extract just the XML part
|
||||
try:
|
||||
with open(nfo_path, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
|
||||
# Find the last </movie> tag and truncate after it
|
||||
last_movie_end = content.rfind('</movie>')
|
||||
if last_movie_end != -1:
|
||||
xml_content = content[:last_movie_end + 8] # +8 for </movie>
|
||||
|
||||
# Try parsing the truncated content
|
||||
root = ET.fromstring(xml_content)
|
||||
print(f"✅ Successfully parsed NFO after removing trailing content: {nfo_path.name}")
|
||||
return root
|
||||
else:
|
||||
# Re-raise original error if we can't find </movie>
|
||||
raise e
|
||||
except Exception:
|
||||
# Re-raise original error if our fix attempt fails
|
||||
raise e
|
||||
|
||||
def create_movie_nfo(self, movie_dir: Path, imdb_id: str, dateadded: str,
|
||||
released: Optional[str] = None, source: str = "unknown",
|
||||
lock_metadata: bool = True) -> None:
|
||||
"""Create or update movie.nfo file preserving existing content"""
|
||||
nfo_path = movie_dir / "movie.nfo"
|
||||
|
||||
# Debug output only if DEBUG=true in environment
|
||||
import os
|
||||
if os.environ.get("DEBUG", "false").lower() == "true":
|
||||
print(f"🔍 create_movie_nfo called: imdb_id={imdb_id}, dateadded={dateadded}, released={released}, source={source}")
|
||||
print(f"🔍 NFO path: {nfo_path}")
|
||||
print(f"🔍 NFO exists: {nfo_path.exists()}")
|
||||
|
||||
try:
|
||||
# Try to load existing NFO file
|
||||
if nfo_path.exists():
|
||||
try:
|
||||
# Try to parse the XML, handling URLs appended after </movie>
|
||||
movie = self._parse_nfo_with_tolerance(nfo_path)
|
||||
|
||||
# Ensure root element is <movie>
|
||||
if movie.tag != "movie":
|
||||
raise ValueError("Root element is not <movie>")
|
||||
|
||||
# Only remove elements that are clearly NFOGuard-managed
|
||||
# Look for elements that have NFOGuard characteristics
|
||||
elements_to_remove = []
|
||||
|
||||
# Remove lockdata=true (this is definitely NFOGuard)
|
||||
for lockdata in movie.findall("lockdata"):
|
||||
if lockdata.text == "true":
|
||||
elements_to_remove.append(lockdata)
|
||||
|
||||
# Remove IMDb uniqueids that we manage
|
||||
for uniqueid in movie.findall("uniqueid[@type='imdb']"):
|
||||
elements_to_remove.append(uniqueid)
|
||||
|
||||
# For dateadded/premiered/year, only remove if they appear to be NFOGuard-managed
|
||||
# (i.e., if lockdata=true exists, these are likely ours)
|
||||
has_nfoguard_lockdata = any(ld.text == "true" for ld in movie.findall("lockdata"))
|
||||
|
||||
if has_nfoguard_lockdata:
|
||||
# This NFO was managed by NFOGuard, safe to remove our fields
|
||||
for tag in ["dateadded", "premiered", "year"]:
|
||||
existing = movie.find(tag)
|
||||
if existing is not None:
|
||||
# Store the value before removing (for premiered/year)
|
||||
if tag == "premiered" and not released:
|
||||
print(f"🔍 Preserving existing premiered date: {existing.text}")
|
||||
released = existing.text
|
||||
elements_to_remove.append(existing)
|
||||
else:
|
||||
# No NFOGuard lockdata found, be more conservative
|
||||
# Only remove dateadded if it looks like NFOGuard format (ISO timestamp)
|
||||
dateadded_elem = movie.find("dateadded")
|
||||
if dateadded_elem is not None and dateadded_elem.text:
|
||||
# NFOGuard uses ISO format like "2025-10-12 16:26:02"
|
||||
if re.match(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', dateadded_elem.text.strip()):
|
||||
elements_to_remove.append(dateadded_elem)
|
||||
|
||||
# Remove all identified elements
|
||||
for elem in elements_to_remove:
|
||||
movie.remove(elem)
|
||||
|
||||
except (ET.ParseError, ValueError) as e:
|
||||
print(f"⚠️ Corrupted NFO detected: {nfo_path} - {str(e)[:100]}...")
|
||||
print(f" Creating new clean NFO file to replace corrupted one")
|
||||
movie = ET.Element("movie")
|
||||
else:
|
||||
# Create new NFO structure
|
||||
movie = ET.Element("movie")
|
||||
|
||||
# Create all NFOGuard elements first, then append them in correct order
|
||||
# This ensures they appear as a group at the bottom of the file
|
||||
nfoguard_elements = []
|
||||
|
||||
# Add IMDb uniqueid
|
||||
uniqueid = ET.Element("uniqueid", type="imdb", default="true")
|
||||
uniqueid.text = imdb_id
|
||||
nfoguard_elements.append(uniqueid)
|
||||
|
||||
# Add premiered date if we have it
|
||||
if released:
|
||||
premiered_elem = ET.Element("premiered")
|
||||
premiered_elem.text = released[:10] if len(released) >= 10 else released
|
||||
nfoguard_elements.append(premiered_elem)
|
||||
|
||||
# Extract year from premiered date for consistency
|
||||
try:
|
||||
year_value = released[:4] if len(released) >= 4 else None
|
||||
if year_value and year_value.isdigit():
|
||||
year_elem = ET.Element("year")
|
||||
year_elem.text = year_value
|
||||
nfoguard_elements.append(year_elem)
|
||||
except:
|
||||
pass # Skip year if we can't extract it
|
||||
|
||||
# Add dateadded - THIS IS CRITICAL FOR EMBY PLUGIN
|
||||
if os.environ.get("DEBUG", "false").lower() == "true":
|
||||
print(f"🔍 About to add dateadded: {dateadded} (type: {type(dateadded)})")
|
||||
if dateadded:
|
||||
dateadded_elem = ET.Element("dateadded")
|
||||
dateadded_elem.text = dateadded
|
||||
nfoguard_elements.append(dateadded_elem)
|
||||
print(f"✅ Adding dateadded to NFO: {dateadded}")
|
||||
else:
|
||||
print(f"❌ dateadded is empty/None, not adding to NFO")
|
||||
|
||||
# Add lockdata at the very end
|
||||
if lock_metadata:
|
||||
lockdata = ET.Element("lockdata")
|
||||
lockdata.text = "true"
|
||||
nfoguard_elements.append(lockdata)
|
||||
|
||||
# Add NFOGuard comment as the very last element (appears at bottom)
|
||||
nfoguard_comment = ET.Comment(f" NFOGuard - Source: {source} ")
|
||||
nfoguard_elements.append(nfoguard_comment)
|
||||
|
||||
# Now append all NFOGuard elements to the movie in one batch
|
||||
# This ensures they appear as a contiguous block at the bottom
|
||||
for elem in nfoguard_elements:
|
||||
movie.append(elem)
|
||||
|
||||
print(f"✅ Added {len(nfoguard_elements)} NFOGuard elements to bottom of NFO")
|
||||
|
||||
# Write file with proper formatting
|
||||
tree = ET.ElementTree(movie)
|
||||
ET.indent(tree, space=" ", level=0)
|
||||
|
||||
# Write directly to file (comment is already embedded in XML at bottom)
|
||||
with open(nfo_path, 'w', encoding='utf-8') as f:
|
||||
f.write('<?xml version="1.0" encoding="utf-8"?>\n')
|
||||
tree.write(f, encoding='unicode', xml_declaration=False)
|
||||
|
||||
print(f"✅ Successfully created/updated movie NFO: {nfo_path}")
|
||||
print(f" IMDb ID: {imdb_id}, Date Added: {dateadded}, Source: {source}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error creating/updating movie NFO {nfo_path}: {e}")
|
||||
|
||||
def create_tvshow_nfo(self, series_dir: Path, imdb_id: str, tvdb_id: Optional[str] = None) -> None:
|
||||
"""Create or update tvshow.nfo file preserving existing content"""
|
||||
nfo_path = series_dir / "tvshow.nfo"
|
||||
|
||||
try:
|
||||
# Try to load existing NFO file
|
||||
if nfo_path.exists():
|
||||
try:
|
||||
tree = ET.parse(nfo_path)
|
||||
tvshow = tree.getroot()
|
||||
|
||||
# Ensure root element is <tvshow>
|
||||
if tvshow.tag != "tvshow":
|
||||
raise ValueError("Root element is not <tvshow>")
|
||||
|
||||
# Remove existing NFOGuard-managed elements to avoid duplicates
|
||||
# These will be re-added at the bottom
|
||||
for tag in ["lockdata"]:
|
||||
existing = tvshow.find(tag)
|
||||
if existing is not None:
|
||||
tvshow.remove(existing)
|
||||
|
||||
# Remove ALL existing uniqueid with type="imdb" regardless of attributes
|
||||
for uniqueid in tvshow.findall("uniqueid[@type='imdb']"):
|
||||
tvshow.remove(uniqueid)
|
||||
|
||||
except (ET.ParseError, ValueError) as e:
|
||||
print(f"⚠️ Corrupted TV show NFO detected: {nfo_path} - {str(e)[:100]}...")
|
||||
print(f" Creating new clean tvshow.nfo file to replace corrupted one")
|
||||
tvshow = ET.Element("tvshow")
|
||||
else:
|
||||
# Create new NFO structure
|
||||
tvshow = ET.Element("tvshow")
|
||||
|
||||
# Add NFOGuard fields at the bottom
|
||||
|
||||
# Add IMDb uniqueid at the end
|
||||
imdb_uniqueid = ET.SubElement(tvshow, "uniqueid", type="imdb", default="true")
|
||||
imdb_uniqueid.text = imdb_id
|
||||
|
||||
# Add TVDB ID if available (preserve existing or add new)
|
||||
if tvdb_id and not tvshow.find("uniqueid[@type='tvdb']"):
|
||||
tvdb_uniqueid = ET.SubElement(tvshow, "uniqueid", type="tvdb")
|
||||
tvdb_uniqueid.text = tvdb_id
|
||||
|
||||
# Add lockdata at the very end
|
||||
lockdata = ET.SubElement(tvshow, "lockdata")
|
||||
lockdata.text = "true"
|
||||
|
||||
# Add NFOGuard comment at the bottom
|
||||
comment = ET.Comment(f" Created by {self.manager_brand} ")
|
||||
tvshow.append(comment)
|
||||
|
||||
# Write file with proper formatting
|
||||
tree = ET.ElementTree(tvshow)
|
||||
ET.indent(tree, space=" ", level=0)
|
||||
tree.write(nfo_path, encoding='utf-8', xml_declaration=True)
|
||||
|
||||
print(f"✅ Successfully created/updated TV show NFO: {nfo_path}")
|
||||
print(f" IMDb ID: {imdb_id}" + (f", TVDB ID: {tvdb_id}" if tvdb_id else ""))
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error creating/updating tvshow NFO {nfo_path}: {e}")
|
||||
|
||||
def create_season_nfo(self, season_dir: Path, season_number: int) -> None:
|
||||
"""Create or update season.nfo file preserving existing content"""
|
||||
nfo_path = season_dir / "season.nfo"
|
||||
|
||||
try:
|
||||
season_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Try to load existing NFO file
|
||||
if nfo_path.exists():
|
||||
try:
|
||||
tree = ET.parse(nfo_path)
|
||||
season = tree.getroot()
|
||||
|
||||
# Ensure root element is <season>
|
||||
if season.tag != "season":
|
||||
raise ValueError("Root element is not <season>")
|
||||
|
||||
# Remove existing NFOGuard-managed elements
|
||||
for tag in ["seasonnumber", "lockdata"]:
|
||||
existing = season.find(tag)
|
||||
if existing is not None:
|
||||
season.remove(existing)
|
||||
|
||||
except (ET.ParseError, ValueError) as e:
|
||||
print(f"⚠️ Corrupted season NFO detected: {nfo_path} - {str(e)[:100]}...")
|
||||
print(f" Creating new clean season.nfo file to replace corrupted one")
|
||||
season = ET.Element("season")
|
||||
else:
|
||||
# Create new NFO structure
|
||||
season = ET.Element("season")
|
||||
|
||||
# Add NFOGuard fields at the bottom
|
||||
seasonnumber = ET.SubElement(season, "seasonnumber")
|
||||
seasonnumber.text = str(season_number)
|
||||
|
||||
# Add lockdata at the end
|
||||
lockdata = ET.SubElement(season, "lockdata")
|
||||
lockdata.text = "true"
|
||||
|
||||
# Add NFOGuard comment at the bottom
|
||||
comment = ET.Comment(f" Created by {self.manager_brand} ")
|
||||
season.append(comment)
|
||||
|
||||
# Write file with proper formatting
|
||||
tree = ET.ElementTree(season)
|
||||
ET.indent(tree, space=" ", level=0)
|
||||
tree.write(nfo_path, encoding='utf-8', xml_declaration=True)
|
||||
|
||||
print(f"✅ Successfully created/updated season NFO: {nfo_path}")
|
||||
print(f" Season: {season_number}")
|
||||
|
||||
except Exception as e:
|
||||
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 any existing episode NFO file that matches season/episode"""
|
||||
if not season_dir.exists():
|
||||
return None
|
||||
|
||||
# Look for NFO files in the season directory
|
||||
for nfo_file in season_dir.glob("*.nfo"):
|
||||
|
||||
# Check if this NFO contains the right season/episode
|
||||
try:
|
||||
tree = ET.parse(nfo_file)
|
||||
root = tree.getroot()
|
||||
|
||||
if root.tag == "episodedetails":
|
||||
# Check for season/episode elements
|
||||
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)
|
||||
|
||||
if file_season == season_num and file_episode == episode_num:
|
||||
print(f"🔍 Found existing episode NFO: {nfo_file.name}")
|
||||
return nfo_file
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
except (ET.ParseError, Exception):
|
||||
# Skip corrupted or non-XML files
|
||||
continue
|
||||
|
||||
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"""
|
||||
# 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:
|
||||
# Check for existing NFO file at target location
|
||||
source_nfo_path = nfo_path if nfo_path.exists() else None
|
||||
|
||||
if source_nfo_path:
|
||||
try:
|
||||
tree = ET.parse(source_nfo_path)
|
||||
episode = tree.getroot()
|
||||
|
||||
# Ensure root element is <episodedetails>
|
||||
if episode.tag != "episodedetails":
|
||||
raise ValueError("Root element is not <episodedetails>")
|
||||
|
||||
print(f"📝 Updating existing NFO: {nfo_path.name}")
|
||||
|
||||
# Preserve existing content fields and store NFOGuard-managed fields for re-adding
|
||||
preserved_values = {}
|
||||
|
||||
# Extract and preserve NFOGuard-managed fields (we'll re-add these at the bottom)
|
||||
nfoguard_fields = ["aired", "dateadded", "lockdata", "season", "episode"]
|
||||
for tag in nfoguard_fields:
|
||||
existing = episode.find(tag)
|
||||
if existing is not None:
|
||||
# Store the value before removing
|
||||
if tag == "aired" and not aired:
|
||||
aired = existing.text # Preserve existing aired date
|
||||
preserved_values[tag] = existing.text
|
||||
episode.remove(existing)
|
||||
|
||||
# Important: DO NOT remove content fields like title, plot, runtime, premiered, etc.
|
||||
# These should be preserved from the long-named NFO files
|
||||
|
||||
# Debug: Show what fields are preserved after removing NFOGuard fields (if DEBUG enabled)
|
||||
if self.debug:
|
||||
preserved_fields = [elem.tag for elem in episode]
|
||||
if preserved_fields:
|
||||
print(f" 🔍 Content preserved after cleanup: {', '.join(preserved_fields)}")
|
||||
else:
|
||||
print(f" ℹ️ NFO contains only NFOGuard metadata (no additional content fields)")
|
||||
|
||||
except (ET.ParseError, ValueError) as e:
|
||||
print(f"⚠️ Corrupted episode NFO detected: {nfo_path} - {str(e)[:100]}...")
|
||||
print(f" Creating new clean episode NFO file to replace corrupted one")
|
||||
episode = ET.Element("episodedetails")
|
||||
else:
|
||||
# Create new NFO structure
|
||||
episode = ET.Element("episodedetails")
|
||||
|
||||
# Add enhanced metadata only if not already present (preserve existing from long-named NFO)
|
||||
if enhanced_metadata:
|
||||
if enhanced_metadata.get("title") and not episode.find("title"):
|
||||
title_elem = ET.SubElement(episode, "title")
|
||||
title_elem.text = enhanced_metadata["title"]
|
||||
|
||||
if enhanced_metadata.get("overview") and not episode.find("plot"):
|
||||
plot_elem = ET.SubElement(episode, "plot")
|
||||
plot_elem.text = enhanced_metadata["overview"]
|
||||
|
||||
if enhanced_metadata.get("runtime") and not episode.find("runtime"):
|
||||
runtime_elem = ET.SubElement(episode, "runtime")
|
||||
runtime_elem.text = str(enhanced_metadata["runtime"])
|
||||
|
||||
# Add NFOGuard fields at the bottom
|
||||
|
||||
# Basic episode info at the end
|
||||
season_elem = ET.SubElement(episode, "season")
|
||||
season_elem.text = str(season_num)
|
||||
|
||||
episode_elem = ET.SubElement(episode, "episode")
|
||||
episode_elem.text = str(episode_num)
|
||||
|
||||
# Dates at the end
|
||||
if aired:
|
||||
aired_elem = ET.SubElement(episode, "aired")
|
||||
# Convert datetime objects to strings
|
||||
aired_str = str(aired)
|
||||
aired_elem.text = aired_str[:10] if len(aired_str) >= 10 else aired_str
|
||||
|
||||
# Debug logging for dateadded
|
||||
print(f"🔍 DEBUG: dateadded value: {repr(dateadded)} (type: {type(dateadded)})")
|
||||
if dateadded:
|
||||
dateadded_elem = ET.SubElement(episode, "dateadded")
|
||||
# Convert datetime objects to strings
|
||||
dateadded_elem.text = str(dateadded)
|
||||
print(f"✅ Added dateadded to episode NFO: {dateadded}")
|
||||
else:
|
||||
print(f"❌ dateadded is empty/None, not adding to episode NFO")
|
||||
|
||||
# Add lockdata at the very end
|
||||
if lock_metadata:
|
||||
lockdata = ET.SubElement(episode, "lockdata")
|
||||
lockdata.text = "true"
|
||||
|
||||
# Add NFOGuard comment at the bottom
|
||||
comment = ET.Comment(f" Created by {self.manager_brand} - Source: {source} ")
|
||||
episode.append(comment)
|
||||
|
||||
# Write file with proper formatting
|
||||
tree = ET.ElementTree(episode)
|
||||
ET.indent(tree, space=" ", level=0)
|
||||
tree.write(nfo_path, encoding='utf-8', xml_declaration=True)
|
||||
|
||||
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}")
|
||||
|
||||
def set_file_mtime(self, file_path: Path, iso_timestamp) -> None:
|
||||
"""Set file modification time to match import date"""
|
||||
try:
|
||||
# Convert datetime objects to strings first
|
||||
if hasattr(iso_timestamp, 'isoformat'):
|
||||
iso_timestamp = iso_timestamp.isoformat()
|
||||
elif not isinstance(iso_timestamp, str):
|
||||
iso_timestamp = str(iso_timestamp)
|
||||
|
||||
# Parse ISO timestamp
|
||||
if iso_timestamp.endswith('Z'):
|
||||
dt = datetime.fromisoformat(iso_timestamp.replace('Z', '+00:00'))
|
||||
elif '+' in iso_timestamp or 'T' in iso_timestamp:
|
||||
dt = datetime.fromisoformat(iso_timestamp)
|
||||
elif ' ' in iso_timestamp:
|
||||
# Handle space-separated datetime format (e.g., "2025-10-16 20:31:22")
|
||||
dt = datetime.fromisoformat(iso_timestamp.replace(' ', 'T'))
|
||||
else:
|
||||
# Assume it's a simple date (e.g., "2025-10-16")
|
||||
dt = datetime.fromisoformat(iso_timestamp + 'T00:00:00')
|
||||
|
||||
# Convert to timestamp
|
||||
timestamp = dt.timestamp()
|
||||
|
||||
# Set both access and modification times
|
||||
os.utime(file_path, (timestamp, timestamp))
|
||||
print(f"✅ Updated file timestamp: {file_path.name} -> {iso_timestamp}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error setting mtime for {file_path}: {e}")
|
||||
|
||||
def update_movie_files_mtime(self, movie_dir: Path, iso_timestamp: str) -> None:
|
||||
"""Update modification times for all video files in movie directory"""
|
||||
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
|
||||
updated_files = []
|
||||
|
||||
for file_path in movie_dir.iterdir():
|
||||
if file_path.is_file() and file_path.suffix.lower() in video_exts:
|
||||
self.set_file_mtime(file_path, iso_timestamp)
|
||||
updated_files.append(file_path.name)
|
||||
|
||||
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}")
|
||||
Reference in New Issue
Block a user