malek-messaoudii
Refactor audio processing and chatbot services; enhance STT and TTS functionalities with base64 support and session management
56dc677
| import base64 | |
| import uuid | |
| from typing import Optional, Dict, Any | |
| from datetime import datetime | |
| from models.audio import ChatbotResponse, UserMessage | |
| from services.tts_service import SimpleTTSService # Use simple version | |
| from services.stt_service import STTService # Use basic version | |
| class ChatbotService: | |
| def __init__(self): | |
| self.tts_service = SimpleTTSService() # Use simple TTS | |
| self.stt_service = STTService() # Use basic STT | |
| self.sessions: Dict[str, Dict[str, Any]] = {} | |
| self.initialized = False | |
| async def initialize(self): | |
| """Initialize the chatbot service""" | |
| await self.stt_service.initialize() | |
| self.initialized = True | |
| print("✓ Chatbot Service 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": | |
| # STT: Convert audio to text | |
| text_input = await self.stt_service.transcribe_audio_base64( | |
| user_message.content | |
| ) | |
| 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 | |
| try: | |
| audio_base64 = await self.tts_service.text_to_speech_base64(chatbot_text) | |
| except Exception as e: | |
| print(f"TTS error: {e}") | |
| audio_base64 = None | |
| # 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""" | |
| # Simple response logic - replace with your actual chatbot model | |
| user_input_lower = user_input.lower() | |
| if "hello" in user_input_lower or "hi" in user_input_lower: | |
| return "Hello! How can I assist you today?" | |
| elif "time" in user_input_lower: | |
| return f"The current time is {datetime.now().strftime('%H:%M')}" | |
| elif "help" in user_input_lower: | |
| return "I'm here to help you. You can ask me questions or request assistance." | |
| elif "audio" in user_input_lower or "voice" in user_input_lower: | |
| return "I can process both text and voice messages. Try sending me a voice message!" | |
| else: | |
| return f"I received your message: '{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() | |
| } | |
| else: | |
| self.sessions[session_id]["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) |