malek-messaoudii
Refactor requirements and chatbot services; enhance service initialization and error handling for TTS and STT functionalities
691fd14
| import base64 | |
| import uuid | |
| from typing import Optional, Dict, Any | |
| from datetime import datetime | |
| from models.audio import ChatbotResponse, UserMessage | |
| class ChatbotService: | |
| def __init__(self): | |
| self.sessions: Dict[str, Dict[str, Any]] = {} | |
| self.initialized = False | |
| self._initialize_services() | |
| def _initialize_services(self): | |
| """Initialize services""" | |
| try: | |
| from services.tts_service import SimpleTTSService | |
| self.tts_service = SimpleTTSService() | |
| print("✓ TTS service initialized") | |
| except ImportError as e: | |
| print(f"⚠️ TTS service not available: {e}") | |
| self.tts_service = None | |
| try: | |
| from services.stt_service import STTService | |
| self.stt_service = STTService() | |
| print("✓ STT service initialized") | |
| except ImportError as e: | |
| print(f"⚠️ STT service not available: {e}") | |
| self.stt_service = None | |
| async def initialize(self): | |
| """Initialize the chatbot service""" | |
| if self.stt_service: | |
| await self.stt_service.initialize() | |
| self.initialized = True | |
| print("✓ Chatbot Service fully initialized") | |
| async def process_user_message(self, user_message: UserMessage) -> ChatbotResponse: | |
| # Update session | |
| session = self._get_or_create_session(user_message.session_id) | |
| # Process message based on type | |
| if user_message.message_type == "audio" and self.stt_service: | |
| text_input = await self.stt_service.transcribe_audio_base64(user_message.content) | |
| elif user_message.message_type == "audio": | |
| text_input = "[Voice message received]" | |
| else: | |
| text_input = user_message.content | |
| # Add to conversation history | |
| session["conversation_history"].append({ | |
| "role": "user", | |
| "content": text_input, | |
| "timestamp": user_message.timestamp | |
| }) | |
| # Generate chatbot response | |
| chatbot_text = await self._generate_chatbot_response(text_input, session) | |
| # TTS: Convert response to audio | |
| audio_base64 = None | |
| if self.tts_service: | |
| audio_base64 = await self.tts_service.text_to_speech_base64(chatbot_text) | |
| # Create response | |
| response = ChatbotResponse( | |
| response_id=str(uuid.uuid4()), | |
| text_response=chatbot_text, | |
| audio_response=audio_base64, | |
| session_id=user_message.session_id | |
| ) | |
| # Add response to history | |
| session["conversation_history"].append({ | |
| "role": "assistant", | |
| "content": chatbot_text, | |
| "audio_response": audio_base64, | |
| "timestamp": response.timestamp | |
| }) | |
| return response | |
| async def _generate_chatbot_response(self, user_input: str, session: Dict[str, Any]) -> str: | |
| """Chatbot response generation logic""" | |
| user_input_lower = user_input.lower() | |
| if any(greet in user_input_lower for greet in ["hello", "hi", "hey"]): | |
| return "Hello! I'm your voice assistant. How can I help you today?" | |
| if "time" in user_input_lower: | |
| return f"The current time is {datetime.now().strftime('%H:%M')}" | |
| if "help" in user_input_lower: | |
| return "I can process both text and voice messages. Try sending me a voice note!" | |
| if "name" in user_input_lower: | |
| return "I'm your AI voice assistant. I'm here to help with your questions!" | |
| if "voice" in user_input_lower or "audio" in user_input_lower: | |
| return "Yes! I support voice messages. You can speak to me and I'll respond with audio too!" | |
| return f"I understand you said: '{user_input}'. How can I assist you further?" | |
| def _get_or_create_session(self, session_id: str) -> Dict[str, Any]: | |
| if session_id not in self.sessions: | |
| self.sessions[session_id] = { | |
| "conversation_history": [], | |
| "created_at": datetime.now(), | |
| "last_activity": datetime.now() | |
| } | |
| return self.sessions[session_id] | |
| def get_session_history(self, session_id: str) -> Optional[Dict[str, Any]]: | |
| return self.sessions.get(session_id) |