sbcrumb
e780b9e125
- Show all search paths being checked and their existence status - Debug title fallback logic showing cleaned title comparisons - Track exactly where false matches occur (e.g., 'girls' in 'thegoldengirls') - Identify why only some search paths are processed
286 lines
8.6 KiB
Python
286 lines
8.6 KiB
Python
"""
|
|
File utility functions for NFOGuard
|
|
Common file operations to eliminate code duplication
|
|
"""
|
|
import glob
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Optional, List, Dict, Tuple, Union
|
|
|
|
from utils.logging import _log
|
|
|
|
|
|
# Video file extensions used throughout the application
|
|
VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.ts'}
|
|
|
|
# Episode pattern for TV series files
|
|
EPISODE_PATTERN = re.compile(
|
|
r'.*[sS](\d{1,2})[eE](\d{1,3}).*|.*(\d{1,2})x(\d{1,3}).*'
|
|
)
|
|
|
|
|
|
def find_media_path_by_imdb_and_title(
|
|
title: str,
|
|
imdb_id: str,
|
|
search_paths: List[Path],
|
|
webhook_path: Optional[str] = None,
|
|
path_mapper = None
|
|
) -> Optional[Path]:
|
|
"""
|
|
Unified media path finder for both TV series and movies
|
|
|
|
Args:
|
|
title: Media title to search for
|
|
imdb_id: IMDb ID to search for
|
|
search_paths: List of paths to search in (tv_paths or movie_paths)
|
|
webhook_path: Optional webhook path to try first
|
|
path_mapper: Optional path mapper for webhook path conversion
|
|
|
|
Returns:
|
|
Path to media directory if found, None otherwise
|
|
"""
|
|
# Try webhook path first if provided
|
|
if webhook_path and path_mapper:
|
|
try:
|
|
if hasattr(path_mapper, 'sonarr_path_to_container_path'):
|
|
container_path = path_mapper.sonarr_path_to_container_path(webhook_path)
|
|
elif hasattr(path_mapper, 'radarr_path_to_container_path'):
|
|
container_path = path_mapper.radarr_path_to_container_path(webhook_path)
|
|
else:
|
|
container_path = webhook_path
|
|
|
|
path_obj = Path(container_path)
|
|
if path_obj.exists():
|
|
return path_obj
|
|
except Exception as e:
|
|
_log("WARNING", f"Failed to process webhook path {webhook_path}: {e}")
|
|
|
|
# Search by IMDb ID or title in configured paths
|
|
print(f"🔍 Total search paths to check: {len(search_paths)} - {[str(p) for p in search_paths]}")
|
|
for media_path in search_paths:
|
|
print(f"🔍 Checking path: {media_path} (exists: {media_path.exists()})")
|
|
if not media_path.exists():
|
|
print(f"❌ Skipping non-existent path: {media_path}")
|
|
continue
|
|
|
|
# Search by IMDb ID first (more reliable)
|
|
if imdb_id:
|
|
# Use proper glob pattern - escape brackets to match literal [imdb-ID]
|
|
pattern = str(media_path / f"*\\[imdb-{imdb_id}\\]*")
|
|
print(f"🔍 Searching with pattern: {pattern}")
|
|
matches = glob.glob(pattern)
|
|
print(f"🔍 Pattern matches found: {len(matches)} - {matches[:3] if matches else 'None'}")
|
|
if matches:
|
|
print(f"✅ Returning first match: {matches[0]}")
|
|
return Path(matches[0])
|
|
|
|
# Search by title as fallback
|
|
if title:
|
|
title_clean = clean_title_for_search(title)
|
|
print(f"🔍 Title fallback: searching for '{title}' → cleaned: '{title_clean}'")
|
|
for item in media_path.iterdir():
|
|
if item.is_dir() and "[imdb-" in item.name.lower():
|
|
item_clean = clean_title_for_search(item.name)
|
|
print(f"🔍 Checking directory: '{item.name}' → cleaned: '{item_clean}'")
|
|
if title_clean in item_clean:
|
|
print(f"✅ Title match found: '{title_clean}' in '{item_clean}' → returning {item}")
|
|
return item
|
|
|
|
return None
|
|
|
|
|
|
def clean_title_for_search(title: str) -> str:
|
|
"""
|
|
Clean title for fuzzy matching
|
|
|
|
Args:
|
|
title: Raw title string
|
|
|
|
Returns:
|
|
Cleaned title for comparison
|
|
"""
|
|
return title.lower().replace(" ", "").replace("-", "").replace(".", "")
|
|
|
|
|
|
def find_video_files(directory: Path, recursive: bool = True) -> List[Path]:
|
|
"""
|
|
Find all video files in a directory
|
|
|
|
Args:
|
|
directory: Directory to search
|
|
recursive: Whether to search recursively
|
|
|
|
Returns:
|
|
List of video file paths
|
|
"""
|
|
if not directory.exists():
|
|
return []
|
|
|
|
video_files = []
|
|
|
|
if recursive:
|
|
for item in directory.rglob('*'):
|
|
if item.is_file() and item.suffix.lower() in VIDEO_EXTENSIONS:
|
|
video_files.append(item)
|
|
else:
|
|
for item in directory.iterdir():
|
|
if item.is_file() and item.suffix.lower() in VIDEO_EXTENSIONS:
|
|
video_files.append(item)
|
|
|
|
return video_files
|
|
|
|
|
|
def extract_episode_info(filename: str) -> Optional[Tuple[int, int]]:
|
|
"""
|
|
Extract season and episode numbers from filename
|
|
|
|
Args:
|
|
filename: Video filename to parse
|
|
|
|
Returns:
|
|
Tuple of (season, episode) if found, None otherwise
|
|
"""
|
|
match = EPISODE_PATTERN.match(filename)
|
|
if not match:
|
|
return None
|
|
|
|
if match.group(1) and match.group(2): # SxxExx format
|
|
season = int(match.group(1))
|
|
episode = int(match.group(2))
|
|
elif match.group(3) and match.group(4): # NxNN format
|
|
season = int(match.group(3))
|
|
episode = int(match.group(4))
|
|
else:
|
|
return None
|
|
|
|
return (season, episode)
|
|
|
|
|
|
def find_episodes_on_disk(series_path: Path) -> Dict[Tuple[int, int], List[Path]]:
|
|
"""
|
|
Find all episodes on disk and return mapping of (season, episode) -> [video_files]
|
|
|
|
Args:
|
|
series_path: Path to series directory
|
|
|
|
Returns:
|
|
Dictionary mapping (season, episode) tuples to lists of video files
|
|
"""
|
|
episodes = {}
|
|
|
|
if not series_path.exists():
|
|
return episodes
|
|
|
|
for video_file in find_video_files(series_path, recursive=True):
|
|
episode_info = extract_episode_info(video_file.name)
|
|
if episode_info:
|
|
season, episode = episode_info
|
|
key = (season, episode)
|
|
if key not in episodes:
|
|
episodes[key] = []
|
|
episodes[key].append(video_file)
|
|
|
|
return episodes
|
|
|
|
|
|
def extract_title_from_directory_name(directory_name: str) -> Optional[str]:
|
|
"""
|
|
Extract clean title from directory name, removing year and IMDb ID
|
|
|
|
Args:
|
|
directory_name: Directory name to parse
|
|
|
|
Returns:
|
|
Cleaned title or None if no title found
|
|
"""
|
|
name = directory_name
|
|
|
|
# Remove IMDb ID part: [imdb-ttXXXXXX] or [ttXXXXXX]
|
|
name = re.sub(r'\s*\[imdb-?tt\d+\]', '', name, flags=re.IGNORECASE)
|
|
name = re.sub(r'\s*\[tt\d+\]', '', name, flags=re.IGNORECASE)
|
|
|
|
# Remove year in parentheses: (YYYY)
|
|
name = re.sub(r'\s*\(\d{4}\)', '', name)
|
|
|
|
# Clean up extra spaces
|
|
name = ' '.join(name.split())
|
|
|
|
return name.strip() if name.strip() else None
|
|
|
|
|
|
def extract_imdb_id_from_path(path: Union[str, Path]) -> Optional[str]:
|
|
"""
|
|
Extract IMDb ID from directory or file path
|
|
|
|
Args:
|
|
path: Path to examine (string or Path object)
|
|
|
|
Returns:
|
|
IMDb ID if found, None otherwise
|
|
"""
|
|
path_str = str(path)
|
|
|
|
# Look for [imdb-ttXXXXXX] or [ttXXXXXX] patterns
|
|
patterns = [
|
|
r'\[imdb-(tt\d+)\]', # [imdb-tt1234567]
|
|
r'\[(tt\d+)\]', # [tt1234567]
|
|
r'imdb[_-]?(tt\d+)', # imdb_tt1234567 or imdb-tt1234567
|
|
r'(tt\d{7,})', # standalone tt1234567 (7+ digits)
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, path_str, re.IGNORECASE)
|
|
if match:
|
|
imdb_id = match.group(1)
|
|
# Ensure it starts with 'tt'
|
|
if not imdb_id.startswith('tt'):
|
|
imdb_id = f'tt{imdb_id}'
|
|
return imdb_id
|
|
|
|
return None
|
|
|
|
|
|
def is_video_file(file_path: Path) -> bool:
|
|
"""
|
|
Check if a file is a video file based on extension
|
|
|
|
Args:
|
|
file_path: Path to check
|
|
|
|
Returns:
|
|
True if it's a video file, False otherwise
|
|
"""
|
|
return file_path.suffix.lower() in VIDEO_EXTENSIONS
|
|
|
|
|
|
def safe_directory_scan(directory: Path, pattern: str = "*") -> List[Path]:
|
|
"""
|
|
Safely scan directory with error handling
|
|
|
|
Args:
|
|
directory: Directory to scan
|
|
pattern: Glob pattern to match
|
|
|
|
Returns:
|
|
List of matching paths, empty list if scan fails
|
|
"""
|
|
try:
|
|
if not directory.exists():
|
|
return []
|
|
return list(directory.glob(pattern))
|
|
except (PermissionError, OSError) as e:
|
|
_log("WARNING", f"Failed to scan directory {directory}: {e}")
|
|
return []
|
|
|
|
|
|
def normalize_path_separators(path: str) -> str:
|
|
"""
|
|
Normalize path separators for cross-platform compatibility
|
|
|
|
Args:
|
|
path: Path string to normalize
|
|
|
|
Returns:
|
|
Normalized path string
|
|
"""
|
|
return str(Path(path).as_posix()) |