498 lines
20 KiB
Python
498 lines
20 KiB
Python
"""
|
|
Configuration Validator for NFOGuard
|
|
Provides comprehensive validation of all configuration settings with detailed error reporting
|
|
"""
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Dict, List, Any, Optional, Union, Type, Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
|
|
from utils.exceptions import ConfigurationError
|
|
from utils.validation import validate_url_format
|
|
|
|
|
|
class ValidationSeverity(Enum):
|
|
"""Severity levels for validation issues"""
|
|
ERROR = "error" # Configuration is invalid, will cause failures
|
|
WARNING = "warning" # Configuration may cause issues but is workable
|
|
INFO = "info" # Configuration could be improved
|
|
|
|
|
|
@dataclass
|
|
class ValidationIssue:
|
|
"""Represents a configuration validation issue"""
|
|
setting: str
|
|
severity: ValidationSeverity
|
|
message: str
|
|
current_value: Any = None
|
|
suggested_value: Any = None
|
|
details: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for structured logging"""
|
|
return {
|
|
"setting": self.setting,
|
|
"severity": self.severity.value,
|
|
"message": self.message,
|
|
"current_value": str(self.current_value) if self.current_value is not None else None,
|
|
"suggested_value": str(self.suggested_value) if self.suggested_value is not None else None,
|
|
"details": self.details
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ValidationResult:
|
|
"""Results of configuration validation"""
|
|
is_valid: bool
|
|
issues: List[ValidationIssue] = field(default_factory=list)
|
|
warnings_count: int = 0
|
|
errors_count: int = 0
|
|
|
|
def add_issue(self, issue: ValidationIssue) -> None:
|
|
"""Add a validation issue"""
|
|
self.issues.append(issue)
|
|
if issue.severity == ValidationSeverity.ERROR:
|
|
self.errors_count += 1
|
|
self.is_valid = False
|
|
elif issue.severity == ValidationSeverity.WARNING:
|
|
self.warnings_count += 1
|
|
|
|
def get_errors(self) -> List[ValidationIssue]:
|
|
"""Get only error-level issues"""
|
|
return [issue for issue in self.issues if issue.severity == ValidationSeverity.ERROR]
|
|
|
|
def get_warnings(self) -> List[ValidationIssue]:
|
|
"""Get only warning-level issues"""
|
|
return [issue for issue in self.issues if issue.severity == ValidationSeverity.WARNING]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for structured logging"""
|
|
return {
|
|
"is_valid": self.is_valid,
|
|
"errors_count": self.errors_count,
|
|
"warnings_count": self.warnings_count,
|
|
"issues": [issue.to_dict() for issue in self.issues]
|
|
}
|
|
|
|
|
|
class ConfigValidator:
|
|
"""Comprehensive configuration validator for NFOGuard"""
|
|
|
|
def __init__(self):
|
|
self.result = ValidationResult(is_valid=True)
|
|
|
|
# Define validation rules
|
|
self._path_settings = {
|
|
"TV_PATHS", "MOVIE_PATHS", "RADARR_ROOT_FOLDERS",
|
|
"SONARR_ROOT_FOLDERS", "DB_PATH", "LOG_DIR"
|
|
}
|
|
|
|
self._url_settings = {
|
|
"RADARR_URL", "SONARR_URL", "JELLYSEERR_URL"
|
|
}
|
|
|
|
self._required_settings = {
|
|
"TV_PATHS", "MOVIE_PATHS"
|
|
}
|
|
|
|
self._numeric_settings = {
|
|
"BATCH_DELAY": (float, 0.1, 300.0),
|
|
"MAX_CONCURRENT_SERIES": (int, 1, 10),
|
|
"TIMEOUT_SECONDS": (int, 10, 300),
|
|
"PORT": (int, 1024, 65535),
|
|
"RADARR_DB_PORT": (int, 1, 65535),
|
|
"MAX_RELEASE_DATE_GAP_YEARS": (int, 1, 50)
|
|
}
|
|
|
|
self._boolean_settings = {
|
|
"MANAGE_NFO", "FIX_DIR_MTIMES", "LOCK_METADATA", "DEBUG",
|
|
"PREFER_RELEASE_DATES_OVER_FILE_DATES", "ALLOW_FILE_DATE_FALLBACK",
|
|
"ENABLE_SMART_DATE_VALIDATION", "PATH_DEBUG", "SUPPRESS_TVDB_WARNINGS"
|
|
}
|
|
|
|
self._choice_settings = {
|
|
"MOVIE_PRIORITY": ["import_then_digital", "digital_first", "file_date_only"],
|
|
"MOVIE_POLL_MODE": ["always", "missing_only", "never"],
|
|
"MOVIE_DATE_UPDATE_MODE": ["overwrite", "backfill_only", "preserve_existing"],
|
|
"TV_WEBHOOK_PROCESSING_MODE": ["targeted", "full_scan", "hybrid"],
|
|
"UPDATE_MODE": ["always", "missing_only", "never"],
|
|
"MTIME_BEHAVIOR": ["update", "leave_alone"],
|
|
"RADARR_DB_TYPE": ["postgresql", "sqlite"]
|
|
}
|
|
|
|
def validate_all(self) -> ValidationResult:
|
|
"""Validate all configuration settings"""
|
|
self.result = ValidationResult(is_valid=True)
|
|
|
|
# Validate required settings
|
|
self._validate_required_settings()
|
|
|
|
# Validate paths
|
|
self._validate_paths()
|
|
|
|
# Validate URLs
|
|
self._validate_urls()
|
|
|
|
# Validate numeric settings
|
|
self._validate_numeric_settings()
|
|
|
|
# Validate boolean settings
|
|
self._validate_boolean_settings()
|
|
|
|
# Validate choice settings
|
|
self._validate_choice_settings()
|
|
|
|
# Validate database configuration
|
|
self._validate_database_config()
|
|
|
|
# Validate release date configuration
|
|
self._validate_release_date_config()
|
|
|
|
# Validate performance settings
|
|
self._validate_performance_settings()
|
|
|
|
# Validate cross-setting dependencies
|
|
self._validate_dependencies()
|
|
|
|
return self.result
|
|
|
|
def _validate_required_settings(self) -> None:
|
|
"""Validate required environment variables are set"""
|
|
for setting in self._required_settings:
|
|
value = os.environ.get(setting)
|
|
if not value:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Required setting {setting} is not set",
|
|
current_value=value
|
|
))
|
|
elif not value.strip():
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Required setting {setting} is empty",
|
|
current_value=value
|
|
))
|
|
|
|
def _validate_paths(self) -> None:
|
|
"""Validate all path-related settings"""
|
|
for setting in self._path_settings:
|
|
value = os.environ.get(setting)
|
|
if not value:
|
|
if setting in self._required_settings:
|
|
continue # Already handled in required validation
|
|
else:
|
|
# Optional path settings
|
|
continue
|
|
|
|
if setting in {"TV_PATHS", "MOVIE_PATHS", "RADARR_ROOT_FOLDERS", "SONARR_ROOT_FOLDERS"}:
|
|
# Multi-path settings
|
|
paths = [p.strip() for p in value.split(",") if p.strip()]
|
|
if not paths:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"No valid paths found in {setting}",
|
|
current_value=value
|
|
))
|
|
continue
|
|
|
|
for path_str in paths:
|
|
self._validate_single_path(setting, path_str)
|
|
else:
|
|
# Single path settings
|
|
self._validate_single_path(setting, value)
|
|
|
|
def _validate_single_path(self, setting: str, path_str: str) -> None:
|
|
"""Validate a single path"""
|
|
try:
|
|
path = Path(path_str)
|
|
|
|
# Check if path is absolute (recommended for Docker)
|
|
if not path.is_absolute():
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.WARNING,
|
|
message=f"Path should be absolute for reliable Docker operation",
|
|
current_value=path_str,
|
|
suggested_value=f"Use absolute path like /media/..."
|
|
))
|
|
|
|
# For media paths, check existence if not in container
|
|
if setting in {"TV_PATHS", "MOVIE_PATHS"} and not self._is_likely_container_path(path_str):
|
|
if not path.exists():
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.WARNING,
|
|
message=f"Path does not exist (may be valid in container)",
|
|
current_value=path_str,
|
|
details={"path_type": "media"}
|
|
))
|
|
elif not path.is_dir():
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Path exists but is not a directory",
|
|
current_value=path_str
|
|
))
|
|
|
|
# Check for database directory
|
|
if setting == "DB_PATH":
|
|
parent_dir = path.parent
|
|
if not self._is_likely_container_path(str(parent_dir)) and not parent_dir.exists():
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.WARNING,
|
|
message=f"Database directory does not exist: {parent_dir}",
|
|
current_value=path_str
|
|
))
|
|
|
|
except (OSError, ValueError) as e:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Invalid path format: {e}",
|
|
current_value=path_str
|
|
))
|
|
|
|
def _is_likely_container_path(self, path: str) -> bool:
|
|
"""Check if path looks like a container path"""
|
|
container_indicators = ["/app/", "/media/", "/config/", "/data/"]
|
|
return any(indicator in path for indicator in container_indicators)
|
|
|
|
def _validate_urls(self) -> None:
|
|
"""Validate URL settings"""
|
|
for setting in self._url_settings:
|
|
value = os.environ.get(setting)
|
|
if value and not validate_url_format(value):
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Invalid URL format",
|
|
current_value=value,
|
|
suggested_value="Use format: http://hostname:port or https://hostname:port"
|
|
))
|
|
|
|
def _validate_numeric_settings(self) -> None:
|
|
"""Validate numeric settings"""
|
|
for setting, (type_class, min_val, max_val) in self._numeric_settings.items():
|
|
value = os.environ.get(setting)
|
|
if not value:
|
|
continue
|
|
|
|
try:
|
|
numeric_value = type_class(value)
|
|
if numeric_value < min_val or numeric_value > max_val:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Value must be between {min_val} and {max_val}",
|
|
current_value=value,
|
|
suggested_value=f"Use value between {min_val}-{max_val}"
|
|
))
|
|
except (ValueError, TypeError):
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Invalid {type_class.__name__} value",
|
|
current_value=value,
|
|
suggested_value=f"Use a valid {type_class.__name__} between {min_val}-{max_val}"
|
|
))
|
|
|
|
def _validate_boolean_settings(self) -> None:
|
|
"""Validate boolean settings"""
|
|
valid_true = {"1", "true", "yes", "y", "on"}
|
|
valid_false = {"0", "false", "no", "n", "off"}
|
|
valid_values = valid_true | valid_false
|
|
|
|
for setting in self._boolean_settings:
|
|
value = os.environ.get(setting)
|
|
if value and value.lower() not in valid_values:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Invalid boolean value",
|
|
current_value=value,
|
|
suggested_value="Use: true/false, 1/0, yes/no, y/n, on/off"
|
|
))
|
|
|
|
def _validate_choice_settings(self) -> None:
|
|
"""Validate settings with predefined choices"""
|
|
for setting, valid_choices in self._choice_settings.items():
|
|
value = os.environ.get(setting)
|
|
if value and value.lower() not in [choice.lower() for choice in valid_choices]:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Invalid choice",
|
|
current_value=value,
|
|
suggested_value=f"Use one of: {', '.join(valid_choices)}"
|
|
))
|
|
|
|
def _validate_database_config(self) -> None:
|
|
"""Validate database configuration"""
|
|
db_type = os.environ.get("RADARR_DB_TYPE", "").lower()
|
|
|
|
if db_type == "postgresql":
|
|
# Check required PostgreSQL settings
|
|
required_pg_settings = ["RADARR_DB_HOST", "RADARR_DB_NAME", "RADARR_DB_USER"]
|
|
for setting in required_pg_settings:
|
|
if not os.environ.get(setting):
|
|
self.result.add_issue(ValidationIssue(
|
|
setting=setting,
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Required for PostgreSQL database connection",
|
|
current_value=os.environ.get(setting)
|
|
))
|
|
|
|
def _validate_release_date_config(self) -> None:
|
|
"""Validate release date processing configuration"""
|
|
priority = os.environ.get("RELEASE_DATE_PRIORITY", "")
|
|
if priority:
|
|
priorities = [p.strip().lower() for p in priority.split(",")]
|
|
valid_priorities = {"digital", "physical", "theatrical"}
|
|
|
|
invalid_priorities = [p for p in priorities if p not in valid_priorities]
|
|
if invalid_priorities:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting="RELEASE_DATE_PRIORITY",
|
|
severity=ValidationSeverity.ERROR,
|
|
message=f"Invalid release date priorities: {', '.join(invalid_priorities)}",
|
|
current_value=priority,
|
|
suggested_value="Use: digital, physical, theatrical"
|
|
))
|
|
|
|
if len(set(priorities)) != len(priorities):
|
|
self.result.add_issue(ValidationIssue(
|
|
setting="RELEASE_DATE_PRIORITY",
|
|
severity=ValidationSeverity.WARNING,
|
|
message="Duplicate priorities found",
|
|
current_value=priority
|
|
))
|
|
|
|
def _validate_performance_settings(self) -> None:
|
|
"""Validate performance-related settings"""
|
|
batch_delay = os.environ.get("BATCH_DELAY")
|
|
max_concurrent = os.environ.get("MAX_CONCURRENT_SERIES")
|
|
|
|
# Performance recommendations
|
|
if batch_delay:
|
|
try:
|
|
delay = float(batch_delay)
|
|
if delay < 1.0:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting="BATCH_DELAY",
|
|
severity=ValidationSeverity.WARNING,
|
|
message="Very low batch delay may increase system load",
|
|
current_value=batch_delay,
|
|
suggested_value="Consider using 1.0 or higher for better stability"
|
|
))
|
|
except ValueError:
|
|
pass # Already caught in numeric validation
|
|
|
|
if max_concurrent:
|
|
try:
|
|
concurrent = int(max_concurrent)
|
|
if concurrent > 5:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting="MAX_CONCURRENT_SERIES",
|
|
severity=ValidationSeverity.WARNING,
|
|
message="High concurrency may overload system resources",
|
|
current_value=max_concurrent,
|
|
suggested_value="Consider using 3-5 for optimal balance"
|
|
))
|
|
except ValueError:
|
|
pass # Already caught in numeric validation
|
|
|
|
def _validate_dependencies(self) -> None:
|
|
"""Validate cross-setting dependencies"""
|
|
# If database is configured, recommend using database over API
|
|
db_type = os.environ.get("RADARR_DB_TYPE")
|
|
radarr_url = os.environ.get("RADARR_URL")
|
|
|
|
if db_type and radarr_url:
|
|
self.result.add_issue(ValidationIssue(
|
|
setting="RADARR_DB_TYPE",
|
|
severity=ValidationSeverity.INFO,
|
|
message="Database connection preferred over API for better performance",
|
|
details={"recommendation": "Database access is faster and more reliable"}
|
|
))
|
|
|
|
# Check path mapping consistency
|
|
tv_paths = os.environ.get("TV_PATHS", "").split(",")
|
|
sonarr_paths = os.environ.get("SONARR_ROOT_FOLDERS", "").split(",")
|
|
|
|
if len([p for p in tv_paths if p.strip()]) != len([p for p in sonarr_paths if p.strip()]):
|
|
self.result.add_issue(ValidationIssue(
|
|
setting="TV_PATHS",
|
|
severity=ValidationSeverity.WARNING,
|
|
message="TV_PATHS and SONARR_ROOT_FOLDERS should have matching number of paths",
|
|
details={
|
|
"tv_paths_count": len([p for p in tv_paths if p.strip()]),
|
|
"sonarr_paths_count": len([p for p in sonarr_paths if p.strip()])
|
|
}
|
|
))
|
|
|
|
|
|
def validate_configuration() -> ValidationResult:
|
|
"""
|
|
Validate the complete NFOGuard configuration
|
|
|
|
Returns:
|
|
ValidationResult with all validation issues found
|
|
"""
|
|
validator = ConfigValidator()
|
|
return validator.validate_all()
|
|
|
|
|
|
def validate_configuration_and_raise() -> None:
|
|
"""
|
|
Validate configuration and raise ConfigurationError if invalid
|
|
|
|
Raises:
|
|
ConfigurationError: If configuration validation fails
|
|
"""
|
|
result = validate_configuration()
|
|
|
|
if not result.is_valid:
|
|
error_messages = []
|
|
for error in result.get_errors():
|
|
error_messages.append(f"{error.setting}: {error.message}")
|
|
|
|
raise ConfigurationError(
|
|
setting="configuration",
|
|
reason=f"Configuration validation failed with {result.errors_count} errors",
|
|
current_value={
|
|
"errors": error_messages,
|
|
"warnings_count": result.warnings_count,
|
|
"validation_details": result.to_dict()
|
|
}
|
|
)
|
|
|
|
|
|
def get_configuration_summary() -> Dict[str, Any]:
|
|
"""
|
|
Get a summary of current configuration status
|
|
|
|
Returns:
|
|
Dictionary with configuration summary
|
|
"""
|
|
result = validate_configuration()
|
|
|
|
return {
|
|
"is_valid": result.is_valid,
|
|
"errors_count": result.errors_count,
|
|
"warnings_count": result.warnings_count,
|
|
"total_issues": len(result.issues),
|
|
"critical_issues": [
|
|
issue.to_dict() for issue in result.issues
|
|
if issue.severity == ValidationSeverity.ERROR
|
|
],
|
|
"recommendations": [
|
|
issue.to_dict() for issue in result.issues
|
|
if issue.severity == ValidationSeverity.WARNING
|
|
]
|
|
} |