improvements #16
@@ -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"""
|
||||
|
||||
+17
-81
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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())
|
||||
@@ -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
|
||||
@@ -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))
|
||||
Reference in New Issue
Block a user