koko210Serve
d44f08af18
Add restore_runtime_settings() to ConfigManager that reads config_runtime.yaml on startup and restores persisted values into globals: - LANGUAGE_MODE, AUTONOMOUS_DEBUG, VOICE_DEBUG_MODE - USE_CHESHIRE_CAT, PREFER_AMD_GPU, DM_MOOD Add missing persistence calls to API endpoints: - POST /language/set now persists to config_runtime.yaml - POST /voice/debug-mode now persists to config_runtime.yaml - POST /memory/toggle now persists to config_runtime.yaml Call restore_runtime_settings() in on_ready() after evil/bipolar restore. Resolves #22
402 lines
14 KiB
Python
402 lines
14 KiB
Python
"""
|
||
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"
|
||
self.runtime_config_path = Path(__file__).parent.parent / "config_runtime.yaml"
|
||
|
||
# Memory directory for server configs and state
|
||
self.memory_dir = Path(__file__).parent / "memory"
|
||
self.memory_dir.mkdir(exist_ok=True)
|
||
|
||
# Load configurations
|
||
self.static_config: Dict = self._load_static_config()
|
||
self.runtime_config: Dict = self._load_runtime_config()
|
||
|
||
# Runtime state (not persisted)
|
||
self.runtime_state: Dict = {
|
||
"dm_mood": "neutral",
|
||
"evil_mode": False,
|
||
"bipolar_mode": False,
|
||
"language_mode": "english",
|
||
"current_gpu": "nvidia",
|
||
}
|
||
|
||
# Load persisted state
|
||
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.runtime_state["current_gpu"] = gpu_state.get("current_gpu", "nvidia")
|
||
logger.debug(f"✅ Loaded GPU state: {self.runtime_state['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.
|
||
|
||
Args:
|
||
key_path: Specific key to reset, or None to reset all runtime 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()
|
||
|
||
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 ==========
|
||
|
||
def get_state(self, key: str, default: Any = None) -> Any:
|
||
"""Get runtime state value (not persisted to config)."""
|
||
return self.runtime_state.get(key, default)
|
||
|
||
def set_state(self, key: str, value: Any):
|
||
"""Set runtime state value."""
|
||
self.runtime_state[key] = value
|
||
logger.debug(f"📊 State: {key} = {value}")
|
||
|
||
# ========== Server Configuration ==========
|
||
|
||
def get_server_config(self, guild_id: int) -> Dict:
|
||
"""Get configuration for a specific server."""
|
||
server_config_file = self.memory_dir / "servers_config.json"
|
||
|
||
try:
|
||
if server_config_file.exists():
|
||
with open(server_config_file, "r") as f:
|
||
all_servers = json.load(f)
|
||
return all_servers.get(str(guild_id), {})
|
||
except Exception as e:
|
||
logger.error(f"❌ Failed to load server config: {e}")
|
||
|
||
return {}
|
||
|
||
def set_server_config(self, guild_id: int, config: Dict):
|
||
"""Set configuration for a specific server."""
|
||
server_config_file = self.memory_dir / "servers_config.json"
|
||
|
||
try:
|
||
# Load existing config
|
||
all_servers = {}
|
||
if server_config_file.exists():
|
||
with open(server_config_file, "r") as f:
|
||
all_servers = json.load(f)
|
||
|
||
# Update server config
|
||
all_servers[str(guild_id)] = {
|
||
**all_servers.get(str(guild_id), {}),
|
||
**config,
|
||
"last_updated": datetime.now().isoformat()
|
||
}
|
||
|
||
# Save
|
||
with open(server_config_file, "w") as f:
|
||
json.dump(all_servers, f, indent=2)
|
||
|
||
logger.info(f"💾 Saved server config for {guild_id}")
|
||
except Exception as e:
|
||
logger.error(f"❌ Failed to save server config: {e}")
|
||
|
||
# ========== 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()
|