sbcrumb
fa1e4e2ec7
Build & Push DEV to DockerHub / docker (push) Failing after 1m12s
Reviewed-on: #15
372 lines
9.3 KiB
Python
372 lines
9.3 KiB
Python
"""
|
|
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)) |