""" Unified Configuration Manager for Miku Discord Bot. Handles: - Static configuration from config.yaml - Runtime overrides from Web UI - Per-server configuration - Priority system: Runtime > Static > Defaults - Persistence of runtime changes """ import json import os from pathlib import Path from typing import Any, Dict, Optional, Union from datetime import datetime import yaml from config import CONFIG, SECRETS from utils.logger import get_logger logger = get_logger('config_manager') class ConfigManager: """ Unified configuration manager with runtime overrides. Priority: 1. Runtime overrides (from Web UI, API, CLI) 2. Static config (from config.yaml) 3. Hardcoded defaults (fallback) """ def __init__(self, config_path: Optional[str] = None): """Initialize configuration manager.""" self.config_path = Path(config_path) if config_path else Path(__file__).parent.parent / "config.yaml" # Memory directory for server configs and state # This directory is volume-mounted in Docker (./bot/memory:/app/memory) self.memory_dir = Path(__file__).parent / "memory" self.memory_dir.mkdir(exist_ok=True) # Runtime config must live inside memory_dir so it persists across container restarts self.runtime_config_path = self.memory_dir / "config_runtime.yaml" # Load configurations self.static_config: Dict = self._load_static_config() self.runtime_config: Dict = self._load_runtime_config() # GPU state (the only piece of runtime_state that is persisted to its own file) self._current_gpu: str = "nvidia" # Load persisted state (GPU) self._load_runtime_state() logger.info("✅ ConfigManager initialized") def _load_static_config(self) -> Dict: """Load static configuration from config.yaml.""" if not self.config_path.exists(): logger.warning(f"⚠️ config.yaml not found: {self.config_path}") return {} try: with open(self.config_path, "r") as f: config = yaml.safe_load(f) or {} logger.debug(f"✅ Loaded static config from {self.config_path}") return config except Exception as e: logger.error(f"❌ Failed to load config.yaml: {e}") return {} def _load_runtime_config(self) -> Dict: """Load runtime overrides from config_runtime.yaml.""" if not self.runtime_config_path.exists(): logger.debug("ℹ️ config_runtime.yaml not found (no overrides)") return {} try: with open(self.runtime_config_path, "r") as f: config = yaml.safe_load(f) or {} logger.debug(f"✅ Loaded runtime config from {self.runtime_config_path}") return config except Exception as e: logger.error(f"❌ Failed to load config_runtime.yaml: {e}") return {} def _load_runtime_state(self): """Load runtime state from memory files.""" # Load GPU state gpu_state_file = self.memory_dir / "gpu_state.json" try: if gpu_state_file.exists(): with open(gpu_state_file, "r") as f: gpu_state = json.load(f) self._current_gpu = gpu_state.get("current_gpu", "nvidia") logger.debug(f"✅ Loaded GPU state: {self._current_gpu}") except Exception as e: logger.error(f"❌ Failed to load GPU state: {e}") def restore_runtime_settings(self): """ Restore persisted runtime settings from config_runtime.yaml into globals. Called once at startup (in on_ready) so that settings changed via the Web UI or API survive bot restarts. Settings with their own persistence (EVIL_MODE, BIPOLAR_MODE) are handled by their respective modules and are intentionally skipped here. """ import globals as g # Map: config_runtime.yaml key path -> (globals attribute, converter) _SETTINGS_MAP = { "discord.language_mode": ("LANGUAGE_MODE", str), "autonomous.debug_mode": ("AUTONOMOUS_DEBUG", bool), "voice.debug_mode": ("VOICE_DEBUG_MODE", bool), "memory.use_cheshire_cat": ("USE_CHESHIRE_CAT", bool), "gpu.prefer_amd": ("PREFER_AMD_GPU", bool), } restored = [] for key_path, (attr, converter) in _SETTINGS_MAP.items(): value = self._get_nested_value(self.runtime_config, key_path) if value is not None: try: setattr(g, attr, converter(value)) restored.append(f"{attr}={getattr(g, attr)}") except (ValueError, TypeError) as exc: logger.warning(f"⚠️ Could not restore {key_path}: {exc}") # DM mood needs special handling (load description too) dm_mood = self._get_nested_value(self.runtime_config, "runtime.mood.dm_mood") if dm_mood and isinstance(dm_mood, str) and dm_mood in getattr(g, "AVAILABLE_MOODS", []): g.DM_MOOD = dm_mood try: from utils.moods import load_mood_description g.DM_MOOD_DESCRIPTION = load_mood_description(dm_mood) except Exception: g.DM_MOOD_DESCRIPTION = f"I'm feeling {dm_mood} today." restored.append(f"DM_MOOD={dm_mood}") if restored: logger.info(f"🔄 Restored {len(restored)} runtime settings: {', '.join(restored)}") else: logger.debug("ℹ️ No runtime settings to restore") def get(self, key_path: str, default: Any = None) -> Any: """ Get configuration value with priority system. Args: key_path: Dot-separated path (e.g., "discord.language_mode") default: Fallback value if not found Returns: Configuration value (runtime > static > default) """ # Try runtime config first value = self._get_nested_value(self.runtime_config, key_path) if value is not None: logger.debug(f"⚡ Runtime config: {key_path} = {value}") return value # Try static config second value = self._get_nested_value(self.static_config, key_path) if value is not None: logger.debug(f"📄 Static config: {key_path} = {value}") return value # Return default logger.debug(f"⚙️ Default value: {key_path} = {default}") return default def _get_nested_value(self, config: Dict, key_path: str) -> Any: """Get nested value from config using dot notation.""" keys = key_path.split(".") value = config for key in keys: if isinstance(value, dict) and key in value: value = value[key] else: return None return value def set(self, key_path: str, value: Any, persist: bool = True): """ Set configuration value. Args: key_path: Dot-separated path (e.g., "discord.language_mode") value: New value to set persist: Whether to save to config_runtime.yaml """ # Set in runtime config keys = key_path.split(".") config = self.runtime_config for key in keys[:-1]: if key not in config: config[key] = {} config = config[key] config[keys[-1]] = value logger.info(f"✅ Config set: {key_path} = {value}") # Persist if requested if persist: self.save_runtime_config() def save_runtime_config(self): """Save runtime configuration to config_runtime.yaml.""" try: with open(self.runtime_config_path, "w") as f: yaml.dump(self.runtime_config, f, default_flow_style=False) logger.info(f"💾 Saved runtime config to {self.runtime_config_path}") except Exception as e: logger.error(f"❌ Failed to save runtime config: {e}") def reset_to_defaults(self, key_path: Optional[str] = None): """ Reset configuration to defaults. Clears runtime overrides from config_runtime.yaml AND resets the corresponding globals to their default values so the change takes effect immediately without a restart. Args: key_path: Specific key to reset, or None to reset all runtime config """ import globals as g from config import CONFIG if key_path: # Remove specific key from runtime config self._remove_nested_key(self.runtime_config, key_path) logger.info(f"🔄 Reset {key_path} to default") else: # Clear all runtime config self.runtime_config = {} logger.info("🔄 Reset all config to defaults") self.save_runtime_config() # ---- Reset live globals to match defaults ---- # Map: config_runtime key path -> (globals attr, default from CONFIG) _DEFAULTS_MAP = { "discord.language_mode": ("LANGUAGE_MODE", CONFIG.discord.language_mode), "autonomous.debug_mode": ("AUTONOMOUS_DEBUG", CONFIG.autonomous.debug_mode), "voice.debug_mode": ("VOICE_DEBUG_MODE", CONFIG.voice.debug_mode), "memory.use_cheshire_cat": ("USE_CHESHIRE_CAT", CONFIG.cheshire_cat.enabled), "gpu.prefer_amd": ("PREFER_AMD_GPU", CONFIG.gpu.prefer_amd), } reset_items = [] if key_path: # Reset only the specific global if key_path in _DEFAULTS_MAP: attr, default = _DEFAULTS_MAP[key_path] setattr(g, attr, default) reset_items.append(f"{attr}={default}") else: # Reset all globals to defaults for kp, (attr, default) in _DEFAULTS_MAP.items(): setattr(g, attr, default) reset_items.append(f"{attr}={default}") # Also reset DM mood to neutral g.DM_MOOD = "neutral" g.DM_MOOD_DESCRIPTION = "I'm feeling neutral and balanced today." reset_items.append("DM_MOOD=neutral") if reset_items: logger.info(f"🔄 Reset {len(reset_items)} globals: {', '.join(reset_items)}") def _remove_nested_key(self, config: Dict, key_path: str): """Remove nested key from config.""" keys = key_path.split(".") obj = config for key in keys[:-1]: if isinstance(obj, dict) and key in obj: obj = obj[key] else: return if isinstance(obj, dict) and keys[-1] in obj: del obj[keys[-1]] # ========== Runtime State Management ========== @property def runtime_state(self) -> Dict: """ Return live runtime state assembled from globals (the actual source of truth). Previously this was a static dict that was never updated, causing /config/state to always return stale defaults. Now it reads the real values each time. """ import globals as g return { "dm_mood": getattr(g, "DM_MOOD", "neutral"), "evil_mode": getattr(g, "EVIL_MODE", False), "bipolar_mode": getattr(g, "BIPOLAR_MODE", False), "language_mode": getattr(g, "LANGUAGE_MODE", "english"), "current_gpu": self._current_gpu, } def get_state(self, key: str, default: Any = None) -> Any: """Get runtime state value.""" return self.runtime_state.get(key, default) def set_state(self, key: str, value: Any): """Set runtime state value. Only current_gpu is managed here; other state lives in globals.""" if key == "current_gpu": self._current_gpu = value logger.debug(f"📊 State: {key} = {value}") # ========== GPU State ========== def get_gpu(self) -> str: """Get current GPU selection.""" return self.get_state("current_gpu", "nvidia") def set_gpu(self, gpu: str): """Set current GPU selection and persist.""" gpu = gpu.lower() if gpu not in ["nvidia", "amd"]: logger.warning(f"⚠️ Invalid GPU: {gpu}") return False # Update state self.set_state("current_gpu", gpu) # Persist to file gpu_state_file = self.memory_dir / "gpu_state.json" try: state = { "current_gpu": gpu, "last_updated": datetime.now().isoformat() } with open(gpu_state_file, "w") as f: json.dump(state, f, indent=2) logger.info(f"💾 Saved GPU state: {gpu}") return True except Exception as e: logger.error(f"❌ Failed to save GPU state: {e}") return False # ========== Configuration Export ========== def get_full_config(self) -> Dict: """ Get full configuration (merged static + runtime). Useful for API responses and debugging. """ return { "static": self.static_config, "runtime": self.runtime_config, "state": self.runtime_state, "merged": self._merge_configs(self.static_config, self.runtime_config) } def _merge_configs(self, base: Dict, override: Dict) -> Dict: """Deep merge two dictionaries.""" result = base.copy() for key, value in override.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = self._merge_configs(result[key], value) else: result[key] = value return result # ========== Validation ========== def validate_config(self) -> tuple[bool, list[str]]: """ Validate current configuration. Returns: Tuple of (is_valid, list_of_errors) """ errors = [] # Check required secrets if not SECRETS.discord_bot_token or SECRETS.discord_bot_token.startswith("your_"): errors.append("DISCORD_BOT_TOKEN not set or using placeholder") # Validate language mode language = self.get("discord.language_mode", "english") if language not in ["english", "japanese"]: errors.append(f"Invalid language_mode: {language}") # Validate GPU gpu = self.get_gpu() if gpu not in ["nvidia", "amd"]: errors.append(f"Invalid GPU selection: {gpu}") return len(errors) == 0, errors # ========== Global Instance ========== # Create global config manager instance config_manager = ConfigManager()