#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Configuration Management System Provides centralized configuration management with environment-based overrides """ import os import json import logging from typing import Any, Dict, Optional from pathlib import Path class ConfigManager: """Manages configuration for automation scripts with environment support.""" def __init__(self, config_file: Optional[str] = None): """Initialize configuration manager. Args: config_file: Path to configuration JSON file. If None, uses default location. """ if config_file is None: # Default to config file in automation_scripts/config directory script_dir = Path(__file__).parent.parent config_file = script_dir / 'config' / 'automation_config.json' self.config_file = Path(config_file) self._config: Dict[str, Any] = {} self._load_config() def _load_config(self): """Load configuration from JSON file.""" try: if self.config_file.exists(): with open(self.config_file, 'r', encoding='utf-8') as f: self._config = json.load(f) else: logging.warning(f"Config file not found: {self.config_file}, using defaults") self._config = self._get_default_config() except Exception as e: logging.error(f"Error loading config file: {e}, using defaults") self._config = self._get_default_config() # Apply environment variable overrides self._apply_env_overrides() def _get_default_config(self) -> Dict[str, Any]: """Get default configuration.""" return { "browser": { "headless": True, "window_size": "1920,1080", "incognito": True, "timeout": 30, "page_load_timeout": 30, "implicit_wait": 10, "explicit_wait": 15, "retry_count": 3, "retry_delay": 2 }, "downloads": { "timeout": 30, "check_interval": 0.5, "max_retries": 10, "retry_delay": 0.5 }, "session": { "check_interval": 1, "login_timeout": 600, "session_refresh_threshold": 300, "max_session_age": 3600 }, "errors": { "max_retries": 3, "base_delay": 1, "max_delay": 10, "exponential_base": 2, "circuit_breaker_threshold": 5, "circuit_breaker_timeout": 60 }, "performance": { "parallel_workers": 2, "batch_size": 10, "cache_enabled": True, "cache_ttl": 3600, "adaptive_timeout": True, "min_timeout": 5, "max_timeout": 60 }, "logging": { "level": "INFO", "format": "%(asctime)s [%(levelname)s] %(message)s", "file_encoding": "utf-8", "console_output": True, "file_output": False }, "services": {} } def _apply_env_overrides(self): """Apply environment variable overrides to configuration.""" env = os.environ # Browser settings if 'AUTOMATION_HEADLESS' in env: self.set('browser.headless', env['AUTOMATION_HEADLESS'].lower() == 'true') if 'AUTOMATION_WINDOW_SIZE' in env: self.set('browser.window_size', env['AUTOMATION_WINDOW_SIZE']) if 'AUTOMATION_TIMEOUT' in env: try: self.set('browser.timeout', int(env['AUTOMATION_TIMEOUT'])) except ValueError: pass # Performance settings if 'AUTOMATION_PARALLEL_WORKERS' in env: try: self.set('performance.parallel_workers', int(env['AUTOMATION_PARALLEL_WORKERS'])) except ValueError: pass # Logging settings if 'AUTOMATION_LOG_LEVEL' in env: self.set('logging.level', env['AUTOMATION_LOG_LEVEL'].upper()) def get(self, key: str, default: Any = None) -> Any: """Get configuration value using dot notation. Args: key: Configuration key in dot notation (e.g., 'browser.headless') default: Default value if key not found Returns: Configuration value or default """ keys = key.split('.') value = self._config for k in keys: if isinstance(value, dict) and k in value: value = value[k] else: return default return value def set(self, key: str, value: Any): """Set configuration value using dot notation. Args: key: Configuration key in dot notation (e.g., 'browser.headless') value: Value to set """ keys = key.split('.') config = self._config for k in keys[:-1]: if k not in config: config[k] = {} config = config[k] config[keys[-1]] = value def get_service_config(self, service_name: str) -> Dict[str, Any]: """Get service-specific configuration. Args: service_name: Name of the service Returns: Dictionary of service configuration """ return self.get(f'services.{service_name}', {}) def get_browser_config(self) -> Dict[str, Any]: """Get browser configuration. Returns: Dictionary of browser configuration """ return self.get('browser', {}) def get_download_config(self) -> Dict[str, Any]: """Get download configuration. Returns: Dictionary of download configuration """ return self.get('downloads', {}) def get_session_config(self) -> Dict[str, Any]: """Get session configuration. Returns: Dictionary of session configuration """ return self.get('session', {}) def get_error_config(self) -> Dict[str, Any]: """Get error handling configuration. Returns: Dictionary of error handling configuration """ return self.get('errors', {}) def get_performance_config(self) -> Dict[str, Any]: """Get performance configuration. Returns: Dictionary of performance configuration """ return self.get('performance', {}) def get_logging_config(self) -> Dict[str, Any]: """Get logging configuration. Returns: Dictionary of logging configuration """ return self.get('logging', {}) def reload(self): """Reload configuration from file.""" self._load_config() def validate(self) -> tuple[bool, list[str]]: """Validate configuration. Returns: Tuple of (is_valid, list_of_errors) """ errors = [] # Validate browser config browser_config = self.get_browser_config() if not isinstance(browser_config.get('timeout'), (int, float)) or browser_config.get('timeout', 0) <= 0: errors.append("browser.timeout must be a positive number") # Validate performance config perf_config = self.get_performance_config() if not isinstance(perf_config.get('parallel_workers'), int) or perf_config.get('parallel_workers', 0) <= 0: errors.append("performance.parallel_workers must be a positive integer") return len(errors) == 0, errors # Global configuration instance _config_instance: Optional[ConfigManager] = None def get_config(config_file: Optional[str] = None) -> ConfigManager: """Get global configuration instance. Args: config_file: Optional path to config file (only used on first call) Returns: ConfigManager instance """ global _config_instance if _config_instance is None: _config_instance = ConfigManager(config_file) return _config_instance def reload_config(): """Reload global configuration.""" global _config_instance if _config_instance: _config_instance.reload()