211 lines
9.2 KiB
Python
211 lines
9.2 KiB
Python
from cryptography.fernet import Fernet
|
|
|
|
from fastapi import APIRouter, Depends, status
|
|
|
|
from app.config import settings
|
|
from app.exceptions import UserDontHavePermissionException, MessageNotFoundException, UserCanNotReadThisChatException
|
|
from app.users.chat.dao import ChatDAO
|
|
from app.users.chat.shemas import SMessage, SLastMessages, SPinnedMessage, SPinnedChat, SDeletedUser, SChat, \
|
|
SDeletedChat
|
|
|
|
from app.users.dao import UserDAO
|
|
from app.users.dependencies import get_current_user
|
|
from app.users.auth import validate_user_access_to_chat, validate_user_admin, get_user_allowed_chats_id, \
|
|
ADMIN_USER_ID, check_verificated_user_with_exc
|
|
from app.users.models import Users
|
|
from app.users.schemas import SCreateInvitationLink, SUserAddedToChat
|
|
|
|
router = APIRouter(
|
|
prefix="/chat",
|
|
tags=["Чат"]
|
|
)
|
|
|
|
|
|
@router.get("", response_model=list[SChat])
|
|
async def get_all_chats(user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
result = await UserDAO.get_user_allowed_chats(user.id)
|
|
return result
|
|
|
|
|
|
@router.post("", status_code=status.HTTP_201_CREATED)
|
|
async def add_message_to_chat(
|
|
chat_id: int,
|
|
message: str,
|
|
user: Users = Depends(get_current_user)
|
|
):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
chats = await get_user_allowed_chats_id(user.id)
|
|
if chat_id not in chats:
|
|
raise UserDontHavePermissionException
|
|
send_message_to_chat = await ChatDAO.send_message(
|
|
user_id=user.id,
|
|
chat_id=chat_id,
|
|
message=message,
|
|
)
|
|
return send_message_to_chat
|
|
|
|
|
|
@router.delete("/delete_message", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_message_from_chat(
|
|
message_id: int,
|
|
user: Users = Depends(get_current_user)
|
|
):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
get_message_sender = await ChatDAO.get_message_by_id(message_id=message_id)
|
|
if get_message_sender is None:
|
|
raise MessageNotFoundException
|
|
if get_message_sender["user_id"] != user.id:
|
|
if not await validate_user_admin(user_id=user.id):
|
|
raise UserDontHavePermissionException
|
|
deleted_message = await ChatDAO.delete_message(message_id=message_id)
|
|
return deleted_message
|
|
|
|
|
|
@router.post("/create_chat", status_code=status.HTTP_201_CREATED)
|
|
async def create_chat(
|
|
user_to_exclude: int,
|
|
chat_name: str,
|
|
user: Users = Depends(get_current_user)
|
|
):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
if user.id == user_to_exclude:
|
|
raise UserCanNotReadThisChatException
|
|
chat_id = await ChatDAO.create(user_id=user_to_exclude, chat_name=chat_name, created_by=user.id)
|
|
user_added_to_chat = await ChatDAO.add_user_to_chat(user.id, chat_id)
|
|
await ChatDAO.add_user_to_chat(ADMIN_USER_ID, chat_id)
|
|
return user_added_to_chat
|
|
|
|
|
|
@router.get("/get_last_message/{chat_id}", response_model=list[SMessage])
|
|
async def get_last_message(chat_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
message = await ChatDAO.get_some_messages(chat_id=chat_id, message_number_from=0, messages_to_get=1)
|
|
if message is None:
|
|
raise MessageNotFoundException
|
|
for mes in message:
|
|
mes["created_at"] = mes["created_at"].isoformat()
|
|
return message
|
|
|
|
|
|
@router.get("/get_some_messages/{chat_id}", response_model=list[SMessage])
|
|
async def get_some_messages(
|
|
chat_id: int, last_messages: SLastMessages = Depends(), user: Users = Depends(get_current_user)
|
|
):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
messages = await ChatDAO.get_some_messages(
|
|
chat_id=chat_id,
|
|
message_number_from=last_messages.messages_loaded,
|
|
messages_to_get=last_messages.messages_to_get
|
|
)
|
|
if not messages:
|
|
raise MessageNotFoundException
|
|
for mes in messages:
|
|
mes["created_at"] = mes["created_at"].isoformat()
|
|
return messages
|
|
|
|
|
|
@router.get("/message/{chat_id}", response_model=SMessage)
|
|
async def get_message_by_id(chat_id: int, message_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
message = await ChatDAO.get_message_by_id(message_id=message_id)
|
|
if not message:
|
|
raise MessageNotFoundException
|
|
message = dict(message)
|
|
message["created_at"] = message["created_at"].isoformat()
|
|
return message
|
|
|
|
|
|
@router.get("/create_invitation_link", response_model=SCreateInvitationLink, status_code=status.HTTP_201_CREATED)
|
|
async def create_invitation_link(chat_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
cipher_suite = Fernet(settings.INVITATION_LINK_TOKEN_KEY)
|
|
invitation_token = cipher_suite.encrypt(str(chat_id).encode())
|
|
invitation_link = settings.INVITATION_LINK_HOST + '/api/chat/invite_to_chat/' + str(invitation_token).split("'")[1]
|
|
return {'invitation_link': invitation_link}
|
|
|
|
|
|
@router.get("/invite_to_chat/{invitation_token}", response_model=SUserAddedToChat, status_code=status.HTTP_200_OK)
|
|
async def invite_to_chat(invitation_token: str, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
invitation_token = invitation_token.encode()
|
|
cipher_suite = Fernet(settings.INVITATION_LINK_TOKEN_KEY)
|
|
chat_id = int(cipher_suite.decrypt(invitation_token))
|
|
chat = await ChatDAO.find_one_or_none(id=chat_id)
|
|
if user.id == chat.chat_for:
|
|
raise UserCanNotReadThisChatException
|
|
return {"user_added_to_chat": await ChatDAO.add_user_to_chat(chat_id=chat_id, user_id=user.id)}
|
|
|
|
|
|
@router.delete("/delete_chat/{chat_id}", response_model=SDeletedChat, status_code=status.HTTP_200_OK)
|
|
async def delete_chat(chat_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
chat = await ChatDAO.find_one_or_none(id=chat_id)
|
|
if user.id == chat.created_by:
|
|
return {'deleted_chat': await ChatDAO.delete_chat(chat_id)}
|
|
raise UserDontHavePermissionException
|
|
|
|
|
|
@router.delete("/delete_user_from_chat/{chat_id}", response_model=SDeletedUser)
|
|
async def delete_user_from_chat(chat_id: int, user_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
chat = await ChatDAO.find_one_or_none(id=chat_id)
|
|
if user.id == chat.created_by:
|
|
return {'deleted_user': await ChatDAO.delete_user(chat_id=chat_id, user_id=user_id)}
|
|
raise UserDontHavePermissionException
|
|
|
|
|
|
@router.post("/pinn_chat", response_model=SPinnedChat)
|
|
async def pinn_chat(chat_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
await ChatDAO.pinn_chat(chat_id=chat_id, user_id=user.id)
|
|
return {'chat_id': chat_id, 'user_id': user.id}
|
|
|
|
|
|
@router.delete("/unpinn_chat", response_model=SPinnedChat)
|
|
async def unpinn_chat(chat_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
await ChatDAO.unpinn_chat(chat_id=chat_id, user_id=user.id)
|
|
return {'chat_id': chat_id, 'user_id': user.id}
|
|
|
|
|
|
@router.get("/get_pinned_chats", response_model=list[SChat])
|
|
async def get_pinned_chats(user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
return await ChatDAO.get_pinned_chats(user_id=user.id)
|
|
|
|
|
|
@router.post("/pinn_message", response_model=SPinnedMessage)
|
|
async def pinn_message(chat_id: int, message_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
await ChatDAO.pinn_message(chat_id=chat_id, message_id=message_id, user_id=user.id)
|
|
return {'message_id': message_id, 'user_id': user.id, 'chat_id': chat_id}
|
|
|
|
|
|
@router.delete("/unpinn_message", response_model=SPinnedMessage)
|
|
async def unpinn_message(chat_id: int, message_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
message_pinner = await ChatDAO.get_message_pinner(chat_id=chat_id, message_id=message_id)
|
|
if message_pinner == user.id:
|
|
await ChatDAO.unpinn_message(chat_id=chat_id, message_id=message_id)
|
|
return {'message_id': message_id, 'user_id': user.id, 'chat_id': chat_id}
|
|
raise UserDontHavePermissionException
|
|
|
|
|
|
@router.get("/pinned_messages/{chat_id}", response_model=list[SMessage] | None)
|
|
async def pinned_messages(chat_id: int, user: Users = Depends(get_current_user)):
|
|
await check_verificated_user_with_exc(user_id=user.id)
|
|
await validate_user_access_to_chat(chat_id=chat_id, user_id=user.id)
|
|
messages = await ChatDAO.get_pinned_messages(chat_id=chat_id)
|
|
if messages:
|
|
for mes in messages:
|
|
mes["created_at"] = mes["created_at"].isoformat()
|
|
return messages
|