""" Configuration Management System for SDG Dashboard Supports JSON/YAML configuration files, environment variables, and runtime updates. Author: Kilo Code Version: 2025.1 """ import json import os import logging from typing import Any, Dict, Optional, Union from pathlib import Path from dataclasses import dataclass, field from copy import deepcopy # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Try to import yaml support try: import yaml YAML_AVAILABLE = True except ImportError: YAML_AVAILABLE = False logger.info("PyYAML not installed. YAML config files will not be supported.") @dataclass class ConfigValidationError(Exception): """Exception raised for configuration validation errors.""" errors: list = field(default_factory=list) def __str__(self): return f"Configuration validation failed: {', '.join(self.errors)}" class ConfigManager: """ Singleton Configuration Manager for SDG Dashboard. Features: - Load configuration from JSON/YAML files - Environment variable support - Default configuration values - Configuration validation - Environment-based configuration (dev, test, prod) - Runtime configuration updates """ _instance: Optional['ConfigManager'] = None _initialized: bool = False # Default configuration DEFAULT_CONFIG = { "app": { "name": "SDG Dashboard", "version": "2025.1", "debug": False, "log_level": "INFO", "secret_key": None }, "data_sources": { "primary_data_path": "data/SDR2025-data.xlsx", "fallback_data_path": "data/sdg_index_2000-2022.csv", "sample_data_enabled": True, "cache_ttl_seconds": 3600, "api_endpoints": { "sdg_api": "https://api.sdgindex.org/v1", "un_stats_api": "https://unstats.un.org/sdgapi" } }, "ai_engine": { "enabled": True, "base_url": None, "api_key": None, "default_model": "azure-gpt-4.1", "temperature": 0.6, "max_tokens": 4096, "cache_enabled": True, "cache_ttl_hours": 24, "retry_attempts": 3, "retry_delay_seconds": 2 }, "visualization": { "default_theme": "plotly_white", "color_scheme": "sdg_official", "chart_height": 500, "map_projection": "equirectangular", "sdg_colors": { "1": "#E5243B", "2": "#DDA63A", "3": "#4C9F38", "4": "#C5192D", "5": "#FF3A21", "6": "#26BDE2", "7": "#FCC30B", "8": "#A21942", "9": "#FD6925", "10": "#DD1367", "11": "#FD9D24", "12": "#BF8B2E", "13": "#3F7E44", "14": "#0A97D9", "15": "#56C02B", "16": "#00689D", "17": "#19486A" }, "traffic_light_thresholds": { "excellent": 80, "good": 65, "needs_improvement": 50 } }, "export": { "default_format": "pdf", "pdf_quality": "high", "pptx_template": None, "output_directory": "exports", "include_metadata": True, "max_content_per_slide": 500 }, "ui": { "page_title": "全球 SDG 互動儀表板 & AI 報告生成器", "page_icon": "🌍", "layout": "wide", "initial_sidebar_state": "expanded", "default_language": "繁體中文", "default_year_range": [2015, 2025], "max_countries_comparison": 5 }, "security": { "require_api_key": False, "allowed_origins": ["*"], "rate_limit_requests": 100, "rate_limit_period_seconds": 60 } } # Environment variable mappings ENV_MAPPINGS = { "SDG_DEBUG": ("app", "debug", lambda x: x.lower() == "true"), "SDG_LOG_LEVEL": ("app", "log_level", str), "SDG_SECRET_KEY": ("app", "secret_key", str), "LITELLM_BASE_URL": ("ai_engine", "base_url", str), "LITELLM_API_KEY": ("ai_engine", "api_key", str), "SDG_AI_MODEL": ("ai_engine", "default_model", str), "SDG_DATA_PATH": ("data_sources", "primary_data_path", str), "SDG_CACHE_TTL": ("data_sources", "cache_ttl_seconds", int), "SDG_EXPORT_DIR": ("export", "output_directory", str), "SDG_ENVIRONMENT": ("app", "environment", str) } def __new__(cls, *args, **kwargs): """Implement singleton pattern.""" if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self, config_dir: Optional[str] = None, environment: Optional[str] = None): """ Initialize the configuration manager. Args: config_dir: Directory containing configuration files environment: Environment name (development, production, test) """ if self._initialized: return # Load environment variables first (before any configuration loading) self._load_dotenv_if_available() self._config: Dict[str, Any] = deepcopy(self.DEFAULT_CONFIG) self._config_dir = config_dir or os.path.join(os.path.dirname(os.path.dirname(__file__)), "config") self._environment = environment or os.environ.get("SDG_ENVIRONMENT", "development") self._loaded_files: list = [] # Load configuration self._load_configuration() self._initialized = True logger.info(f"ConfigManager initialized for environment: {self._environment}") def _load_configuration(self): """Load configuration from files and environment variables.""" # 1. Load default configuration file self._load_config_file("default.json") # 2. Load environment-specific configuration env_config_file = f"{self._environment}.json" self._load_config_file(env_config_file) # 3. Load environment variables (highest priority) self._load_environment_variables() # 4. Validate configuration self._validate_configuration() def _load_config_file(self, filename: str): """Load a configuration file and merge with current config.""" config_path = os.path.join(self._config_dir, filename) if not os.path.exists(config_path): logger.debug(f"Config file not found: {config_path}") return try: with open(config_path, 'r', encoding='utf-8') as f: if filename.endswith('.yaml') or filename.endswith('.yml'): if not YAML_AVAILABLE: logger.warning(f"Cannot load YAML file {filename}: PyYAML not installed") return file_config = yaml.safe_load(f) else: file_config = json.load(f) if file_config: self._merge_config(self._config, file_config) self._loaded_files.append(filename) logger.info(f"Loaded configuration from: {filename}") except Exception as e: logger.error(f"Error loading config file {filename}: {e}") def _merge_config(self, base: Dict, override: Dict): """Deep merge override config into base config.""" for key, value in override.items(): if key in base and isinstance(base[key], dict) and isinstance(value, dict): self._merge_config(base[key], value) else: base[key] = value def _load_dotenv_if_available(self): """Load .env file if python-dotenv is available.""" try: from dotenv import load_dotenv # Try to load .env file from current directory and parent directories load_dotenv() logger.debug("Environment variables loaded from .env file") except ImportError: logger.debug("python-dotenv not available, skipping .env file loading") except Exception as e: logger.warning(f"Failed to load .env file: {e}") def _load_environment_variables(self): """Load configuration from environment variables.""" for env_var, (section, key, converter) in self.ENV_MAPPINGS.items(): value = os.environ.get(env_var) if value is not None: try: converted_value = converter(value) if section not in self._config: self._config[section] = {} self._config[section][key] = converted_value logger.debug(f"Loaded {env_var} into config[{section}][{key}]") except Exception as e: logger.warning(f"Failed to convert {env_var}={value}: {e}") def _validate_configuration(self): """Validate the loaded configuration.""" errors = [] # Validate required sections exist required_sections = ["app", "data_sources", "ai_engine", "visualization", "export", "ui"] for section in required_sections: if section not in self._config: errors.append(f"Missing required section: {section}") # Validate data paths primary_path = self.get("data_sources.primary_data_path") fallback_path = self.get("data_sources.fallback_data_path") if not primary_path and not fallback_path: errors.append("At least one data source path must be configured") # Validate AI engine settings if self.get("ai_engine.enabled"): if not self.get("ai_engine.base_url") and not self.get("ai_engine.api_key"): logger.info("AI engine is enabled but no credentials configured - will run in mock mode") logger.info("To enable AI features, set LITELLM_BASE_URL and LITELLM_API_KEY environment variables") # Validate thresholds thresholds = self.get("visualization.traffic_light_thresholds", {}) if thresholds: if thresholds.get("excellent", 0) < thresholds.get("good", 0): errors.append("Excellent threshold must be greater than good threshold") if thresholds.get("good", 0) < thresholds.get("needs_improvement", 0): errors.append("Good threshold must be greater than needs_improvement threshold") if errors: raise ConfigValidationError(errors=errors) logger.info("Configuration validation passed") def get(self, key: str, default: Any = None) -> Any: """ Get a configuration value using dot notation. Args: key: Configuration key (e.g., "app.debug", "ai_engine.base_url") default: Default value if key not found Returns: Configuration value or default """ keys = key.split(".") value = self._config try: for k in keys: value = value[k] return value except (KeyError, TypeError): return default def set(self, key: str, value: Any, persist: bool = False): """ Set a configuration value at runtime. Args: key: Configuration key (dot notation) value: Value to set persist: Whether to persist to config file (not implemented) """ 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 logger.debug(f"Set config[{key}] = {value}") if persist: logger.warning("Config persistence is not implemented yet") def get_section(self, section: str) -> Dict[str, Any]: """ Get an entire configuration section. Args: section: Section name (e.g., "app", "ai_engine") Returns: Configuration section dictionary """ return deepcopy(self._config.get(section, {})) def get_all(self) -> Dict[str, Any]: """Get the entire configuration dictionary.""" return deepcopy(self._config) @property def environment(self) -> str: """Get the current environment name.""" return self._environment @property def is_debug(self) -> bool: """Check if debug mode is enabled.""" return self.get("app.debug", False) @property def loaded_files(self) -> list: """Get list of loaded configuration files.""" return self._loaded_files.copy() def reload(self): """Reload configuration from files and environment.""" self._config = deepcopy(self.DEFAULT_CONFIG) self._loaded_files = [] self._load_configuration() logger.info("Configuration reloaded") @classmethod def reset_instance(cls): """Reset the singleton instance (useful for testing).""" cls._instance = None cls._initialized = False # Convenience function for global access def get_config() -> ConfigManager: """ Get the global ConfigManager instance. Returns: ConfigManager singleton instance """ return ConfigManager() # Configuration shortcuts def get_setting(key: str, default: Any = None) -> Any: """ Quick access to configuration settings. Args: key: Configuration key (dot notation) default: Default value if not found Returns: Configuration value """ return get_config().get(key, default) # --- End of ConfigManager ---