106 lines
2.8 KiB
Python
106 lines
2.8 KiB
Python
|
"""
|
||
|
|
Service for chatbot functionality without database dependency.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from typing import List, Dict, Any, Optional
|
||
|
|
|
||
|
|
class ChatbotService:
|
||
|
|
"""Service for chatbot functionality."""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
"""Initialize the chatbot service."""
|
||
|
|
# In-memory storage for chat history
|
||
|
|
self.chat_history = {}
|
||
|
|
self.current_chat_id = 0
|
||
|
|
|
||
|
|
def create_chat(self, user_id: str) -> int:
|
||
|
|
"""
|
||
|
|
Create a new chat session.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
user_id: ID of the user creating the chat.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
ID of the created chat.
|
||
|
|
"""
|
||
|
|
self.current_chat_id += 1
|
||
|
|
chat_id = self.current_chat_id
|
||
|
|
|
||
|
|
self.chat_history[chat_id] = {
|
||
|
|
'user_id': user_id,
|
||
|
|
'messages': []
|
||
|
|
}
|
||
|
|
|
||
|
|
return chat_id
|
||
|
|
|
||
|
|
def add_message(self, chat_id: int, content: str, is_user: bool = True) -> Dict[str, Any]:
|
||
|
|
"""
|
||
|
|
Add a message to a chat.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
chat_id: ID of the chat.
|
||
|
|
content: Message content.
|
||
|
|
is_user: Whether this is a user message (vs. bot message).
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Added message.
|
||
|
|
"""
|
||
|
|
if chat_id not in self.chat_history:
|
||
|
|
raise ValueError(f"Chat with ID {chat_id} not found")
|
||
|
|
|
||
|
|
message = {
|
||
|
|
'content': content,
|
||
|
|
'is_user': is_user,
|
||
|
|
'timestamp': self._get_timestamp()
|
||
|
|
}
|
||
|
|
|
||
|
|
self.chat_history[chat_id]['messages'].append(message)
|
||
|
|
|
||
|
|
return message
|
||
|
|
|
||
|
|
def get_chat_messages(self, chat_id: int) -> List[Dict[str, Any]]:
|
||
|
|
"""
|
||
|
|
Get all messages for a chat.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
chat_id: ID of the chat.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
List of messages.
|
||
|
|
"""
|
||
|
|
if chat_id not in self.chat_history:
|
||
|
|
raise ValueError(f"Chat with ID {chat_id} not found")
|
||
|
|
|
||
|
|
return self.chat_history[chat_id]['messages']
|
||
|
|
|
||
|
|
def get_response(self, chat_id: int, message: str) -> str:
|
||
|
|
"""
|
||
|
|
Get a response from the chatbot.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
chat_id: ID of the chat.
|
||
|
|
message: User message.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Bot response.
|
||
|
|
"""
|
||
|
|
# Add user message to chat history
|
||
|
|
self.add_message(chat_id, message, is_user=True)
|
||
|
|
|
||
|
|
# Simple echo response for now
|
||
|
|
response = f"You said: {message}"
|
||
|
|
|
||
|
|
# Add bot response to chat history
|
||
|
|
self.add_message(chat_id, response, is_user=False)
|
||
|
|
|
||
|
|
return response
|
||
|
|
|
||
|
|
def _get_timestamp(self) -> str:
|
||
|
|
"""Get current timestamp."""
|
||
|
|
from datetime import datetime
|
||
|
|
return datetime.utcnow().isoformat()
|
||
|
|
|
||
|
|
|
||
|
|
# Create a singleton instance
|
||
|
|
chatbot_service = ChatbotService()
|