"""Configuration management with validation, profiles, and import/export.""" import json import logging from pathlib import Path from typing import Any, Dict, List, Optional from webdrop_bridge.config import Config, ConfigurationError logger = logging.getLogger(__name__) class ConfigValidator: """Validates configuration values against schema. Provides detailed error messages for invalid configurations. """ # Schema definition for configuration SCHEMA = { "app_name": {"type": str, "min_length": 1, "max_length": 100}, "app_version": {"type": str, "pattern": r"^\d+\.\d+\.\d+$"}, "log_level": {"type": str, "allowed": ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]}, "log_file": {"type": (str, type(None)), "optional": True}, "allowed_roots": {"type": list, "item_type": (str, Path), "min_items": 0}, "allowed_urls": {"type": list, "item_type": str, "min_items": 0}, "webapp_url": {"type": str, "min_length": 1}, "window_width": {"type": int, "min_value": 400, "max_value": 5000}, "window_height": {"type": int, "min_value": 300, "max_value": 5000}, "enable_logging": {"type": bool}, } @staticmethod def validate(config_dict: Dict[str, Any]) -> List[str]: """Validate configuration dictionary. Args: config_dict: Configuration dictionary to validate Returns: List of validation error messages (empty if valid) """ errors = [] for field, rules in ConfigValidator.SCHEMA.items(): if field not in config_dict: if not rules.get("optional", False): errors.append(f"Missing required field: {field}") continue value = config_dict[field] # Check type expected_type = rules.get("type") if expected_type and not isinstance(value, expected_type): errors.append(f"{field}: expected {expected_type.__name__}, got {type(value).__name__}") continue # Check allowed values if "allowed" in rules and value not in rules["allowed"]: errors.append(f"{field}: must be one of {rules['allowed']}, got {value}") # Check string length if isinstance(value, str): if "min_length" in rules and len(value) < rules["min_length"]: errors.append(f"{field}: minimum length is {rules['min_length']}") if "max_length" in rules and len(value) > rules["max_length"]: errors.append(f"{field}: maximum length is {rules['max_length']}") # Check numeric range if isinstance(value, int): if "min_value" in rules and value < rules["min_value"]: errors.append(f"{field}: minimum value is {rules['min_value']}") if "max_value" in rules and value > rules["max_value"]: errors.append(f"{field}: maximum value is {rules['max_value']}") # Check list items if isinstance(value, list): if "min_items" in rules and len(value) < rules["min_items"]: errors.append(f"{field}: minimum {rules['min_items']} items required") return errors @staticmethod def validate_or_raise(config_dict: Dict[str, Any]) -> None: """Validate configuration and raise error if invalid. Args: config_dict: Configuration dictionary to validate Raises: ConfigurationError: If configuration is invalid """ errors = ConfigValidator.validate(config_dict) if errors: raise ConfigurationError(f"Configuration validation failed:\n" + "\n".join(errors)) class ConfigProfile: """Manages named configuration profiles. Profiles are stored in ~/.webdrop-bridge/profiles/ directory as JSON files. """ PROFILES_DIR = Path.home() / ".webdrop-bridge" / "profiles" def __init__(self): """Initialize profile manager.""" self.PROFILES_DIR.mkdir(parents=True, exist_ok=True) def save_profile(self, profile_name: str, config: Config) -> Path: """Save configuration as a named profile. Args: profile_name: Name of the profile (e.g., "work", "personal") config: Config object to save Returns: Path to the saved profile file Raises: ConfigurationError: If profile name is invalid """ if not profile_name or "/" in profile_name or "\\" in profile_name: raise ConfigurationError(f"Invalid profile name: {profile_name}") profile_path = self.PROFILES_DIR / f"{profile_name}.json" config_data = { "app_name": config.app_name, "app_version": config.app_version, "log_level": config.log_level, "log_file": str(config.log_file) if config.log_file else None, "allowed_roots": [str(p) for p in config.allowed_roots], "allowed_urls": config.allowed_urls, "webapp_url": config.webapp_url, "window_width": config.window_width, "window_height": config.window_height, "enable_logging": config.enable_logging, } try: profile_path.write_text(json.dumps(config_data, indent=2)) logger.info(f"Profile saved: {profile_name}") return profile_path except (OSError, IOError) as e: raise ConfigurationError(f"Failed to save profile {profile_name}: {e}") def load_profile(self, profile_name: str) -> Dict[str, Any]: """Load configuration from a named profile. Args: profile_name: Name of the profile to load Returns: Configuration dictionary Raises: ConfigurationError: If profile not found or invalid """ profile_path = self.PROFILES_DIR / f"{profile_name}.json" if not profile_path.exists(): raise ConfigurationError(f"Profile not found: {profile_name}") try: config_data = json.loads(profile_path.read_text()) # Validate before returning ConfigValidator.validate_or_raise(config_data) return config_data except json.JSONDecodeError as e: raise ConfigurationError(f"Invalid JSON in profile {profile_name}: {e}") def list_profiles(self) -> List[str]: """List all available profiles. Returns: List of profile names (without .json extension) """ if not self.PROFILES_DIR.exists(): return [] return sorted([p.stem for p in self.PROFILES_DIR.glob("*.json")]) def delete_profile(self, profile_name: str) -> None: """Delete a profile. Args: profile_name: Name of the profile to delete Raises: ConfigurationError: If profile not found """ profile_path = self.PROFILES_DIR / f"{profile_name}.json" if not profile_path.exists(): raise ConfigurationError(f"Profile not found: {profile_name}") try: profile_path.unlink() logger.info(f"Profile deleted: {profile_name}") except OSError as e: raise ConfigurationError(f"Failed to delete profile {profile_name}: {e}") class ConfigExporter: """Handle configuration import and export operations.""" @staticmethod def export_to_json(config: Config, output_path: Path) -> None: """Export configuration to JSON file. Args: config: Config object to export output_path: Path to write JSON file Raises: ConfigurationError: If export fails """ config_data = { "app_name": config.app_name, "app_version": config.app_version, "log_level": config.log_level, "log_file": str(config.log_file) if config.log_file else None, "allowed_roots": [str(p) for p in config.allowed_roots], "allowed_urls": config.allowed_urls, "webapp_url": config.webapp_url, "window_width": config.window_width, "window_height": config.window_height, "enable_logging": config.enable_logging, } try: output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(config_data, indent=2)) logger.info(f"Configuration exported to: {output_path}") except (OSError, IOError) as e: raise ConfigurationError(f"Failed to export configuration: {e}") @staticmethod def import_from_json(input_path: Path) -> Dict[str, Any]: """Import configuration from JSON file. Args: input_path: Path to JSON file to import Returns: Configuration dictionary Raises: ConfigurationError: If import fails or validation fails """ if not input_path.exists(): raise ConfigurationError(f"File not found: {input_path}") try: config_data = json.loads(input_path.read_text()) # Validate before returning ConfigValidator.validate_or_raise(config_data) logger.info(f"Configuration imported from: {input_path}") return config_data except json.JSONDecodeError as e: raise ConfigurationError(f"Invalid JSON file: {e}")