File size: 3,916 Bytes
56dc677 4a13628 56dc677 95cb26e 56dc677 4a13628 56dc677 4a13628 56dc677 4a13628 56dc677 95cb26e 56dc677 95cb26e 56dc677 95cb26e 56dc677 4a13628 56dc677 |
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 |
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) |