File size: 5,089 Bytes
1366db9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 |
# bot_registry.py
"""Bot registry pattern for managing different bot types."""
from typing import Any, Callable, Dict, List, Optional
from bot_constants import DEFAULT_DIALIN_EXAMPLE
from bot_runner_helpers import ensure_dialout_settings_array
from fastapi import HTTPException
class BotType:
"""Bot type configuration and handling."""
def __init__(
self,
name: str,
settings_creator: Callable[[Dict[str, Any]], Dict[str, Any]],
required_settings: list = None,
incompatible_with: list = None,
auto_add_settings: dict = None,
):
"""Initialize a bot type.
Args:
name: Name of the bot type
settings_creator: Function to create/update settings for this bot type
required_settings: List of settings this bot type requires
incompatible_with: List of bot types this one cannot be used with
auto_add_settings: Settings to add if this bot is being run in test mode
"""
self.name = name
self.settings_creator = settings_creator
self.required_settings = required_settings or []
self.incompatible_with = incompatible_with or []
self.auto_add_settings = auto_add_settings or {}
def has_test_mode(self, body: Dict[str, Any]) -> bool:
"""Check if this bot type is configured for test mode."""
return self.name in body and body[self.name].get("testInPrebuilt", False)
def create_settings(self, body: Dict[str, Any]) -> Dict[str, Any]:
"""Create or update settings for this bot type."""
body[self.name] = self.settings_creator(body)
return body
def prepare_for_test(self, body: Dict[str, Any]) -> Dict[str, Any]:
"""Add required settings for test mode if they don't exist."""
for setting, default_value in self.auto_add_settings.items():
if setting not in body:
body[setting] = default_value
return body
class BotRegistry:
"""Registry for managing different bot types."""
def __init__(self):
self.bots = {}
self.bot_validation_rules = []
def register(self, bot_type: BotType):
"""Register a bot type."""
self.bots[bot_type.name] = bot_type
return self
def get_bot(self, name: str) -> BotType:
"""Get a bot type by name."""
return self.bots.get(name)
def detect_bot_type(self, body: Dict[str, Any]) -> Optional[str]:
"""Detect which bot type to use based on configuration."""
# First check for test mode bots
for name, bot in self.bots.items():
if bot.has_test_mode(body):
return name
# Then check for specific combinations of settings
for name, bot in self.bots.items():
if name in body and all(req in body for req in bot.required_settings):
return name
# Default for dialin settings
if "dialin_settings" in body:
return DEFAULT_DIALIN_EXAMPLE
return None
def validate_bot_combination(self, body: Dict[str, Any]) -> List[str]:
"""Validate that bot types in the configuration are compatible."""
errors = []
bot_types_in_config = [name for name in self.bots.keys() if name in body]
# Check each bot type against its incompatible list
for bot_name in bot_types_in_config:
bot = self.bots[bot_name]
for incompatible in bot.incompatible_with:
if incompatible in body:
errors.append(
f"Cannot have both '{bot_name}' and '{incompatible}' in the same configuration"
)
return errors
def setup_configuration(self, body: Dict[str, Any]) -> Dict[str, Any]:
"""Set up bot configuration based on detected bot type."""
# Ensure dialout_settings is an array if present
body = ensure_dialout_settings_array(body)
# Detect which bot type to use
bot_type_name = self.detect_bot_type(body)
if not bot_type_name:
raise HTTPException(
status_code=400, detail="Configuration doesn't match any supported scenario"
)
# If we have a dialin scenario but no explicit bot type, add the default
if "dialin_settings" in body and bot_type_name == DEFAULT_DIALIN_EXAMPLE:
if bot_type_name not in body:
body[bot_type_name] = {}
# Get the bot type object
bot_type = self.get_bot(bot_type_name)
# Create/update settings for the bot type
body = bot_type.create_settings(body)
# If in test mode, add any required settings
if bot_type.has_test_mode(body):
body = bot_type.prepare_for_test(body)
# Validate bot combinations
errors = self.validate_bot_combination(body)
if errors:
error_message = "Invalid configuration: " + "; ".join(errors)
raise HTTPException(status_code=400, detail=error_message)
return body
|