Implement comprehensive config system and clean up codebase
Major changes: - Add Pydantic-based configuration system (bot/config.py, bot/config_manager.py) - Add config.yaml with all service URLs, models, and feature flags - Fix config.yaml path resolution in Docker (check /app/config.yaml first) - Remove Fish Audio API integration (tested feature that didn't work) - Remove hardcoded ERROR_WEBHOOK_URL, import from config instead - Add missing Pydantic models (LogConfigUpdateRequest, LogFilterUpdateRequest) - Enable Cheshire Cat memory system by default (USE_CHESHIRE_CAT=true) - Add .env.example template with all required environment variables - Add setup.sh script for user-friendly initialization - Update docker-compose.yml with proper env file mounting - Update .gitignore for config files and temporary files Config system features: - Static configuration from config.yaml - Runtime overrides from config_runtime.yaml - Environment variables for secrets (.env) - Web UI integration via config_manager - Graceful fallback to defaults Secrets handling: - Move ERROR_WEBHOOK_URL from hardcoded to .env - Add .env.example with all placeholder values - Document all required secrets - Fish API key and voice ID removed from .env Documentation: - CONFIG_README.md - Configuration system guide - CONFIG_SYSTEM_COMPLETE.md - Implementation summary - FISH_API_REMOVAL_COMPLETE.md - Removal record - SECRETS_CONFIGURED.md - Secrets setup record - BOT_STARTUP_FIX.md - Pydantic model fixes - MIGRATION_CHECKLIST.md - Setup checklist - WEB_UI_INTEGRATION_COMPLETE.md - Web UI config guide - Updated readmes/README.md with new features
This commit is contained in:
353
bot/config_manager.py
Normal file
353
bot/config_manager.py
Normal file
@@ -0,0 +1,353 @@
|
||||
"""
|
||||
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 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()
|
||||
Reference in New Issue
Block a user