|
|
from app.backend.models.users import User, get_user_chats |
|
|
from app.backend.models.chats import ( |
|
|
new_chat, |
|
|
get_chat_by_id, |
|
|
get_chats_by_user_id, |
|
|
refresh_title, |
|
|
) |
|
|
from app.backend.models.messages import get_messages_by_chat_id, Message |
|
|
from app.settings import BASE_DIR |
|
|
from fastapi import HTTPException |
|
|
from datetime import datetime, timedelta |
|
|
import os |
|
|
|
|
|
|
|
|
def create_new_chat(title: str | None, user: User) -> dict: |
|
|
chat_id = new_chat(title, user) |
|
|
try: |
|
|
path_to_chat = os.path.join( |
|
|
BASE_DIR, |
|
|
"chats_storage", |
|
|
f"user_id={user.id}", |
|
|
f"chat_id={chat_id}", |
|
|
"documents", |
|
|
) |
|
|
os.makedirs(path_to_chat, exist_ok=True) |
|
|
except Exception as e: |
|
|
print(e) |
|
|
raise HTTPException(500, "Unable to create a new chat") |
|
|
|
|
|
return {"url": f"/chats/id={chat_id}", "chat_id": chat_id} |
|
|
|
|
|
|
|
|
def dump_messages_dict(messages: list[Message], dst: dict) -> None: |
|
|
history = [] |
|
|
|
|
|
for message in messages: |
|
|
history.append({"role": message.sender, "content": message.content}) |
|
|
print(message.sender, message.content[:100]) |
|
|
|
|
|
dst.update({"history": history}) |
|
|
|
|
|
|
|
|
def get_chat_with_messages(id: int) -> dict: |
|
|
response = {"chat_id": id} |
|
|
|
|
|
chat = get_chat_by_id(id=id) |
|
|
if chat is None: |
|
|
raise HTTPException(418, f"Invalid chat id. Chat with id={id} does not exists!") |
|
|
|
|
|
messages = get_messages_by_chat_id(id=id) |
|
|
dump_messages_dict(messages, response) |
|
|
|
|
|
return response |
|
|
|
|
|
|
|
|
def create_dict_from_chat(chat) -> dict: |
|
|
return {"id": chat.id, "title": chat.title} |
|
|
|
|
|
|
|
|
def list_user_chats(user_id: int) -> list[dict]: |
|
|
current_date = datetime.now() |
|
|
|
|
|
today = [] |
|
|
last_week = [] |
|
|
last_month = [] |
|
|
other = [] |
|
|
|
|
|
chats = get_chats_by_user_id(user_id) |
|
|
for chat in chats: |
|
|
if current_date - timedelta(days=1) <= chat.created_at: |
|
|
today.append(chat) |
|
|
elif current_date - timedelta(weeks=1) <= chat.created_at: |
|
|
last_week.append(chat) |
|
|
elif current_date - timedelta(weeks=4) <= chat.created_at: |
|
|
last_month.append(chat) |
|
|
else: |
|
|
other.append(chat) |
|
|
|
|
|
result = [] |
|
|
|
|
|
|
|
|
if len(today): |
|
|
result.append( |
|
|
{"title": "TODAY", "chats": [create_dict_from_chat(chat) for chat in today]} |
|
|
) |
|
|
if len(last_week): |
|
|
result.append( |
|
|
{ |
|
|
"title": "LAST WEEK", |
|
|
"chats": [create_dict_from_chat(chat) for chat in last_week], |
|
|
} |
|
|
) |
|
|
if len(last_month): |
|
|
result.append( |
|
|
{ |
|
|
"title": "LAST MONTH", |
|
|
"chats": [create_dict_from_chat(chat) for chat in last_month], |
|
|
} |
|
|
) |
|
|
if len(other): |
|
|
result.append( |
|
|
{"title": "LATER", "chats": [create_dict_from_chat(chat) for chat in other]} |
|
|
) |
|
|
|
|
|
return result |
|
|
|
|
|
|
|
|
def verify_ownership_rights(user: User, chat_id: int) -> bool: |
|
|
return chat_id in [chat.id for chat in get_user_chats(user)] |
|
|
|
|
|
|
|
|
def update_title(chat_id: int) -> bool: |
|
|
return refresh_title(chat_id) |
|
|
|
