diff --git a/processors/movie_processor.py b/processors/movie_processor.py index 1cb0dc0..326a354 100644 --- a/processors/movie_processor.py +++ b/processors/movie_processor.py @@ -3,7 +3,6 @@ Movie Processor for NFOGuard Handles movie processing and metadata management """ import os -import glob import re import xml.etree.ElementTree as ET from pathlib import Path @@ -18,6 +17,7 @@ from clients.radarr_client import RadarrClient from clients.external_clients import ExternalClientManager from config.settings import config from utils.logging import _log +from utils.file_utils import find_media_path_by_imdb_and_title def _get_local_timezone(): @@ -79,36 +79,14 @@ class MovieProcessor: self.external_clients = ExternalClientManager() def find_movie_path(self, movie_title: str, imdb_id: str, radarr_path: str = None) -> Optional[Path]: - """Find movie directory path""" - # Try webhook path first - if radarr_path: - container_path = self.path_mapper.radarr_path_to_container_path(radarr_path) - path_obj = Path(container_path) - if path_obj.exists(): - return path_obj - - # Search by IMDb ID or title - for media_path in config.movie_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 movie_title: - title_clean = movie_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 + """Find movie directory path using unified file utilities""" + return find_media_path_by_imdb_and_title( + title=movie_title, + imdb_id=imdb_id, + search_paths=config.movie_paths, + webhook_path=radarr_path, + path_mapper=self.path_mapper + ) def process_movie(self, movie_path: Path, webhook_mode: bool = False) -> None: """Process a movie directory""" diff --git a/processors/tv_processor.py b/processors/tv_processor.py index 7e14e40..c30b1b2 100644 --- a/processors/tv_processor.py +++ b/processors/tv_processor.py @@ -3,7 +3,6 @@ TV Series Processor for NFOGuard Handles TV series processing and episode management """ import os -import glob import re from pathlib import Path from typing import Optional, Dict, List, Set, Tuple, Any @@ -16,6 +15,11 @@ from clients.sonarr_client import SonarrClient from clients.external_clients import ExternalClientManager from config.settings import config from utils.logging import _log +from utils.file_utils import ( + find_media_path_by_imdb_and_title, + find_episodes_on_disk, + extract_title_from_directory_name +) class TVProcessor: @@ -32,36 +36,14 @@ class TVProcessor: 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 + """Find series directory path using unified file utilities""" + return find_media_path_by_imdb_and_title( + title=series_title, + imdb_id=imdb_id, + search_paths=config.tv_paths, + webhook_path=sonarr_path, + path_mapper=self.path_mapper + ) def process_series(self, series_path: Path) -> None: """Process a TV series directory""" @@ -76,7 +58,7 @@ class TVProcessor: self.db.upsert_series(imdb_id, str(series_path)) # Find video files - disk_episodes = self._find_disk_episodes(series_path) + disk_episodes = find_episodes_on_disk(series_path) _log("INFO", f"Found {len(disk_episodes)} episodes on disk") # Get episode dates @@ -108,55 +90,9 @@ class TVProcessor: _log("INFO", f"Completed processing TV series: {series_path.name}") def _extract_series_title_from_path(self, series_path: Path) -> Optional[str]: - """Extract series title from directory path, removing year and IMDb ID""" - name = series_path.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 + """Extract series title from directory path using unified file utilities""" + return extract_title_from_directory_name(series_path.name) - def _find_disk_episodes(self, series_path: Path) -> Dict[Tuple[int, int], List[Path]]: - """Find all episodes on disk and return mapping of (season, episode) -> [video_files]""" - episodes = {} - - if not series_path.exists(): - return episodes - - video_extensions = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.ts'} - - # Define regex pattern for episode files - episode_pattern = re.compile( - r'.*[sS](\d{1,2})[eE](\d{1,3}).*|.*(\d{1,2})x(\d{1,3}).*' - ) - - for item in series_path.rglob('*'): - if item.is_file() and item.suffix.lower() in video_extensions: - # Try to extract season/episode from filename - match = episode_pattern.match(item.name) - if match: - 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: - continue - - key = (season, episode) - if key not in episodes: - episodes[key] = [] - episodes[key].append(item) - - return episodes def _gather_episode_dates(self, series_path: Path, imdb_id: str, disk_episodes: Dict[Tuple[int, int], List[Path]]) -> Dict[Tuple[int, int], Tuple[Optional[str], Optional[str], str]]: """Gather episode air dates and date added information""" @@ -251,7 +187,7 @@ class TVProcessor: _log("INFO", f"Processing season {season_num} of series: {series_path_obj.name}") # Find episodes in this season - disk_episodes = self._find_disk_episodes(series_path_obj) + disk_episodes = find_episodes_on_disk(series_path_obj) season_episodes = {k: v for k, v in disk_episodes.items() if k[0] == season_num} if not season_episodes: diff --git a/utils/error_handler.py b/utils/error_handler.py new file mode 100644 index 0000000..3148558 --- /dev/null +++ b/utils/error_handler.py @@ -0,0 +1,284 @@ +""" +Error handling utilities for NFOGuard +Provides structured error handling, retry mechanisms, and error reporting +""" +import time +import functools +from typing import Callable, Optional, Type, Union, List, Any +from pathlib import Path + +from utils.logging import _log +from utils.exceptions import ( + NFOGuardException, + RetryableError, + NetworkRetryableError, + TemporaryFileError, + ExternalAPIError, + FileOperationError +) + + +def with_error_handling( + operation_name: str, + log_errors: bool = True, + reraise: bool = True, + fallback_value: Any = None +): + """ + Decorator for standardized error handling + + Args: + operation_name: Name of the operation for logging + log_errors: Whether to log errors automatically + reraise: Whether to reraise exceptions after logging + fallback_value: Value to return if error occurs and reraise=False + """ + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except NFOGuardException as e: + if log_errors: + _log("ERROR", f"{operation_name} failed: {e.message}") + if e.details: + _log("DEBUG", f"{operation_name} error details: {e.details}") + if reraise: + raise + return fallback_value + except Exception as e: + if log_errors: + _log("ERROR", f"{operation_name} failed with unexpected error: {e}") + if reraise: + # Wrap unexpected errors in our custom exception + raise NFOGuardException( + f"{operation_name} failed: {str(e)}", + {"original_error": str(e), "error_type": type(e).__name__} + ) + return fallback_value + return wrapper + return decorator + + +def with_retry( + max_attempts: int = 3, + delay: float = 1.0, + backoff_factor: float = 2.0, + retry_on: Union[Type[Exception], List[Type[Exception]]] = None +): + """ + Decorator for retry logic on retryable errors + + Args: + max_attempts: Maximum number of retry attempts + delay: Initial delay between retries in seconds + backoff_factor: Factor to multiply delay by after each attempt + retry_on: Exception types to retry on (defaults to RetryableError) + """ + if retry_on is None: + retry_on = [RetryableError] + elif not isinstance(retry_on, list): + retry_on = [retry_on] + + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + current_delay = delay + last_exception = None + + for attempt in range(max_attempts): + try: + return func(*args, **kwargs) + except Exception as e: + last_exception = e + + # Check if this is a retryable error + should_retry = any(isinstance(e, exc_type) for exc_type in retry_on) + + if not should_retry or attempt == max_attempts - 1: + # Don't retry or max attempts reached + raise + + # Use custom retry delay if available + retry_delay = current_delay + if isinstance(e, RetryableError) and e.retry_after: + retry_delay = e.retry_after + + _log("WARNING", f"Attempt {attempt + 1}/{max_attempts} failed: {e}. Retrying in {retry_delay}s...") + time.sleep(retry_delay) + current_delay *= backoff_factor + + # This should never be reached, but just in case + raise last_exception + return wrapper + return decorator + + +def safe_file_operation( + operation: str, + file_path: Union[str, Path], + operation_func: Callable, + *args, + **kwargs +) -> Any: + """ + Safely perform file operations with error handling + + Args: + operation: Description of the operation + file_path: Path to the file being operated on + operation_func: Function to execute + *args, **kwargs: Arguments to pass to operation_func + + Returns: + Result of operation_func or None if error + + Raises: + FileOperationError: If file operation fails + """ + try: + return operation_func(*args, **kwargs) + except PermissionError as e: + raise FileOperationError(operation, str(file_path), f"Permission denied: {e}") + except FileNotFoundError as e: + raise FileOperationError(operation, str(file_path), f"File not found: {e}") + except OSError as e: + # Check if this might be a temporary error + if e.errno in [28, 122]: # No space left, quota exceeded + raise TemporaryFileError(str(file_path), operation, f"Disk space issue: {e}") + raise FileOperationError(operation, str(file_path), f"OS error: {e}") + except Exception as e: + raise FileOperationError(operation, str(file_path), f"Unexpected error: {e}") + + +def safe_api_call( + api_name: str, + operation: str, + api_func: Callable, + *args, + **kwargs +) -> Any: + """ + Safely perform API calls with error handling + + Args: + api_name: Name of the API (e.g., "Sonarr", "TMDB") + operation: Description of the operation + api_func: Function to execute + *args, **kwargs: Arguments to pass to api_func + + Returns: + Result of api_func + + Raises: + ExternalAPIError: If API call fails + NetworkRetryableError: If network error that can be retried + """ + try: + return api_func(*args, **kwargs) + except ConnectionError as e: + raise NetworkRetryableError(f"{api_name} API", f"Connection error: {e}") + except TimeoutError as e: + raise NetworkRetryableError(f"{api_name} API", f"Timeout error: {e}") + except Exception as e: + # Check if it's an HTTP error with status code + status_code = getattr(e, 'status_code', None) or getattr(e, 'response', {}).get('status_code') + response_text = getattr(e, 'text', None) or str(e) + + # Retry on certain HTTP status codes + if status_code in [429, 502, 503, 504]: # Rate limit, bad gateway, service unavailable, gateway timeout + raise NetworkRetryableError(f"{api_name} API", f"HTTP {status_code}: {response_text}") + + raise ExternalAPIError(api_name, operation, status_code, response_text) + + +def log_structured_error(error: NFOGuardException, context: Optional[str] = None) -> None: + """ + Log structured error information + + Args: + error: NFOGuardException to log + context: Additional context for the error + """ + error_dict = error.to_dict() + if context: + error_dict['context'] = context + + _log("ERROR", f"Structured error: {error.message}") + _log("DEBUG", f"Error details: {error_dict}") + + +def create_error_response(error: NFOGuardException, include_details: bool = False) -> dict: + """ + Create standardized error response for API endpoints + + Args: + error: NFOGuardException to convert + include_details: Whether to include detailed error information + + Returns: + Dictionary suitable for JSON response + """ + response = { + "status": "error", + "error_type": error.__class__.__name__, + "message": error.message + } + + if include_details and error.details: + response["details"] = error.details + + return response + + +def validate_required_config(config_dict: dict, required_keys: List[str]) -> None: + """ + Validate that required configuration keys are present and not empty + + Args: + config_dict: Configuration dictionary to validate + required_keys: List of required configuration keys + + Raises: + ConfigurationError: If required configuration is missing or invalid + """ + from utils.exceptions import ConfigurationError + + missing_keys = [] + empty_keys = [] + + for key in required_keys: + if key not in config_dict: + missing_keys.append(key) + elif not config_dict[key]: + empty_keys.append(key) + + if missing_keys: + raise ConfigurationError( + "missing_required_config", + f"Missing required configuration keys: {missing_keys}", + {"missing_keys": missing_keys} + ) + + if empty_keys: + raise ConfigurationError( + "empty_required_config", + f"Required configuration keys are empty: {empty_keys}", + {"empty_keys": empty_keys} + ) + + +class ErrorContext: + """Context manager for adding context to errors""" + + def __init__(self, context: str): + self.context = context + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if exc_type and issubclass(exc_type, NFOGuardException): + exc_val.details = exc_val.details or {} + exc_val.details['error_context'] = self.context + return False # Don't suppress the exception \ No newline at end of file diff --git a/utils/exceptions.py b/utils/exceptions.py new file mode 100644 index 0000000..f2ca212 --- /dev/null +++ b/utils/exceptions.py @@ -0,0 +1,172 @@ +""" +Custom exceptions for NFOGuard +Provides structured error handling and better error reporting +""" +from typing import Optional, Dict, Any + + +class NFOGuardException(Exception): + """Base exception for all NFOGuard errors""" + + def __init__(self, message: str, details: Optional[Dict[str, Any]] = None): + super().__init__(message) + self.message = message + self.details = details or {} + + def to_dict(self) -> Dict[str, Any]: + """Convert exception to dictionary for structured logging""" + return { + "error_type": self.__class__.__name__, + "message": self.message, + "details": self.details + } + + +class MediaPathNotFoundError(NFOGuardException): + """Raised when media directory cannot be found""" + + def __init__(self, media_type: str, title: str, imdb_id: Optional[str] = None, search_paths: Optional[list] = None): + details = { + "media_type": media_type, + "title": title, + "imdb_id": imdb_id, + "search_paths": [str(p) for p in (search_paths or [])] + } + message = f"{media_type.title()} directory not found: {title}" + if imdb_id: + message += f" (IMDb: {imdb_id})" + super().__init__(message, details) + + +class IMDbIDNotFoundError(NFOGuardException): + """Raised when IMDb ID cannot be extracted from path or files""" + + def __init__(self, path: str, media_type: str = "media"): + details = { + "path": path, + "media_type": media_type + } + message = f"No IMDb ID found for {media_type}: {path}" + super().__init__(message, details) + + +class WebhookProcessingError(NFOGuardException): + """Raised when webhook processing fails""" + + def __init__(self, webhook_type: str, reason: str, payload: Optional[Dict] = None): + details = { + "webhook_type": webhook_type, + "reason": reason, + "payload": payload + } + message = f"{webhook_type} webhook processing failed: {reason}" + super().__init__(message, details) + + +class ExternalAPIError(NFOGuardException): + """Raised when external API calls fail""" + + def __init__(self, api_name: str, operation: str, status_code: Optional[int] = None, response: Optional[str] = None): + details = { + "api_name": api_name, + "operation": operation, + "status_code": status_code, + "response": response + } + message = f"{api_name} API error during {operation}" + if status_code: + message += f" (HTTP {status_code})" + super().__init__(message, details) + + +class DatabaseError(NFOGuardException): + """Raised when database operations fail""" + + def __init__(self, operation: str, table: Optional[str] = None, original_error: Optional[Exception] = None): + details = { + "operation": operation, + "table": table, + "original_error": str(original_error) if original_error else None + } + message = f"Database error during {operation}" + if table: + message += f" on table {table}" + super().__init__(message, details) + + +class NFOCreationError(NFOGuardException): + """Raised when NFO file creation fails""" + + def __init__(self, nfo_path: str, reason: str, media_type: str = "media"): + details = { + "nfo_path": nfo_path, + "reason": reason, + "media_type": media_type + } + message = f"Failed to create {media_type} NFO file: {reason}" + super().__init__(message, details) + + +class ConfigurationError(NFOGuardException): + """Raised when configuration is invalid or missing""" + + def __init__(self, setting: str, reason: str, current_value: Optional[Any] = None): + details = { + "setting": setting, + "reason": reason, + "current_value": current_value + } + message = f"Configuration error for {setting}: {reason}" + super().__init__(message, details) + + +class FileOperationError(NFOGuardException): + """Raised when file operations fail""" + + def __init__(self, operation: str, file_path: str, reason: str): + details = { + "operation": operation, + "file_path": file_path, + "reason": reason + } + message = f"File {operation} failed for {file_path}: {reason}" + super().__init__(message, details) + + +class DateProcessingError(NFOGuardException): + """Raised when date processing or parsing fails""" + + def __init__(self, date_value: str, operation: str, media_type: str = "media"): + details = { + "date_value": date_value, + "operation": operation, + "media_type": media_type + } + message = f"Date processing error during {operation} for {media_type}: {date_value}" + super().__init__(message, details) + + +class RetryableError(NFOGuardException): + """Base class for errors that can be retried""" + + def __init__(self, message: str, details: Optional[Dict[str, Any]] = None, retry_after: Optional[int] = None): + super().__init__(message, details) + self.retry_after = retry_after # Seconds to wait before retry + + +class NetworkRetryableError(RetryableError): + """Network errors that can be retried""" + + def __init__(self, url: str, reason: str, retry_after: Optional[int] = 30): + details = {"url": url, "reason": reason} + message = f"Network error for {url}: {reason}" + super().__init__(message, details, retry_after) + + +class TemporaryFileError(RetryableError): + """Temporary file system errors that can be retried""" + + def __init__(self, file_path: str, operation: str, reason: str, retry_after: Optional[int] = 5): + details = {"file_path": file_path, "operation": operation, "reason": reason} + message = f"Temporary file error during {operation} for {file_path}: {reason}" + super().__init__(message, details, retry_after) \ No newline at end of file diff --git a/utils/file_utils.py b/utils/file_utils.py new file mode 100644 index 0000000..205be99 --- /dev/null +++ b/utils/file_utils.py @@ -0,0 +1,276 @@ +""" +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 + for media_path in search_paths: + if not media_path.exists(): + continue + + # Search by IMDb ID first (more reliable) + if imdb_id: + pattern = str(media_path / f"*[imdb-{imdb_id}]*") + matches = glob.glob(pattern) + if matches: + return Path(matches[0]) + + # Search by title as fallback + if title: + title_clean = clean_title_for_search(title) + for item in media_path.iterdir(): + if item.is_dir() and "[imdb-" in item.name.lower(): + item_clean = clean_title_for_search(item.name) + if title_clean in item_clean: + 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()) \ No newline at end of file diff --git a/utils/nfo_patterns.py b/utils/nfo_patterns.py new file mode 100644 index 0000000..0bc9d20 --- /dev/null +++ b/utils/nfo_patterns.py @@ -0,0 +1,375 @@ +""" +NFO parsing patterns and utilities for NFOGuard +Consolidates common NFO parsing logic and patterns +""" +import re +import xml.etree.ElementTree as ET +from pathlib import Path +from typing import Optional, Dict, Any, List +from datetime import datetime + +from utils.logging import _log +from utils.exceptions import NFOCreationError, FileOperationError +from utils.validation import validate_imdb_id, validate_date_string + + +# Common IMDb ID patterns used across the application +IMDB_PATTERNS = [ + r'\[imdb-?(tt\d+)\]', # [imdb-tt1234567] or [imdb-1234567] + r'\[(tt\d+)\]', # [tt1234567] + r'\{imdb-?(tt\d+)\}', # {imdb-tt1234567} or {imdb-1234567} + r'\(imdb-?(tt\d+)\)', # (imdb-tt1234567) or (imdb-1234567) + r'[-_\s](tt\d+)$', # tt1234567 at end of string + r'imdb[_-]?(tt\d+)', # imdb_tt1234567 or imdb-tt1234567 +] + +# Episode filename patterns +EPISODE_PATTERNS = [ + r'.*[sS](\d{1,2})[eE](\d{1,3}).*', # S01E01 + r'.*(\d{1,2})x(\d{1,3}).*', # 1x01 + r'.*[sS](\d{1,2})\.?[eE](\d{1,3}).*', # S01.E01 + r'.*Season[_\s]?(\d{1,2})[_\s]?Episode[_\s]?(\d{1,3}).*', # Season 1 Episode 1 +] + +# Common NFO XML namespaces +NFO_NAMESPACES = { + 'kodi': 'http://kodi.tv/moviedb', + 'tmdb': 'https://www.themoviedb.org', + 'imdb': 'https://www.imdb.com' +} + + +def extract_imdb_id_from_text(text: str) -> Optional[str]: + """ + Extract IMDb ID from text using consolidated patterns + + Args: + text: Text to search for IMDb ID + + Returns: + IMDb ID if found, None otherwise + """ + if not text: + return None + + text_lower = text.lower() + + for pattern in IMDB_PATTERNS: + match = re.search(pattern, text_lower) + if match: + imdb_id = match.group(1) + # Ensure it starts with 'tt' + if not imdb_id.startswith('tt'): + imdb_id = f'tt{imdb_id}' + + if validate_imdb_id(imdb_id): + return imdb_id + + return None + + +def extract_episode_info_from_filename(filename: str) -> Optional[Dict[str, int]]: + """ + Extract season and episode information from filename + + Args: + filename: Filename to parse + + Returns: + Dictionary with 'season' and 'episode' keys if found, None otherwise + """ + for pattern in EPISODE_PATTERNS: + match = re.search(pattern, filename, re.IGNORECASE) + if match: + try: + season = int(match.group(1)) + episode = int(match.group(2)) + + # Validate reasonable ranges + if 0 <= season <= 99 and 1 <= episode <= 999: + return {"season": season, "episode": episode} + except (ValueError, IndexError): + continue + + return None + + +def parse_nfo_with_tolerance(nfo_path: Path) -> Optional[ET.Element]: + """ + Parse NFO file with error tolerance + + Args: + nfo_path: Path to NFO file + + Returns: + XML root element if successful, None otherwise + """ + if not nfo_path.exists(): + return None + + try: + # Try normal parsing first + tree = ET.parse(nfo_path) + return tree.getroot() + except ET.ParseError as e: + _log("WARNING", f"NFO parse error for {nfo_path}: {e}. Trying with tolerance...") + + try: + # Try reading and cleaning the content + content = nfo_path.read_text(encoding='utf-8', errors='ignore') + + # Basic cleanup for common issues + content = content.replace('&', '&') # Fix unescaped ampersands + content = re.sub(r'<(\w+)([^>]*?)(?', r'<\1\2/>', content) # Fix unclosed tags + + root = ET.fromstring(content) + return root + except Exception as e: + _log("ERROR", f"Failed to parse NFO file {nfo_path} even with tolerance: {e}") + return None + + +def extract_text_from_nfo_element(root: ET.Element, xpath: str, namespaces: Optional[Dict] = None) -> Optional[str]: + """ + Extract text content from NFO element using XPath + + Args: + root: XML root element + xpath: XPath expression + namespaces: Optional namespace dictionary + + Returns: + Text content if found, None otherwise + """ + try: + if namespaces: + elements = root.findall(xpath, namespaces) + else: + elements = root.findall(xpath) + + if elements and elements[0].text: + return elements[0].text.strip() + except Exception as e: + _log("DEBUG", f"Failed to extract text from XPath {xpath}: {e}") + + return None + + +def extract_imdb_from_nfo_content(root: ET.Element) -> Optional[str]: + """ + Extract IMDb ID from NFO XML content + + Args: + root: XML root element + + Returns: + IMDb ID if found, None otherwise + """ + # Common XPath patterns for IMDb ID + imdb_xpaths = [ + './/imdb', + './/imdbid', + './/id[@type="imdb"]', + './/uniqueid[@type="imdb"]', + './/uniqueid[@default="true"]', + './/id', + './/uniqueid' + ] + + for xpath in imdb_xpaths: + imdb_text = extract_text_from_nfo_element(root, xpath) + if imdb_text: + imdb_id = extract_imdb_id_from_text(imdb_text) + if imdb_id: + return imdb_id + + # Check in plot/overview text as fallback + plot_xpaths = ['.//plot', './/overview', './/summary'] + for xpath in plot_xpaths: + plot_text = extract_text_from_nfo_element(root, xpath) + if plot_text: + imdb_id = extract_imdb_id_from_text(plot_text) + if imdb_id: + return imdb_id + + return None + + +def extract_dates_from_nfo(root: ET.Element) -> Dict[str, Optional[str]]: + """ + Extract various date fields from NFO content + + Args: + root: XML root element + + Returns: + Dictionary with date fields (premiered, aired, dateadded, etc.) + """ + date_fields = { + 'premiered': ['.//premiered', './/releasedate', './/year'], + 'aired': ['.//aired', './/firstaired'], + 'dateadded': ['.//dateadded', './/added'], + 'lastplayed': ['.//lastplayed'], + 'filelastmodified': ['.//filelastmodified'] + } + + result = {} + + for field_name, xpaths in date_fields.items(): + for xpath in xpaths: + date_text = extract_text_from_nfo_element(root, xpath) + if date_text and validate_date_string(date_text): + result[field_name] = date_text + break + else: + result[field_name] = None + + return result + + +def create_basic_nfo_structure( + media_type: str, + title: str, + imdb_id: Optional[str] = None, + dates: Optional[Dict[str, str]] = None, + additional_fields: Optional[Dict[str, str]] = None +) -> ET.Element: + """ + Create basic NFO XML structure + + Args: + media_type: Type of media ('movie', 'tvshow', 'episode') + title: Media title + imdb_id: Optional IMDb ID + dates: Optional dictionary of date fields + additional_fields: Optional additional fields to include + + Returns: + XML root element + """ + root = ET.Element(media_type) + + # Add title + title_elem = ET.SubElement(root, 'title') + title_elem.text = title + + # Add IMDb ID if provided + if imdb_id and validate_imdb_id(imdb_id): + imdb_elem = ET.SubElement(root, 'imdb') + imdb_elem.text = imdb_id + + # Also add as uniqueid + uniqueid_elem = ET.SubElement(root, 'uniqueid', type='imdb', default='true') + uniqueid_elem.text = imdb_id + + # Add dates if provided + if dates: + for field_name, date_value in dates.items(): + if date_value and validate_date_string(date_value): + date_elem = ET.SubElement(root, field_name) + date_elem.text = date_value + + # Add additional fields + if additional_fields: + for field_name, field_value in additional_fields.items(): + if field_value: + field_elem = ET.SubElement(root, field_name) + field_elem.text = str(field_value) + + return root + + +def write_nfo_file( + nfo_path: Path, + root: ET.Element, + lock_metadata: bool = True +) -> None: + """ + Write NFO XML content to file + + Args: + nfo_path: Path where to write the NFO file + root: XML root element to write + lock_metadata: Whether to add file locking attributes + + Raises: + NFOCreationError: If writing fails + """ + try: + # Ensure parent directory exists + nfo_path.parent.mkdir(parents=True, exist_ok=True) + + # Add file locking if requested + if lock_metadata: + root.set('nfoguard_managed', 'true') + root.set('last_updated', datetime.now().isoformat()) + + # Create tree and write + tree = ET.ElementTree(root) + ET.indent(tree, space=" ", level=0) # Pretty formatting + + # Write with proper XML declaration + with open(nfo_path, 'wb') as f: + tree.write(f, encoding='utf-8', xml_declaration=True) + + _log("DEBUG", f"Successfully wrote NFO file: {nfo_path}") + + except (OSError, ET.ParseError) as e: + raise NFOCreationError( + str(nfo_path), + f"Failed to write NFO file: {e}", + "unknown" + ) + + +def is_nfo_managed_by_nfoguard(nfo_path: Path) -> bool: + """ + Check if NFO file is managed by NFOGuard + + Args: + nfo_path: Path to NFO file + + Returns: + True if managed by NFOGuard, False otherwise + """ + root = parse_nfo_with_tolerance(nfo_path) + if root is None: + return False + + return root.get('nfoguard_managed') == 'true' + + +def extract_title_from_directory_name(directory_name: str) -> Optional[str]: + """ + Extract clean title from directory name + + Args: + directory_name: Directory name to parse + + Returns: + Cleaned title or None if no title found + """ + name = directory_name + + # Remove IMDb ID patterns + for pattern in IMDB_PATTERNS: + name = re.sub(pattern, '', name, flags=re.IGNORECASE) + + # Remove year in parentheses: (YYYY) + name = re.sub(r'\s*\(\d{4}\)', '', name) + + # Remove common release info patterns + release_patterns = [ + r'\s*\[.*?\]', # [1080p], [BluRay], etc. + r'\s*\{.*?\}', # {edition info} + r'\s*\(.*?\)', # (additional info) + ] + + for pattern in release_patterns: + name = re.sub(pattern, '', name, flags=re.IGNORECASE) + + # Clean up extra spaces and special characters + name = re.sub(r'[._-]+', ' ', name) # Replace dots, underscores, dashes with spaces + name = ' '.join(name.split()) # Normalize whitespace + + return name.strip() if name.strip() else None \ No newline at end of file diff --git a/utils/validation.py b/utils/validation.py new file mode 100644 index 0000000..d7322e6 --- /dev/null +++ b/utils/validation.py @@ -0,0 +1,372 @@ +""" +Validation utilities for NFOGuard +Provides runtime validation and type checking for critical paths +""" +import re +from pathlib import Path +from typing import Any, Dict, List, Optional, Union, Callable, TypeVar, Type +from datetime import datetime + +from utils.exceptions import ConfigurationError, NFOGuardException + + +T = TypeVar('T') + + +def validate_imdb_id(imdb_id: str) -> bool: + """ + Validate IMDb ID format + + Args: + imdb_id: IMDb ID to validate + + Returns: + True if valid, False otherwise + """ + if not imdb_id or not isinstance(imdb_id, str): + return False + + # Must start with 'tt' followed by 7+ digits + return bool(re.match(r'^tt\d{7,}$', imdb_id)) + + +def validate_tmdb_id(tmdb_id: str) -> bool: + """ + Validate TMDB ID format + + Args: + tmdb_id: TMDB ID to validate + + Returns: + True if valid, False otherwise + """ + if not tmdb_id or not isinstance(tmdb_id, str): + return False + + # Can be numeric or have tmdb- prefix + if tmdb_id.startswith('tmdb-'): + return tmdb_id[5:].isdigit() + return tmdb_id.isdigit() + + +def validate_season_episode(season: int, episode: int) -> bool: + """ + Validate season and episode numbers + + Args: + season: Season number + episode: Episode number + + Returns: + True if valid, False otherwise + """ + return ( + isinstance(season, int) and season >= 0 and + isinstance(episode, int) and episode >= 1 + ) + + +def validate_date_string(date_str: str) -> bool: + """ + Validate date string format (ISO format) + + Args: + date_str: Date string to validate + + Returns: + True if valid ISO date, False otherwise + """ + if not date_str or not isinstance(date_str, str): + return False + + try: + datetime.fromisoformat(date_str.replace('Z', '+00:00')) + return True + except ValueError: + return False + + +def validate_path_exists(path: Union[str, Path]) -> bool: + """ + Validate that a path exists + + Args: + path: Path to validate + + Returns: + True if path exists, False otherwise + """ + try: + return Path(path).exists() + except (OSError, ValueError): + return False + + +def validate_video_file(file_path: Union[str, Path]) -> bool: + """ + Validate that a file is a video file + + Args: + file_path: Path to validate + + Returns: + True if valid video file, False otherwise + """ + try: + path = Path(file_path) + video_extensions = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.ts'} + return path.is_file() and path.suffix.lower() in video_extensions + except (OSError, ValueError): + return False + + +def validate_webhook_payload(payload: Dict[str, Any], required_fields: List[str]) -> List[str]: + """ + Validate webhook payload has required fields + + Args: + payload: Webhook payload to validate + required_fields: List of required field names + + Returns: + List of missing field names (empty if all present) + """ + missing_fields = [] + for field in required_fields: + if field not in payload or payload[field] is None: + missing_fields.append(field) + return missing_fields + + +def validate_config_paths(paths: List[Union[str, Path]], path_type: str) -> None: + """ + Validate configuration paths exist and are directories + + Args: + paths: List of paths to validate + path_type: Type of paths for error messages (e.g., "TV", "Movie") + + Raises: + ConfigurationError: If any paths are invalid + """ + invalid_paths = [] + + for path in paths: + try: + path_obj = Path(path) + if not path_obj.exists(): + invalid_paths.append(f"{path} (does not exist)") + elif not path_obj.is_dir(): + invalid_paths.append(f"{path} (not a directory)") + except (OSError, ValueError) as e: + invalid_paths.append(f"{path} (invalid: {e})") + + if invalid_paths: + raise ConfigurationError( + f"{path_type.lower()}_paths", + f"Invalid {path_type} paths found", + {"invalid_paths": invalid_paths} + ) + + +def require_type(value: Any, expected_type: Type[T], name: str) -> T: + """ + Require value to be of specific type + + Args: + value: Value to check + expected_type: Expected type + name: Name of the value for error messages + + Returns: + The value if it matches the type + + Raises: + TypeError: If value is not of expected type + """ + if not isinstance(value, expected_type): + raise TypeError( + f"{name} must be {expected_type.__name__}, got {type(value).__name__}" + ) + return value + + +def require_non_empty(value: Optional[str], name: str) -> str: + """ + Require string value to be non-empty + + Args: + value: String value to check + name: Name of the value for error messages + + Returns: + The value if it's non-empty + + Raises: + ValueError: If value is None or empty + """ + if not value: + raise ValueError(f"{name} cannot be None or empty") + return value + + +def validate_and_clean_imdb_id(imdb_id: Optional[str]) -> Optional[str]: + """ + Validate and clean IMDb ID + + Args: + imdb_id: IMDb ID to validate and clean + + Returns: + Cleaned IMDb ID or None if invalid + """ + if not imdb_id: + return None + + # Clean the ID + cleaned = imdb_id.strip().lower() + + # Remove common prefixes + if cleaned.startswith('imdb-'): + cleaned = cleaned[5:] + elif cleaned.startswith('imdb_'): + cleaned = cleaned[5:] + + # Ensure it starts with 'tt' + if not cleaned.startswith('tt'): + cleaned = f'tt{cleaned}' + + # Validate format + if validate_imdb_id(cleaned): + return cleaned + + return None + + +def create_validator(validation_func: Callable[[Any], bool], error_message: str) -> Callable: + """ + Create a validator decorator + + Args: + validation_func: Function that returns True if value is valid + error_message: Error message to raise if validation fails + + Returns: + Decorator function + """ + def decorator(func: Callable) -> Callable: + def wrapper(*args, **kwargs): + # Apply validation to first argument + if args and not validation_func(args[0]): + raise ValueError(error_message.format(args[0])) + return func(*args, **kwargs) + return wrapper + return decorator + + +# Common validators +validate_imdb_required = create_validator( + lambda x: validate_imdb_id(x), + "Invalid IMDb ID format: {}" +) + +validate_path_required = create_validator( + lambda x: validate_path_exists(x), + "Path does not exist: {}" +) + +validate_date_required = create_validator( + lambda x: validate_date_string(x), + "Invalid date format: {}" +) + + +class ValidationError(NFOGuardException): + """Raised when validation fails""" + + def __init__(self, field_name: str, value: Any, reason: str): + details = { + "field_name": field_name, + "value": str(value), + "reason": reason + } + message = f"Validation failed for {field_name}: {reason}" + super().__init__(message, details) + + +def validate_episode_file_pattern(filename: str) -> Optional[Dict[str, int]]: + """ + Validate and extract episode information from filename + + Args: + filename: Filename to validate + + Returns: + Dictionary with season and episode if valid, None otherwise + """ + # Episode patterns + patterns = [ + r'[sS](\d{1,2})[eE](\d{1,3})', # S01E01 + r'(\d{1,2})x(\d{1,3})', # 1x01 + r'[sS](\d{1,2})\.?[eE](\d{1,3})', # S01.E01 + ] + + for pattern in patterns: + match = re.search(pattern, filename) + if match: + season = int(match.group(1)) + episode = int(match.group(2)) + if validate_season_episode(season, episode): + return {"season": season, "episode": episode} + + return None + + +def sanitize_filename(filename: str) -> str: + """ + Sanitize filename by removing invalid characters + + Args: + filename: Filename to sanitize + + Returns: + Sanitized filename + """ + # Remove invalid characters for most filesystems + invalid_chars = r'<>:"/\|?*' + for char in invalid_chars: + filename = filename.replace(char, '_') + + # Remove leading/trailing dots and spaces + filename = filename.strip('. ') + + # Ensure it's not empty + if not filename: + filename = 'unnamed' + + return filename + + +def validate_url_format(url: str) -> bool: + """ + Validate URL format + + Args: + url: URL to validate + + Returns: + True if valid URL format, False otherwise + """ + if not url or not isinstance(url, str): + return False + + # Basic URL validation + url_pattern = re.compile( + r'^https?://' # http:// or https:// + r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' # domain... + r'localhost|' # localhost... + r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip + r'(?::\d+)?' # optional port + r'(?:/?|[/?]\S+)$', re.IGNORECASE) + + return bool(url_pattern.match(url)) \ No newline at end of file