feat: add config validation
This commit is contained in:
+241
-23
@@ -1,10 +1,17 @@
|
||||
"""
|
||||
NFOGuard Configuration Module
|
||||
Handles all configuration loading and validation
|
||||
Handles all configuration loading and validation with comprehensive error reporting
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
from typing import List, Optional, Dict, Any
|
||||
|
||||
from utils.exceptions import ConfigurationError
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _bool_env(name: str, default: bool) -> bool:
|
||||
@@ -16,20 +23,30 @@ def _bool_env(name: str, default: bool) -> bool:
|
||||
|
||||
|
||||
class NFOGuardConfig:
|
||||
"""Configuration class for NFOGuard"""
|
||||
"""Configuration class for NFOGuard with integrated validation"""
|
||||
|
||||
def __init__(self):
|
||||
# Paths - No hardcoded defaults, must be configured via environment
|
||||
tv_paths_env = os.environ.get("TV_PATHS", "")
|
||||
movie_paths_env = os.environ.get("MOVIE_PATHS", "")
|
||||
def __init__(self, validate_on_init: bool = True, strict_validation: bool = False):
|
||||
"""
|
||||
Initialize NFOGuard configuration
|
||||
|
||||
if not tv_paths_env:
|
||||
raise ValueError("TV_PATHS environment variable is required but not set")
|
||||
if not movie_paths_env:
|
||||
raise ValueError("MOVIE_PATHS environment variable is required but not set")
|
||||
|
||||
self.tv_paths = [Path(p.strip()) for p in tv_paths_env.split(",") if p.strip()]
|
||||
self.movie_paths = [Path(p.strip()) for p in movie_paths_env.split(",") if p.strip()]
|
||||
Args:
|
||||
validate_on_init: Run validation during initialization
|
||||
strict_validation: Treat warnings as errors
|
||||
"""
|
||||
self.strict_validation = strict_validation
|
||||
self._validation_issues = []
|
||||
|
||||
# Initialize configuration
|
||||
self._load_configuration()
|
||||
|
||||
# Run validation if requested
|
||||
if validate_on_init:
|
||||
self._validate_configuration()
|
||||
|
||||
def _load_configuration(self) -> None:
|
||||
"""Load all configuration from environment variables"""
|
||||
# Core paths - Required
|
||||
self._load_paths()
|
||||
|
||||
# Core settings
|
||||
self.manage_nfo = _bool_env("MANAGE_NFO", True)
|
||||
@@ -38,28 +55,229 @@ class NFOGuardConfig:
|
||||
self.debug = _bool_env("DEBUG", False)
|
||||
self.manager_brand = os.environ.get("MANAGER_BRAND", "NFOGuard")
|
||||
|
||||
# Batching
|
||||
self.batch_delay = float(os.environ.get("BATCH_DELAY", "5.0"))
|
||||
self.max_concurrent = int(os.environ.get("MAX_CONCURRENT_SERIES", "3"))
|
||||
# Batching and performance
|
||||
self.batch_delay = self._get_float_env("BATCH_DELAY", 5.0, 0.1, 300.0)
|
||||
self.max_concurrent = self._get_int_env("MAX_CONCURRENT_SERIES", 3, 1, 10)
|
||||
|
||||
# Database
|
||||
self.db_path = Path(os.environ.get("DB_PATH", "/app/data/media_dates.db"))
|
||||
|
||||
# External connections
|
||||
self._load_external_connections()
|
||||
|
||||
# Movie processing
|
||||
self._load_movie_settings()
|
||||
|
||||
# TV processing
|
||||
self._load_tv_settings()
|
||||
|
||||
def _load_paths(self) -> None:
|
||||
"""Load and validate path configuration"""
|
||||
tv_paths_env = os.environ.get("TV_PATHS", "")
|
||||
movie_paths_env = os.environ.get("MOVIE_PATHS", "")
|
||||
|
||||
if not tv_paths_env:
|
||||
raise ConfigurationError(
|
||||
setting="TV_PATHS",
|
||||
reason="TV_PATHS environment variable is required but not set"
|
||||
)
|
||||
|
||||
if not movie_paths_env:
|
||||
raise ConfigurationError(
|
||||
setting="MOVIE_PATHS",
|
||||
reason="MOVIE_PATHS environment variable is required but not set"
|
||||
)
|
||||
|
||||
# Parse paths
|
||||
self.tv_paths = [Path(p.strip()) for p in tv_paths_env.split(",") if p.strip()]
|
||||
self.movie_paths = [Path(p.strip()) for p in movie_paths_env.split(",") if p.strip()]
|
||||
|
||||
if not self.tv_paths:
|
||||
raise ConfigurationError(
|
||||
setting="TV_PATHS",
|
||||
reason="No valid TV paths found after parsing",
|
||||
current_value=tv_paths_env
|
||||
)
|
||||
|
||||
if not self.movie_paths:
|
||||
raise ConfigurationError(
|
||||
setting="MOVIE_PATHS",
|
||||
reason="No valid movie paths found after parsing",
|
||||
current_value=movie_paths_env
|
||||
)
|
||||
|
||||
def _load_external_connections(self) -> None:
|
||||
"""Load external API and database connection settings"""
|
||||
# API URLs
|
||||
self.radarr_url = os.environ.get("RADARR_URL", "")
|
||||
self.sonarr_url = os.environ.get("SONARR_URL", "")
|
||||
self.jellyseerr_url = os.environ.get("JELLYSEERR_URL", "")
|
||||
|
||||
# Database settings
|
||||
self.db_type = os.environ.get("RADARR_DB_TYPE", "").lower()
|
||||
self.db_host = os.environ.get("RADARR_DB_HOST", "")
|
||||
self.db_port = self._get_int_env("RADARR_DB_PORT", 5432, 1, 65535)
|
||||
self.db_name = os.environ.get("RADARR_DB_NAME", "")
|
||||
self.db_user = os.environ.get("RADARR_DB_USER", "")
|
||||
|
||||
# Timeout settings
|
||||
self.timeout_seconds = self._get_int_env("TIMEOUT_SECONDS", 45, 10, 300)
|
||||
|
||||
def _load_movie_settings(self) -> None:
|
||||
"""Load movie processing settings"""
|
||||
self.movie_priority = os.environ.get("MOVIE_PRIORITY", "import_then_digital").lower()
|
||||
self.prefer_release_dates_over_file_dates = _bool_env("PREFER_RELEASE_DATES_OVER_FILE_DATES", True)
|
||||
self.allow_file_date_fallback = _bool_env("ALLOW_FILE_DATE_FALLBACK", False)
|
||||
self.release_date_priority = [p.strip() for p in os.environ.get("RELEASE_DATE_PRIORITY", "digital,physical,theatrical").split(",")]
|
||||
|
||||
# Release date settings
|
||||
release_priority_env = os.environ.get("RELEASE_DATE_PRIORITY", "digital,physical,theatrical")
|
||||
self.release_date_priority = [p.strip() for p in release_priority_env.split(",") if p.strip()]
|
||||
|
||||
self.enable_smart_date_validation = _bool_env("ENABLE_SMART_DATE_VALIDATION", True)
|
||||
self.max_release_date_gap_years = int(os.environ.get("MAX_RELEASE_DATE_GAP_YEARS", "10"))
|
||||
self.max_release_date_gap_years = self._get_int_env("MAX_RELEASE_DATE_GAP_YEARS", 10, 1, 50)
|
||||
self.movie_poll_mode = os.environ.get("MOVIE_POLL_MODE", "always").lower()
|
||||
self.movie_update_mode = os.environ.get("MOVIE_DATE_UPDATE_MODE", "backfill_only").lower()
|
||||
|
||||
# TV processing
|
||||
|
||||
def _load_tv_settings(self) -> None:
|
||||
"""Load TV processing settings"""
|
||||
self.tv_season_dir_format = os.environ.get("TV_SEASON_DIR_FORMAT", "Season {season:02d}")
|
||||
self.tv_season_dir_pattern = os.environ.get("TV_SEASON_DIR_PATTERN", "season ").lower()
|
||||
self.tv_webhook_processing_mode = os.environ.get("TV_WEBHOOK_PROCESSING_MODE", "targeted").lower()
|
||||
|
||||
def _get_int_env(self, name: str, default: int, min_val: int, max_val: int) -> int:
|
||||
"""Get integer environment variable with validation"""
|
||||
value_str = os.environ.get(name)
|
||||
if not value_str:
|
||||
return default
|
||||
|
||||
try:
|
||||
value = int(value_str)
|
||||
if value < min_val or value > max_val:
|
||||
raise ConfigurationError(
|
||||
setting=name,
|
||||
reason=f"Value must be between {min_val} and {max_val}",
|
||||
current_value=value_str
|
||||
)
|
||||
return value
|
||||
except ValueError:
|
||||
raise ConfigurationError(
|
||||
setting=name,
|
||||
reason=f"Invalid integer value",
|
||||
current_value=value_str
|
||||
)
|
||||
|
||||
def _get_float_env(self, name: str, default: float, min_val: float, max_val: float) -> float:
|
||||
"""Get float environment variable with validation"""
|
||||
value_str = os.environ.get(name)
|
||||
if not value_str:
|
||||
return default
|
||||
|
||||
try:
|
||||
value = float(value_str)
|
||||
if value < min_val or value > max_val:
|
||||
raise ConfigurationError(
|
||||
setting=name,
|
||||
reason=f"Value must be between {min_val} and {max_val}",
|
||||
current_value=value_str
|
||||
)
|
||||
return value
|
||||
except ValueError:
|
||||
raise ConfigurationError(
|
||||
setting=name,
|
||||
reason=f"Invalid float value",
|
||||
current_value=value_str
|
||||
)
|
||||
|
||||
def _validate_configuration(self) -> None:
|
||||
"""Validate configuration using the validator"""
|
||||
try:
|
||||
# Import here to avoid circular imports
|
||||
from config.validator import validate_configuration_and_raise
|
||||
validate_configuration_and_raise()
|
||||
|
||||
except ImportError:
|
||||
# Fallback to basic validation if validator not available
|
||||
logger.warning("Configuration validator not available, using basic validation")
|
||||
self._basic_validation()
|
||||
except ConfigurationError:
|
||||
if self.strict_validation:
|
||||
raise
|
||||
else:
|
||||
# Log warning but continue
|
||||
logger.warning("Configuration validation found issues", exc_info=True)
|
||||
|
||||
def _basic_validation(self) -> None:
|
||||
"""Basic fallback validation"""
|
||||
# Validate that paths exist (basic check)
|
||||
for path_list, path_type in [(self.tv_paths, "TV"), (self.movie_paths, "Movie")]:
|
||||
for path in path_list:
|
||||
if not path.is_absolute():
|
||||
logger.warning(f"{path_type} path should be absolute: {path}")
|
||||
|
||||
def get_configuration_summary(self) -> Dict[str, Any]:
|
||||
"""Get a summary of current configuration"""
|
||||
return {
|
||||
"tv_paths": [str(p) for p in self.tv_paths],
|
||||
"movie_paths": [str(p) for p in self.movie_paths],
|
||||
"database_path": str(self.db_path),
|
||||
"external_apis": {
|
||||
"radarr": bool(self.radarr_url),
|
||||
"sonarr": bool(self.sonarr_url),
|
||||
"jellyseerr": bool(self.jellyseerr_url)
|
||||
},
|
||||
"database_connection": {
|
||||
"type": self.db_type,
|
||||
"configured": bool(self.db_type and self.db_host)
|
||||
},
|
||||
"performance": {
|
||||
"batch_delay": self.batch_delay,
|
||||
"max_concurrent": self.max_concurrent,
|
||||
"timeout_seconds": self.timeout_seconds
|
||||
},
|
||||
"features": {
|
||||
"manage_nfo": self.manage_nfo,
|
||||
"fix_dir_mtimes": self.fix_dir_mtimes,
|
||||
"lock_metadata": self.lock_metadata,
|
||||
"debug": self.debug
|
||||
}
|
||||
}
|
||||
|
||||
def validate_runtime_access(self) -> Dict[str, bool]:
|
||||
"""Quick runtime validation of critical paths"""
|
||||
results = {
|
||||
"tv_paths_accessible": True,
|
||||
"movie_paths_accessible": True,
|
||||
"database_writable": True
|
||||
}
|
||||
|
||||
# Test TV paths
|
||||
for path in self.tv_paths:
|
||||
if not path.exists() or not path.is_dir():
|
||||
results["tv_paths_accessible"] = False
|
||||
break
|
||||
|
||||
# Test movie paths
|
||||
for path in self.movie_paths:
|
||||
if not path.exists() or not path.is_dir():
|
||||
results["movie_paths_accessible"] = False
|
||||
break
|
||||
|
||||
# Test database directory
|
||||
db_dir = self.db_path.parent
|
||||
try:
|
||||
if not db_dir.exists():
|
||||
db_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Test write access
|
||||
test_file = db_dir / ".nfoguard_write_test"
|
||||
test_file.write_text("test")
|
||||
test_file.unlink()
|
||||
except (PermissionError, OSError):
|
||||
results["database_writable"] = False
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# Global config instance
|
||||
config = NFOGuardConfig()
|
||||
# Global config instance - Initialize with validation disabled by default for backwards compatibility
|
||||
# Applications can enable validation by creating their own instance with validate_on_init=True
|
||||
config = NFOGuardConfig(validate_on_init=False)
|
||||
Reference in New Issue
Block a user