From 81433e0b7d535318c0a7ece24c45b1fe3e4521eb Mon Sep 17 00:00:00 2001 From: sbcrumb Date: Mon, 29 Sep 2025 08:53:25 -0400 Subject: [PATCH] feat: add config validation --- config/runtime_validator.py | 547 ++++++++++++++++++++++++++++++++++++ config/settings.py | 264 +++++++++++++++-- config/validation_cli.py | 293 +++++++++++++++++++ config/validator.py | 498 ++++++++++++++++++++++++++++++++ 4 files changed, 1579 insertions(+), 23 deletions(-) create mode 100644 config/runtime_validator.py create mode 100755 config/validation_cli.py create mode 100644 config/validator.py diff --git a/config/runtime_validator.py b/config/runtime_validator.py new file mode 100644 index 0000000..a01c019 --- /dev/null +++ b/config/runtime_validator.py @@ -0,0 +1,547 @@ +""" +Runtime Configuration Validator for NFOGuard +Provides validation of configuration at runtime with health checks +""" +import asyncio +import time +from pathlib import Path +from typing import Dict, List, Any, Optional, Tuple +from dataclasses import dataclass, field +from contextlib import asynccontextmanager +import aiohttp +import sqlite3 +import psycopg2 + +from config.validator import ValidationIssue, ValidationResult, ValidationSeverity +from utils.exceptions import ConfigurationError, NetworkRetryableError, DatabaseError + + +@dataclass +class HealthCheckResult: + """Result of a runtime health check""" + component: str + is_healthy: bool + response_time_ms: Optional[float] = None + message: str = "" + details: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "component": self.component, + "is_healthy": self.is_healthy, + "response_time_ms": self.response_time_ms, + "message": self.message, + "details": self.details, + "timestamp": time.time() + } + + +class RuntimeValidator: + """Runtime configuration and system health validator""" + + def __init__(self, config): + self.config = config + self._health_cache = {} + self._cache_ttl = 60 # Cache health results for 60 seconds + + async def validate_runtime_config(self) -> ValidationResult: + """Perform runtime validation of configuration""" + result = ValidationResult(is_valid=True) + + # Validate filesystem access + await self._validate_filesystem_access(result) + + # Validate database connectivity + await self._validate_database_connectivity(result) + + # Validate external API connectivity + await self._validate_api_connectivity(result) + + # Validate permissions + await self._validate_permissions(result) + + # Validate resource availability + await self._validate_resources(result) + + return result + + async def _validate_filesystem_access(self, result: ValidationResult) -> None: + """Validate filesystem access for media paths""" + all_paths = [] + + # Collect all media paths + all_paths.extend(self.config.tv_paths) + all_paths.extend(self.config.movie_paths) + + for path in all_paths: + try: + # Check if path exists and is accessible + if not path.exists(): + result.add_issue(ValidationIssue( + setting="media_paths", + severity=ValidationSeverity.ERROR, + message=f"Media path does not exist: {path}", + current_value=str(path) + )) + continue + + # Check if path is readable + test_file = None + try: + # Try to list directory contents + list(path.iterdir()) + except PermissionError: + result.add_issue(ValidationIssue( + setting="media_paths", + severity=ValidationSeverity.ERROR, + message=f"No read permission for media path: {path}", + current_value=str(path) + )) + continue + except OSError as e: + result.add_issue(ValidationIssue( + setting="media_paths", + severity=ValidationSeverity.WARNING, + message=f"Error accessing media path: {e}", + current_value=str(path) + )) + continue + + # Check write permissions (needed for NFO files) + try: + test_file = path / ".nfoguard_write_test" + test_file.write_text("test") + test_file.unlink() + except PermissionError: + result.add_issue(ValidationIssue( + setting="media_paths", + severity=ValidationSeverity.ERROR, + message=f"No write permission for media path: {path}", + current_value=str(path), + details={"required_for": "NFO file creation"} + )) + except OSError as e: + result.add_issue(ValidationIssue( + setting="media_paths", + severity=ValidationSeverity.WARNING, + message=f"Write test failed for media path: {e}", + current_value=str(path) + )) + finally: + # Cleanup test file if it exists + if test_file and test_file.exists(): + try: + test_file.unlink() + except: + pass + + except Exception as e: + result.add_issue(ValidationIssue( + setting="media_paths", + severity=ValidationSeverity.ERROR, + message=f"Unexpected error accessing path: {e}", + current_value=str(path) + )) + + # Validate database directory + db_path = Path(self.config.db_path) + db_dir = db_path.parent + + if not db_dir.exists(): + try: + db_dir.mkdir(parents=True, exist_ok=True) + except PermissionError: + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.ERROR, + message=f"Cannot create database directory: {db_dir}", + current_value=str(db_path) + )) + except OSError as e: + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.ERROR, + message=f"Error creating database directory: {e}", + current_value=str(db_path) + )) + + # Test database file access + if not db_path.exists(): + try: + # Try to create the database + with sqlite3.connect(str(db_path)) as conn: + conn.execute("SELECT 1") + except PermissionError: + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.ERROR, + message=f"Cannot create database file: {db_path}", + current_value=str(db_path) + )) + except sqlite3.Error as e: + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.ERROR, + message=f"Database error: {e}", + current_value=str(db_path) + )) + + async def _validate_database_connectivity(self, result: ValidationResult) -> None: + """Validate database connectivity""" + # Test Radarr database if configured + if hasattr(self.config, 'db_type') or 'RADARR_DB_TYPE' in os.environ: + import os + db_type = getattr(self.config, 'db_type', os.environ.get('RADARR_DB_TYPE', '')).lower() + + if db_type == 'postgresql': + await self._test_postgresql_connection(result) + elif db_type == 'sqlite': + await self._test_sqlite_connection(result) + + # Test local SQLite database + await self._test_local_database(result) + + async def _test_postgresql_connection(self, result: ValidationResult) -> None: + """Test PostgreSQL connection""" + import os + + try: + host = os.environ.get('RADARR_DB_HOST') + port = int(os.environ.get('RADARR_DB_PORT', 5432)) + database = os.environ.get('RADARR_DB_NAME') + user = os.environ.get('RADARR_DB_USER') + password = os.environ.get('RADARR_DB_PASSWORD', '') + + start_time = time.time() + + # Use asyncio to run blocking DB call + def test_connection(): + conn = psycopg2.connect( + host=host, port=port, database=database, + user=user, password=password, + connect_timeout=10 + ) + with conn: + with conn.cursor() as cur: + cur.execute("SELECT 1") + return conn + + conn = await asyncio.get_event_loop().run_in_executor(None, test_connection) + conn.close() + + response_time = (time.time() - start_time) * 1000 + + if response_time > 5000: # 5 seconds + result.add_issue(ValidationIssue( + setting="RADARR_DB", + severity=ValidationSeverity.WARNING, + message=f"Slow database connection ({response_time:.0f}ms)", + details={"response_time_ms": response_time} + )) + + except psycopg2.OperationalError as e: + result.add_issue(ValidationIssue( + setting="RADARR_DB", + severity=ValidationSeverity.ERROR, + message=f"Cannot connect to PostgreSQL database: {e}", + details={"error_type": "connection_failed"} + )) + except Exception as e: + result.add_issue(ValidationIssue( + setting="RADARR_DB", + severity=ValidationSeverity.ERROR, + message=f"Database connection error: {e}", + details={"error_type": "unexpected_error"} + )) + + async def _test_local_database(self, result: ValidationResult) -> None: + """Test local SQLite database""" + try: + db_path = Path(self.config.db_path) + + start_time = time.time() + + def test_db(): + with sqlite3.connect(str(db_path), timeout=10) as conn: + conn.execute("SELECT 1") + return True + + await asyncio.get_event_loop().run_in_executor(None, test_db) + + response_time = (time.time() - start_time) * 1000 + + if response_time > 1000: # 1 second is slow for SQLite + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.WARNING, + message=f"Slow local database access ({response_time:.0f}ms)", + current_value=str(db_path), + details={"response_time_ms": response_time} + )) + + except sqlite3.OperationalError as e: + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.ERROR, + message=f"Local database error: {e}", + current_value=str(self.config.db_path) + )) + + async def _validate_api_connectivity(self, result: ValidationResult) -> None: + """Validate external API connectivity""" + apis = [] + + if hasattr(self.config, 'radarr_url') and self.config.radarr_url: + apis.append(("Radarr", self.config.radarr_url)) + + if hasattr(self.config, 'sonarr_url') and self.config.sonarr_url: + apis.append(("Sonarr", self.config.sonarr_url)) + + for api_name, base_url in apis: + await self._test_api_connectivity(result, api_name, base_url) + + async def _test_api_connectivity(self, result: ValidationResult, api_name: str, base_url: str) -> None: + """Test connectivity to a specific API""" + try: + timeout = aiohttp.ClientTimeout(total=10) + + async with aiohttp.ClientSession(timeout=timeout) as session: + start_time = time.time() + + # Test basic connectivity + test_url = f"{base_url.rstrip('/')}/api/v1/health" + + async with session.get(test_url) as response: + response_time = (time.time() - start_time) * 1000 + + if response.status == 200: + if response_time > 5000: # 5 seconds + result.add_issue(ValidationIssue( + setting=f"{api_name.upper()}_URL", + severity=ValidationSeverity.WARNING, + message=f"Slow {api_name} API response ({response_time:.0f}ms)", + current_value=base_url, + details={"response_time_ms": response_time} + )) + else: + result.add_issue(ValidationIssue( + setting=f"{api_name.upper()}_URL", + severity=ValidationSeverity.WARNING, + message=f"{api_name} API returned HTTP {response.status}", + current_value=base_url, + details={"status_code": response.status} + )) + + except asyncio.TimeoutError: + result.add_issue(ValidationIssue( + setting=f"{api_name.upper()}_URL", + severity=ValidationSeverity.WARNING, + message=f"{api_name} API connection timeout", + current_value=base_url, + details={"error_type": "timeout"} + )) + except Exception as e: + result.add_issue(ValidationIssue( + setting=f"{api_name.upper()}_URL", + severity=ValidationSeverity.WARNING, + message=f"{api_name} API connection error: {e}", + current_value=base_url, + details={"error_type": "connection_error"} + )) + + async def _validate_permissions(self, result: ValidationResult) -> None: + """Validate file system permissions""" + # This is partially covered in filesystem validation + # Additional permission checks can be added here + pass + + async def _validate_resources(self, result: ValidationResult) -> None: + """Validate system resources""" + import shutil + + try: + # Check disk space for database directory + db_path = Path(self.config.db_path) + db_dir = db_path.parent if not db_path.is_dir() else db_path + + if db_dir.exists(): + free_space = shutil.disk_usage(str(db_dir)).free + free_space_mb = free_space / (1024 * 1024) + + if free_space_mb < 100: # Less than 100MB + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.ERROR, + message=f"Low disk space in database directory ({free_space_mb:.1f}MB free)", + current_value=str(db_path), + details={"free_space_mb": free_space_mb} + )) + elif free_space_mb < 1000: # Less than 1GB + result.add_issue(ValidationIssue( + setting="DB_PATH", + severity=ValidationSeverity.WARNING, + message=f"Low disk space in database directory ({free_space_mb:.1f}MB free)", + current_value=str(db_path), + details={"free_space_mb": free_space_mb} + )) + + except Exception as e: + result.add_issue(ValidationIssue( + setting="system_resources", + severity=ValidationSeverity.WARNING, + message=f"Could not check disk space: {e}", + details={"error_type": "resource_check_failed"} + )) + + async def get_system_health(self) -> Dict[str, HealthCheckResult]: + """Get comprehensive system health status""" + health_checks = {} + + # Database health + health_checks["database"] = await self._check_database_health() + + # Filesystem health + health_checks["filesystem"] = await self._check_filesystem_health() + + # API health + health_checks["external_apis"] = await self._check_api_health() + + return health_checks + + async def _check_database_health(self) -> HealthCheckResult: + """Check database health""" + try: + start_time = time.time() + + def test_db(): + with sqlite3.connect(str(self.config.db_path), timeout=5) as conn: + conn.execute("SELECT COUNT(*) FROM sqlite_master") + return True + + await asyncio.get_event_loop().run_in_executor(None, test_db) + + response_time = (time.time() - start_time) * 1000 + + return HealthCheckResult( + component="database", + is_healthy=True, + response_time_ms=response_time, + message="Database accessible", + details={"db_path": str(self.config.db_path)} + ) + + except Exception as e: + return HealthCheckResult( + component="database", + is_healthy=False, + message=f"Database error: {e}", + details={"db_path": str(self.config.db_path), "error": str(e)} + ) + + async def _check_filesystem_health(self) -> HealthCheckResult: + """Check filesystem health""" + try: + accessible_paths = 0 + total_paths = len(self.config.tv_paths) + len(self.config.movie_paths) + + for path in list(self.config.tv_paths) + list(self.config.movie_paths): + if path.exists() and path.is_dir(): + try: + # Quick access test + next(path.iterdir(), None) + accessible_paths += 1 + except: + pass + + is_healthy = accessible_paths == total_paths + health_percentage = (accessible_paths / total_paths * 100) if total_paths > 0 else 0 + + return HealthCheckResult( + component="filesystem", + is_healthy=is_healthy, + message=f"{accessible_paths}/{total_paths} media paths accessible ({health_percentage:.1f}%)", + details={ + "accessible_paths": accessible_paths, + "total_paths": total_paths, + "health_percentage": health_percentage + } + ) + + except Exception as e: + return HealthCheckResult( + component="filesystem", + is_healthy=False, + message=f"Filesystem check error: {e}", + details={"error": str(e)} + ) + + async def _check_api_health(self) -> HealthCheckResult: + """Check external API health""" + apis_tested = 0 + apis_healthy = 0 + api_details = {} + + try: + # Test configured APIs + if hasattr(self.config, 'radarr_url') and self.config.radarr_url: + apis_tested += 1 + healthy = await self._test_single_api("Radarr", self.config.radarr_url) + if healthy: + apis_healthy += 1 + api_details["radarr"] = {"healthy": healthy} + + if hasattr(self.config, 'sonarr_url') and self.config.sonarr_url: + apis_tested += 1 + healthy = await self._test_single_api("Sonarr", self.config.sonarr_url) + if healthy: + apis_healthy += 1 + api_details["sonarr"] = {"healthy": healthy} + + if apis_tested == 0: + return HealthCheckResult( + component="external_apis", + is_healthy=True, + message="No external APIs configured", + details={"apis_configured": 0} + ) + + is_healthy = apis_healthy == apis_tested + health_percentage = (apis_healthy / apis_tested * 100) if apis_tested > 0 else 0 + + return HealthCheckResult( + component="external_apis", + is_healthy=is_healthy, + message=f"{apis_healthy}/{apis_tested} APIs healthy ({health_percentage:.1f}%)", + details={ + "healthy_apis": apis_healthy, + "total_apis": apis_tested, + "health_percentage": health_percentage, + "api_status": api_details + } + ) + + except Exception as e: + return HealthCheckResult( + component="external_apis", + is_healthy=False, + message=f"API health check error: {e}", + details={"error": str(e)} + ) + + async def _test_single_api(self, api_name: str, base_url: str) -> bool: + """Test a single API for health""" + try: + timeout = aiohttp.ClientTimeout(total=5) + + async with aiohttp.ClientSession(timeout=timeout) as session: + test_url = f"{base_url.rstrip('/')}/api/v1/health" + + async with session.get(test_url) as response: + return response.status == 200 + + except: + return False + + +# Import needed for database validation +import os \ No newline at end of file diff --git a/config/settings.py b/config/settings.py index 4cf4461..34b6eb5 100644 --- a/config/settings.py +++ b/config/settings.py @@ -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() \ No newline at end of file +# 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) \ No newline at end of file diff --git a/config/validation_cli.py b/config/validation_cli.py new file mode 100755 index 0000000..3d54d59 --- /dev/null +++ b/config/validation_cli.py @@ -0,0 +1,293 @@ +#!/usr/bin/env python3 +""" +Configuration Validation CLI for NFOGuard +Provides command-line validation and reporting of configuration issues +""" +import sys +import json +import argparse +from pathlib import Path +from typing import Dict, Any, List +from datetime import datetime + +from config.validator import validate_configuration, ValidationSeverity +from config.runtime_validator import RuntimeValidator +from config.settings import NFOGuardConfig + + +class ValidationReporter: + """Formats and displays validation results""" + + def __init__(self, verbose: bool = False, json_output: bool = False): + self.verbose = verbose + self.json_output = json_output + + # Color codes for terminal output + self.colors = { + 'error': '\033[91m', # Red + 'warning': '\033[93m', # Yellow + 'info': '\033[94m', # Blue + 'success': '\033[92m', # Green + 'reset': '\033[0m', # Reset + 'bold': '\033[1m' # Bold + } + + def report_validation_results(self, result, runtime_result=None) -> int: + """ + Report validation results + + Returns: + Exit code (0 for success, 1 for warnings, 2 for errors) + """ + if self.json_output: + return self._report_json(result, runtime_result) + else: + return self._report_human_readable(result, runtime_result) + + def _report_json(self, result, runtime_result=None) -> int: + """Report results in JSON format""" + output = { + "timestamp": datetime.now().isoformat(), + "validation": result.to_dict() + } + + if runtime_result: + output["runtime_validation"] = runtime_result.to_dict() + + print(json.dumps(output, indent=2)) + + if result.errors_count > 0: + return 2 + elif result.warnings_count > 0: + return 1 + return 0 + + def _report_human_readable(self, result, runtime_result=None) -> int: + """Report results in human-readable format""" + print(f"{self.colors['bold']}NFOGuard Configuration Validation Report{self.colors['reset']}") + print("=" * 50) + + # Overall status + if result.is_valid: + status_color = self.colors['success'] + status_text = "✓ VALID" + else: + status_color = self.colors['error'] + status_text = "✗ INVALID" + + print(f"Status: {status_color}{status_text}{self.colors['reset']}") + print(f"Errors: {result.errors_count}") + print(f"Warnings: {result.warnings_count}") + print(f"Total Issues: {len(result.issues)}") + print() + + # Report issues by severity + if result.issues: + self._report_issues_by_severity(result.issues) + + # Report runtime validation if available + if runtime_result: + print(f"\n{self.colors['bold']}Runtime Validation{self.colors['reset']}") + print("-" * 20) + if runtime_result.issues: + self._report_issues_by_severity(runtime_result.issues, "Runtime") + else: + print(f"{self.colors['success']}✓ All runtime checks passed{self.colors['reset']}") + + # Summary and recommendations + self._report_summary_and_recommendations(result) + + # Return appropriate exit code + if result.errors_count > 0 or (runtime_result and runtime_result.errors_count > 0): + return 2 + elif result.warnings_count > 0 or (runtime_result and runtime_result.warnings_count > 0): + return 1 + return 0 + + def _report_issues_by_severity(self, issues: List, context: str = "Configuration") -> None: + """Report issues grouped by severity""" + errors = [issue for issue in issues if issue.severity == ValidationSeverity.ERROR] + warnings = [issue for issue in issues if issue.severity == ValidationSeverity.WARNING] + info_issues = [issue for issue in issues if issue.severity == ValidationSeverity.INFO] + + if errors: + print(f"{self.colors['error']}{self.colors['bold']}ERRORS ({len(errors)}):{self.colors['reset']}") + for issue in errors: + self._format_issue(issue) + print() + + if warnings: + print(f"{self.colors['warning']}{self.colors['bold']}WARNINGS ({len(warnings)}):{self.colors['reset']}") + for issue in warnings: + self._format_issue(issue) + print() + + if info_issues and self.verbose: + print(f"{self.colors['info']}{self.colors['bold']}INFO ({len(info_issues)}):{self.colors['reset']}") + for issue in info_issues: + self._format_issue(issue) + print() + + def _format_issue(self, issue) -> None: + """Format a single validation issue""" + severity_colors = { + ValidationSeverity.ERROR: self.colors['error'], + ValidationSeverity.WARNING: self.colors['warning'], + ValidationSeverity.INFO: self.colors['info'] + } + + color = severity_colors.get(issue.severity, '') + + print(f" {color}• {issue.setting}:{self.colors['reset']} {issue.message}") + + if issue.current_value is not None: + print(f" Current: {issue.current_value}") + + if issue.suggested_value is not None: + print(f" Suggested: {issue.suggested_value}") + + if self.verbose and issue.details: + print(f" Details: {issue.details}") + + def _report_summary_and_recommendations(self, result) -> None: + """Report summary and general recommendations""" + print(f"\n{self.colors['bold']}Summary{self.colors['reset']}") + print("-" * 10) + + if result.is_valid: + print(f"{self.colors['success']}✓ Your configuration is valid and ready to use!{self.colors['reset']}") + else: + print(f"{self.colors['error']}✗ Your configuration has issues that need to be resolved.{self.colors['reset']}") + + # Provide specific recommendations based on issue types + recommendations = self._generate_recommendations(result) + if recommendations: + print(f"\n{self.colors['bold']}Recommendations{self.colors['reset']}") + print("-" * 15) + for rec in recommendations: + print(f" {self.colors['info']}• {rec}{self.colors['reset']}") + + def _generate_recommendations(self, result) -> List[str]: + """Generate recommendations based on validation results""" + recommendations = [] + + # Check for common patterns + path_issues = [issue for issue in result.issues if 'path' in issue.setting.lower()] + if path_issues: + recommendations.append("Verify all file paths are correct for your environment") + recommendations.append("Ensure paths are absolute when using Docker") + + db_issues = [issue for issue in result.issues if 'db' in issue.setting.lower()] + if db_issues: + recommendations.append("Check database connection settings and credentials") + + url_issues = [issue for issue in result.issues if 'url' in issue.setting.lower()] + if url_issues: + recommendations.append("Verify API URLs are reachable and include the correct port") + + perf_issues = [issue for issue in result.issues if any(perf in issue.setting.lower() + for perf in ['concurrent', 'delay', 'timeout'])] + if perf_issues: + recommendations.append("Consider adjusting performance settings based on your system resources") + + # General recommendations + if result.errors_count > 0: + recommendations.append("Fix all ERROR-level issues before starting NFOGuard") + + if result.warnings_count > 0: + recommendations.append("Review WARNING-level issues to optimize performance and reliability") + + return recommendations + + +async def run_validation(args) -> int: + """Run configuration validation""" + reporter = ValidationReporter(verbose=args.verbose, json_output=args.json) + + try: + # Run static validation + print("Running configuration validation..." if not args.json else "", file=sys.stderr) + result = validate_configuration() + + runtime_result = None + + # Run runtime validation if requested + if args.runtime: + print("Running runtime validation..." if not args.json else "", file=sys.stderr) + try: + config = NFOGuardConfig() + runtime_validator = RuntimeValidator(config) + runtime_result = await runtime_validator.validate_runtime_config() + except Exception as e: + if not args.json: + print(f"Runtime validation failed: {e}", file=sys.stderr) + # Continue with static validation results + + # Report results + return reporter.report_validation_results(result, runtime_result) + + except Exception as e: + if args.json: + error_output = { + "timestamp": datetime.now().isoformat(), + "error": { + "message": str(e), + "type": type(e).__name__ + } + } + print(json.dumps(error_output, indent=2)) + else: + print(f"Validation failed: {e}", file=sys.stderr) + return 2 + + +def main(): + """Main CLI entry point""" + parser = argparse.ArgumentParser( + description="Validate NFOGuard configuration", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + %(prog)s # Basic validation + %(prog)s --runtime # Include runtime checks + %(prog)s --verbose # Show detailed information + %(prog)s --json # Output JSON format + %(prog)s --runtime --json # Runtime validation with JSON output + """ + ) + + parser.add_argument( + "--runtime", + action="store_true", + help="Perform runtime validation (tests actual connectivity and permissions)" + ) + + parser.add_argument( + "--verbose", "-v", + action="store_true", + help="Show verbose output including info-level messages" + ) + + parser.add_argument( + "--json", + action="store_true", + help="Output results in JSON format" + ) + + args = parser.parse_args() + + # Run validation + import asyncio + try: + exit_code = asyncio.run(run_validation(args)) + sys.exit(exit_code) + except KeyboardInterrupt: + print("Validation interrupted", file=sys.stderr) + sys.exit(130) + except Exception as e: + print(f"Unexpected error: {e}", file=sys.stderr) + sys.exit(2) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/config/validator.py b/config/validator.py new file mode 100644 index 0000000..6dbb1d3 --- /dev/null +++ b/config/validator.py @@ -0,0 +1,498 @@ +""" +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 + ] + } \ No newline at end of file