""" Bot Configuration Service for MCube Voice Bot Handles dynamic bot configuration based on DID number mapping """ import mysql.connector from mysql.connector import Error import logging from typing import Optional, Dict, Any from datetime import datetime from config import Config from services.log_utils import Log logger = logging.getLogger(__name__) class BotConfigurationService: """ Service for managing dynamic bot configurations based on DID number mapping. Maps DID numbers to specific bot configurations stored in database. """ def __init__(self): # Master database config (for DID mapping) self.master_db_config = { 'host': Config.DATABASE_HOST, 'database': 'mainvoicebot_master', # Master database for DID mapping 'user': Config.DATABASE_USER, 'password': Config.DATABASE_PASSWORD, 'charset': Config.DATABASE_CHARSET, 'autocommit': True } # Cluster database config (for bot configurations) self.cluster_db_config = { 'host': Config.DATABASE_HOST, 'database': Config.DATABASE_NAME, # devvoicelabs_cluster 'user': Config.DATABASE_USER, 'password': Config.DATABASE_PASSWORD, 'charset': Config.DATABASE_CHARSET, 'autocommit': True } Log.info("🤖 Bot Configuration Service initialized") def get_connection(self, database_type: str = 'cluster'): """ Get database connection for master or cluster database. Args: database_type: 'master' or 'cluster' Returns: Database connection """ try: config = self.master_db_config if database_type == 'master' else self.cluster_db_config connection = mysql.connector.connect(**config) return connection except Error as e: Log.error(f"❌ Error connecting to {database_type} database: {e}") raise def get_bot_id_by_did(self, did_number: str) -> Optional[int]: """ Get bot_id from DID number mapping in master database. Args: did_number: DID number to lookup Returns: bot_id if found, None otherwise """ connection = None cursor = None try: connection = self.get_connection('master') cursor = connection.cursor(dictionary=True) query = """ SELECT bot_id, business_id FROM did_numbers WHERE did_no = %s AND status = '1' """ cursor.execute(query, (did_number,)) result = cursor.fetchone() if result: bot_id = result['bot_id'] business_id = result['business_id'] Log.info(f"🔍 Found bot mapping - DID: {did_number}, Bot ID: {bot_id}, Business ID: {business_id}") return bot_id, business_id else: Log.warning(f"⚠️ No bot mapping found for DID: {did_number}") return None, None except Error as e: Log.error(f"❌ Error looking up DID {did_number}: {e}") return None, None finally: if cursor: cursor.close() if connection: connection.close() async def get_bot_id_by_did_async(self, did_number: str): """Async version of get_bot_id_by_did using thread pool.""" import asyncio def _sync_get_bot_id(): return self.get_bot_id_by_did(did_number) loop = asyncio.get_event_loop() return await loop.run_in_executor(None, _sync_get_bot_id) def get_business_name_by_id(self, business_id: int) -> Optional[str]: """ Get business name from business table in master database. Args: business_id: Business ID to lookup Returns: Business name if found, None otherwise """ connection = None cursor = None try: connection = self.get_connection('master') cursor = connection.cursor(dictionary=True) query = """ SELECT business_name FROM businesses WHERE business_id = %s AND status = 'active' """ cursor.execute(query, (business_id,)) result = cursor.fetchone() if result: business_name = result['business_name'] Log.info(f"🏢 Found business name - Business ID: {business_id}, Name: {business_name}") return business_name else: Log.warning(f"⚠️ No business found for Business ID: {business_id}") return None except Error as e: Log.error(f"❌ Error looking up business {business_id}: {e}") return None finally: if cursor: cursor.close() if connection: connection.close() def get_bot_configuration(self, business_id: int, bot_id: int) -> Optional[Dict[str, Any]]: """ Get bot configuration from cluster database. Args: business_id: Business ID bot_id: Bot ID Returns: Bot configuration dictionary or None """ connection = None cursor = None try: connection = self.get_connection('cluster') cursor = connection.cursor(dictionary=True) # Dynamic table name based on business_id table_name = f"{business_id}_bots" query = f""" SELECT bot_id, bot_name, agent_id FROM `{table_name}` WHERE bot_id = %s AND is_active = 1 """ cursor.execute(query, (bot_id,)) result = cursor.fetchone() if result: bot_config = { 'bot_id': result['bot_id'], 'bot_name': result['bot_name'], 'agent_id': result['agent_id'] } Log.info(f"🤖 Loaded bot configuration - Name: {result['bot_name']}, ID: {bot_id}") return bot_config else: Log.warning(f"⚠️ No active bot configuration found for Bot ID: {bot_id}") return None except Error as e: Log.error(f"❌ Error fetching bot configuration for Bot ID {bot_id}: {e}") return None finally: if cursor: cursor.close() if connection: connection.close() async def get_bot_configuration_by_did(self, did_number: str) -> Dict[str, Any]: """ Get complete bot configuration by DID number - OPTIMIZED for separate databases. All database operations run in thread pool to avoid blocking event loop. Args: did_number: DID number to lookup Returns: Dictionary containing bot configuration and system message """ try: # OPTIMIZATION: Use connection pooling and faster queries start_time = datetime.now() # Step 1: Get bot_id and business_id from DID mapping (Master DB) - NOW ASYNC result = await self.get_bot_id_by_did_async(did_number) if not result or result[0] is None or result[1] is None: Log.warning(f"⚠️ No bot mapping found for DID: {did_number}, using default configuration") return { 'bot_config': None, 'bot_name': 'Default Assistant', 'business_id': None, 'bot_id': None, 'company_name': Config.COMPANY_NAME } bot_id, business_id = result # OPTIMIZATION: Run remaining queries in parallel using asyncio import asyncio # Create tasks for parallel execution tasks = [] # Task 1: Get bot configuration (Cluster DB) bot_task = asyncio.create_task(self._get_bot_config_async(business_id, bot_id)) tasks.append(('bot_config', bot_task)) # Task 2: Get company name (Master DB) company_task = asyncio.create_task(self._get_company_name_async(business_id)) tasks.append(('company_name', company_task)) # Wait for both tasks to complete results = await asyncio.gather(*[task for _, task in tasks], return_exceptions=True) # Extract results bot_config = results[0] if not isinstance(results[0], Exception) else None company_name = results[1] if not isinstance(results[1], Exception) else None if not bot_config: Log.warning(f"⚠️ No bot configuration found for Bot ID: {bot_id}, using default") return { 'bot_config': None, 'bot_name': 'Default Assistant', 'business_id': business_id, 'bot_id': bot_id, 'company_name': company_name or Config.COMPANY_NAME } # Removed: System message building - not needed for ElevenLabs WebSocket (agents configured in dashboard) # Log timing total_time = (datetime.now() - start_time).total_seconds() Log.info(f"⚡ Bot configuration loaded in {total_time:.2f} seconds") return { 'bot_config': bot_config, 'bot_name': bot_config.get('name', 'Assistant'), 'business_id': business_id, 'bot_id': bot_id, 'company_name': company_name or Config.COMPANY_NAME } except Exception as e: Log.error(f"❌ Error getting bot configuration for DID {did_number}: {e}") return { 'bot_config': None, 'bot_name': 'Default Assistant', 'business_id': None, 'bot_id': None, 'company_name': Config.COMPANY_NAME } async def _get_bot_config_async(self, business_id: int, bot_id: int) -> Optional[Dict[str, Any]]: """Async version of bot configuration retrieval.""" import asyncio import mysql.connector def _sync_get_bot_config(): return self.get_bot_configuration(business_id, bot_id) # Run sync function in thread pool loop = asyncio.get_event_loop() return await loop.run_in_executor(None, _sync_get_bot_config) async def _get_company_name_async(self, business_id: int) -> Optional[str]: """Async version of company name retrieval.""" import asyncio def _sync_get_company_name(): return self.get_business_name_by_id(business_id) # Run sync function in thread pool loop = asyncio.get_event_loop() return await loop.run_in_executor(None, _sync_get_company_name) def test_database_connections(self) -> Dict[str, bool]: """ Test connections to both master and cluster databases. Returns: Dictionary with connection test results """ results = {'master': False, 'cluster': False} # Test master database connection try: connection = self.get_connection('master') cursor = connection.cursor() cursor.execute("SELECT 1") cursor.fetchone() cursor.close() connection.close() results['master'] = True Log.info("✅ Master database connection successful") except Exception as e: Log.error(f"❌ Master database connection failed: {e}") # Test cluster database connection try: connection = self.get_connection('cluster') cursor = connection.cursor() cursor.execute("SELECT 1") cursor.fetchone() cursor.close() connection.close() results['cluster'] = True Log.info("✅ Cluster database connection successful") except Exception as e: Log.error(f"❌ Cluster database connection failed: {e}") return results # Global instance bot_configuration_service = BotConfigurationService()