376 lines
14 KiB
Python
376 lines
14 KiB
Python
"""
|
|
Bot Configuration Manager
|
|
|
|
This module handles per-lobby bot configuration management including:
|
|
- Configuration schema discovery from bot providers
|
|
- Configuration storage and retrieval per lobby
|
|
- Real-time configuration updates
|
|
- Validation of configuration values
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
import httpx
|
|
from typing import Dict, List, Optional, Any
|
|
from pathlib import Path
|
|
|
|
from logger import logger
|
|
|
|
# Import shared models with fallback handling
|
|
try:
|
|
from ...shared.models import (
|
|
BotConfigSchema,
|
|
BotConfigParameter,
|
|
BotLobbyConfig,
|
|
BotConfigUpdateRequest,
|
|
BotConfigUpdateResponse,
|
|
BotConfigListResponse,
|
|
BotInfoModel
|
|
)
|
|
except ImportError:
|
|
try:
|
|
from shared.models import (
|
|
BotConfigSchema,
|
|
BotConfigParameter,
|
|
BotLobbyConfig,
|
|
BotConfigUpdateRequest,
|
|
BotConfigUpdateResponse,
|
|
BotConfigListResponse,
|
|
BotInfoModel
|
|
)
|
|
except ImportError:
|
|
# Create dummy models for standalone testing
|
|
from pydantic import BaseModel
|
|
|
|
class BotConfigSchema(BaseModel):
|
|
bot_name: str
|
|
version: str = "1.0"
|
|
parameters: List[Dict[str, Any]]
|
|
|
|
class BotLobbyConfig(BaseModel):
|
|
bot_name: str
|
|
lobby_id: str
|
|
provider_id: str
|
|
config_values: Dict[str, Any]
|
|
created_at: float
|
|
updated_at: float
|
|
created_by: str
|
|
|
|
|
|
class BotConfigManager:
|
|
"""Manages bot configurations for lobbies"""
|
|
|
|
def __init__(self, storage_dir: str = "./bot_configs"):
|
|
self.storage_dir = Path(storage_dir)
|
|
self.storage_dir.mkdir(exist_ok=True)
|
|
|
|
# In-memory cache for fast access
|
|
self.config_cache: Dict[str, Dict[str, BotLobbyConfig]] = {} # lobby_id -> bot_name -> config
|
|
self.schema_cache: Dict[str, BotConfigSchema] = {} # bot_name -> schema
|
|
|
|
# Load existing configurations
|
|
self._load_configurations()
|
|
|
|
def _get_config_file(self, lobby_id: str) -> Path:
|
|
"""Get configuration file path for a lobby"""
|
|
return self.storage_dir / f"lobby_{lobby_id}.json"
|
|
|
|
def _get_schema_file(self, bot_name: str) -> Path:
|
|
"""Get schema file path for a bot"""
|
|
return self.storage_dir / f"schema_{bot_name}.json"
|
|
|
|
def _load_configurations(self):
|
|
"""Load all configurations from disk"""
|
|
try:
|
|
# Load lobby configurations
|
|
for config_file in self.storage_dir.glob("lobby_*.json"):
|
|
try:
|
|
lobby_id = config_file.stem.replace("lobby_", "")
|
|
with open(config_file, 'r') as f:
|
|
data = json.load(f)
|
|
|
|
self.config_cache[lobby_id] = {}
|
|
for bot_name, config_data in data.items():
|
|
config = BotLobbyConfig(**config_data)
|
|
self.config_cache[lobby_id][bot_name] = config
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load lobby config {config_file}: {e}")
|
|
|
|
# Load bot schemas
|
|
for schema_file in self.storage_dir.glob("schema_*.json"):
|
|
try:
|
|
bot_name = schema_file.stem.replace("schema_", "")
|
|
with open(schema_file, 'r') as f:
|
|
schema_data = json.load(f)
|
|
|
|
schema = BotConfigSchema(**schema_data)
|
|
self.schema_cache[bot_name] = schema
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load bot schema {schema_file}: {e}")
|
|
|
|
logger.info(f"Loaded configurations for {len(self.config_cache)} lobbies and {len(self.schema_cache)} bot schemas")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load configurations: {e}")
|
|
|
|
def _save_lobby_config(self, lobby_id: str):
|
|
"""Save lobby configuration to disk"""
|
|
try:
|
|
config_file = self._get_config_file(lobby_id)
|
|
|
|
if lobby_id not in self.config_cache:
|
|
return
|
|
|
|
# Convert to serializable format
|
|
data = {}
|
|
for bot_name, config in self.config_cache[lobby_id].items():
|
|
data[bot_name] = config.model_dump()
|
|
|
|
with open(config_file, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to save lobby config {lobby_id}: {e}")
|
|
|
|
def _save_bot_schema(self, bot_name: str):
|
|
"""Save bot schema to disk"""
|
|
try:
|
|
if bot_name not in self.schema_cache:
|
|
return
|
|
|
|
schema_file = self._get_schema_file(bot_name)
|
|
schema_data = self.schema_cache[bot_name].model_dump()
|
|
|
|
with open(schema_file, 'w') as f:
|
|
json.dump(schema_data, f, indent=2)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to save bot schema {bot_name}: {e}")
|
|
|
|
async def discover_bot_config_schema(self, bot_name: str, provider_url: str) -> Optional[BotConfigSchema]:
|
|
"""Discover configuration schema from bot provider"""
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
# Try to get configuration schema from bot provider
|
|
response = await client.get(
|
|
f"{provider_url}/bots/{bot_name}/config-schema",
|
|
timeout=10.0
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
schema_data = response.json()
|
|
schema = BotConfigSchema(**schema_data)
|
|
|
|
# Cache the schema
|
|
self.schema_cache[bot_name] = schema
|
|
self._save_bot_schema(bot_name)
|
|
|
|
logger.info(f"Discovered config schema for bot {bot_name}")
|
|
return schema
|
|
else:
|
|
logger.warning(f"Bot {bot_name} does not support configuration (HTTP {response.status_code})")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Failed to discover config schema for bot {bot_name}: {e}")
|
|
|
|
return None
|
|
|
|
def get_bot_config_schema(self, bot_name: str) -> Optional[BotConfigSchema]:
|
|
"""Get cached configuration schema for a bot"""
|
|
return self.schema_cache.get(bot_name)
|
|
|
|
def get_lobby_bot_config(self, lobby_id: str, bot_name: str) -> Optional[BotLobbyConfig]:
|
|
"""Get bot configuration for a specific lobby"""
|
|
if lobby_id in self.config_cache and bot_name in self.config_cache[lobby_id]:
|
|
return self.config_cache[lobby_id][bot_name]
|
|
return None
|
|
|
|
def get_lobby_configs(self, lobby_id: str) -> List[BotLobbyConfig]:
|
|
"""Get all bot configurations for a lobby"""
|
|
if lobby_id in self.config_cache:
|
|
return list(self.config_cache[lobby_id].values())
|
|
return []
|
|
|
|
def set_bot_config(self,
|
|
lobby_id: str,
|
|
bot_name: str,
|
|
provider_id: str,
|
|
config_values: Dict[str, Any],
|
|
session_id: str) -> BotLobbyConfig:
|
|
"""Set or update bot configuration for a lobby"""
|
|
|
|
# Validate configuration against schema if available
|
|
schema = self.get_bot_config_schema(bot_name)
|
|
if schema:
|
|
validated_values = self._validate_config_values(config_values, schema)
|
|
else:
|
|
validated_values = config_values
|
|
|
|
# Create or update configuration
|
|
now = time.time()
|
|
|
|
existing_config = self.get_lobby_bot_config(lobby_id, bot_name)
|
|
if existing_config:
|
|
# Update existing
|
|
config = BotLobbyConfig(
|
|
bot_name=bot_name,
|
|
lobby_id=lobby_id,
|
|
provider_id=provider_id,
|
|
config_values=validated_values,
|
|
created_at=existing_config.created_at,
|
|
updated_at=now,
|
|
created_by=existing_config.created_by
|
|
)
|
|
else:
|
|
# Create new
|
|
config = BotLobbyConfig(
|
|
bot_name=bot_name,
|
|
lobby_id=lobby_id,
|
|
provider_id=provider_id,
|
|
config_values=validated_values,
|
|
created_at=now,
|
|
updated_at=now,
|
|
created_by=session_id
|
|
)
|
|
|
|
# Store in cache
|
|
if lobby_id not in self.config_cache:
|
|
self.config_cache[lobby_id] = {}
|
|
|
|
self.config_cache[lobby_id][bot_name] = config
|
|
|
|
# Save to disk
|
|
self._save_lobby_config(lobby_id)
|
|
|
|
logger.info(f"Updated config for bot {bot_name} in lobby {lobby_id}")
|
|
return config
|
|
|
|
def _validate_config_values(self, values: Dict[str, Any], schema: BotConfigSchema) -> Dict[str, Any]:
|
|
"""Validate configuration values against schema"""
|
|
validated = {}
|
|
|
|
for param in schema.parameters:
|
|
value = values.get(param.name)
|
|
|
|
# Check required parameters
|
|
if param.required and value is None:
|
|
if param.default_value is not None:
|
|
value = param.default_value
|
|
else:
|
|
raise ValueError(f"Required parameter '{param.name}' is missing")
|
|
|
|
# Skip None values for optional parameters
|
|
if value is None:
|
|
continue
|
|
|
|
# Type validation
|
|
if param.type == "string":
|
|
value = str(value)
|
|
if param.max_length and len(value) > param.max_length:
|
|
raise ValueError(f"Parameter '{param.name}' exceeds max length {param.max_length}")
|
|
|
|
elif param.type == "number":
|
|
value = float(value)
|
|
if param.min_value is not None and value < param.min_value:
|
|
raise ValueError(f"Parameter '{param.name}' below minimum {param.min_value}")
|
|
if param.max_value is not None and value > param.max_value:
|
|
raise ValueError(f"Parameter '{param.name}' above maximum {param.max_value}")
|
|
|
|
elif param.type == "boolean":
|
|
value = bool(value)
|
|
|
|
elif param.type == "select":
|
|
if param.options:
|
|
valid_values = [opt["value"] for opt in param.options]
|
|
if value not in valid_values:
|
|
raise ValueError(f"Parameter '{param.name}' must be one of {valid_values}")
|
|
|
|
elif param.type == "range":
|
|
value = float(value)
|
|
if param.min_value is not None and value < param.min_value:
|
|
value = param.min_value
|
|
if param.max_value is not None and value > param.max_value:
|
|
value = param.max_value
|
|
|
|
validated[param.name] = value
|
|
|
|
return validated
|
|
|
|
def delete_bot_config(self, lobby_id: str, bot_name: str) -> bool:
|
|
"""Delete bot configuration for a lobby"""
|
|
if lobby_id in self.config_cache and bot_name in self.config_cache[lobby_id]:
|
|
del self.config_cache[lobby_id][bot_name]
|
|
|
|
# Clean up empty lobby configs
|
|
if not self.config_cache[lobby_id]:
|
|
del self.config_cache[lobby_id]
|
|
config_file = self._get_config_file(lobby_id)
|
|
if config_file.exists():
|
|
config_file.unlink()
|
|
else:
|
|
self._save_lobby_config(lobby_id)
|
|
|
|
logger.info(f"Deleted config for bot {bot_name} in lobby {lobby_id}")
|
|
return True
|
|
|
|
return False
|
|
|
|
def delete_lobby_configs(self, lobby_id: str) -> bool:
|
|
"""Delete all bot configurations for a lobby"""
|
|
if lobby_id in self.config_cache:
|
|
del self.config_cache[lobby_id]
|
|
|
|
config_file = self._get_config_file(lobby_id)
|
|
if config_file.exists():
|
|
config_file.unlink()
|
|
|
|
logger.info(f"Deleted all configs for lobby {lobby_id}")
|
|
return True
|
|
|
|
return False
|
|
|
|
async def notify_bot_config_change(self,
|
|
provider_url: str,
|
|
bot_name: str,
|
|
lobby_id: str,
|
|
config: BotLobbyConfig):
|
|
"""Notify bot provider of configuration change"""
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
# Notify the bot provider of the configuration change
|
|
response = await client.post(
|
|
f"{provider_url}/bots/{bot_name}/config",
|
|
json={
|
|
"lobby_id": lobby_id,
|
|
"config_values": config.config_values
|
|
},
|
|
timeout=10.0
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
logger.info(f"Successfully notified bot {bot_name} of config change")
|
|
return True
|
|
else:
|
|
logger.warning(f"Failed to notify bot {bot_name} of config change: HTTP {response.status_code}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to notify bot {bot_name} of config change: {e}")
|
|
|
|
return False
|
|
|
|
def get_statistics(self) -> Dict[str, Any]:
|
|
"""Get configuration manager statistics"""
|
|
total_configs = sum(len(lobby_configs) for lobby_configs in self.config_cache.values())
|
|
|
|
return {
|
|
"total_lobbies": len(self.config_cache),
|
|
"total_configs": total_configs,
|
|
"cached_schemas": len(self.schema_cache),
|
|
"lobbies": {
|
|
lobby_id: len(configs)
|
|
for lobby_id, configs in self.config_cache.items()
|
|
}
|
|
}
|