From 72dec3b757520bdf5be59296a81ee2672f2dd962 Mon Sep 17 00:00:00 2001 From: xingc Date: Fri, 16 May 2025 23:20:02 +0800 Subject: [PATCH] =?UTF-8?q?feat(chat):=20=E5=8D=95=E5=90=91=E8=81=8A?= =?UTF-8?q?=E5=A4=A9=E5=8F=AF=E4=BA=8C=E6=AC=A1=E6=BF=80=E6=B4=BB=20?= =?UTF-8?q?=E5=8D=95=E5=90=91=E8=81=8A=E5=A4=A9=E5=8F=AF=E4=BA=8C=E6=AC=A1?= =?UTF-8?q?=E6=BF=80=E6=B4=BB=EF=BC=8C=E6=9C=AA=E5=88=A0=E9=99=A4=E6=96=B9?= =?UTF-8?q?=E8=81=8A=E5=A4=A9=E6=B6=88=E6=81=AF=E4=BF=9D=E7=95=99=EF=BC=8C?= =?UTF-8?q?=E5=8F=8C=E5=90=91=E5=88=A0=E9=99=A4=E5=90=8E=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E6=B8=85=E7=A9=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...merage_chat_participant_and_read_status.py | 52 +++++++++++++++++ core/db/models.py | 28 +++------ core/routers/chat/handlers.py | 23 ++++---- core/routers/chat/routes.py | 57 +++++++++++++++---- core/routers/chat/services.py | 4 +- 5 files changed, 117 insertions(+), 47 deletions(-) create mode 100644 alembic/versions/43b2d2efe0f7_merage_chat_participant_and_read_status.py diff --git a/alembic/versions/43b2d2efe0f7_merage_chat_participant_and_read_status.py b/alembic/versions/43b2d2efe0f7_merage_chat_participant_and_read_status.py new file mode 100644 index 0000000..0164e86 --- /dev/null +++ b/alembic/versions/43b2d2efe0f7_merage_chat_participant_and_read_status.py @@ -0,0 +1,52 @@ +"""merage chat_participant and read_status + +Revision ID: 43b2d2efe0f7 +Revises: e9466c397aa2 +Create Date: 2025-05-16 22:20:04.592819 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import mysql + +# revision identifiers, used by Alembic. +revision: str = '43b2d2efe0f7' +down_revision: Union[str, None] = 'e9466c397aa2' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index('idx_read_status_on_chat_id', table_name='system_read_status') + op.drop_index('idx_read_status_on_user_id', table_name='system_read_status') + op.drop_table('system_read_status') + op.add_column('system_chat_participant', sa.Column('last_read_message_id', sa.Integer(), nullable=True)) + op.add_column('system_chat_participant', sa.Column('last_del_message_id', sa.Integer(), nullable=True)) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('system_chat_participant', 'last_del_message_id') + op.drop_column('system_chat_participant', 'last_read_message_id') + op.create_table('system_read_status', + sa.Column('last_read_message_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=True), + sa.Column('user_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False), + sa.Column('chat_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False), + sa.Column('id', mysql.INTEGER(display_width=11), autoincrement=True, nullable=False, comment='自增id'), + sa.Column('created_at', mysql.DATETIME(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False, comment='创建时间'), + sa.Column('updated_at', mysql.DATETIME(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False, comment='更新时间'), + sa.ForeignKeyConstraint(['chat_id'], ['system_chats.id'], name='system_read_status_ibfk_1'), + sa.ForeignKeyConstraint(['user_id'], ['system_users.user_id'], name='system_read_status_ibfk_2'), + sa.PrimaryKeyConstraint('id'), + comment='聊天消息读取状态表', + mysql_comment='聊天消息读取状态表', + mysql_default_charset='utf8mb4', + mysql_engine='InnoDB' + ) + op.create_index('idx_read_status_on_user_id', 'system_read_status', ['user_id'], unique=False) + op.create_index('idx_read_status_on_chat_id', 'system_read_status', ['chat_id'], unique=False) + # ### end Alembic commands ### diff --git a/core/db/models.py b/core/db/models.py index 6d136db..54079e3 100644 --- a/core/db/models.py +++ b/core/db/models.py @@ -99,7 +99,7 @@ class User(BaseModel, DateModel): ) chats: Mapped[List['Chat']] = relationship(secondary='system_chat_participant', back_populates='users') messages: Mapped[List['Message']] = relationship(back_populates='user') - read_statuses: Mapped[List['ReadStatus']] = relationship(back_populates='user') + chat_participants: Mapped[List['ChatParticipant']] = relationship(back_populates='user') def __str__(self): return f'{self.user_id}' @@ -264,7 +264,7 @@ class Chat(BaseModel, DateModel): users: Mapped[List['User']] = relationship(secondary='system_chat_participant', back_populates='chats') messages: Mapped[List['Message']] = relationship(back_populates='chat', cascade='all,delete') is_deleted: Mapped[bool] = mapped_column(Boolean, default=False) - read_statuses: Mapped[List['ReadStatus']] = relationship(back_populates='chat', cascade='all,delete') + chat_participants: Mapped[List['ChatParticipant']] = relationship(back_populates='chat', cascade='all,delete') class Message(BaseModel, DateModel): @@ -292,25 +292,6 @@ class Message(BaseModel, DateModel): return f'{self.content}' -class ReadStatus(BaseModel, DateModel): - __tablename__ = 'system_read_status' - __table_args__ = ( - Index('idx_read_status_on_chat_id', 'chat_id'), - Index('idx_read_status_on_user_id', 'user_id'), - {'comment': '聊天消息读取状态表'} - ) - - last_read_message_id: Mapped[int] = mapped_column(nullable=True) - user_id: Mapped[int] = mapped_column(ForeignKey('system_users.user_id')) - chat_id: Mapped[int] = mapped_column(ForeignKey('system_chats.id')) - - chat: Mapped['Chat'] = relationship(back_populates='read_statuses') - user: Mapped['User'] = relationship(back_populates='read_statuses') - - def __str__(self): - return f'User: {self.user_id}, Message: {self.last_read_message_id}' - - class PointsTransaction(BaseModel, DateModel): __tablename__ = 'system_points_transaction' __table_args__ = ( @@ -335,3 +316,8 @@ class ChatParticipant(BaseModel): user_id: Mapped[int] = mapped_column(Integer, ForeignKey('system_users.user_id'), primary_key=True) chat_id: Mapped[int] = mapped_column(Integer, ForeignKey('system_chats.id'), primary_key=True) is_deleted: Mapped[bool] = mapped_column(Boolean, default=False) + last_read_message_id: Mapped[int] = mapped_column(default=0, nullable=True) + last_del_message_id: Mapped[int] = mapped_column(default=0, nullable=True) + + chat: Mapped['Chat'] = relationship(back_populates='chat_participants') + user: Mapped['User'] = relationship(back_populates='chat_participants') diff --git a/core/routers/chat/handlers.py b/core/routers/chat/handlers.py index 19884a1..f8aead0 100644 --- a/core/routers/chat/handlers.py +++ b/core/routers/chat/handlers.py @@ -10,7 +10,7 @@ from fastapi import WebSocket from sqlalchemy.ext.asyncio import AsyncSession from core.db.crud import CRUD -from core.db.models import ReadStatus, User, Chat, Message +from core.db.models import ChatParticipant, User, Chat, Message from .managers import SocketManager from . import schema @@ -29,7 +29,6 @@ async def send_new_message( incoming_message: schema.WSMessage, ): chat_guid: str = str(incoming_message.chat_guid) - # chat: Chat | None = await CRUD(db_session, Chat).get(filters={'guid': chat_guid, 'is_deleted': False} chat: Chat | None = await get_chat(chat_guid, current_user, db_session) # 判断是否为新聊天 @@ -73,16 +72,7 @@ async def send_new_message( content=message.content, friend_user=current_user.dict(), ) - send_status_replay = schema.WSMessageOut( - type='chat.send_status_reply', - message_id=message.id, - message_guid=message.guid, - message_type=message.message_type, - chat_guid=chat.guid, - ) - await socket_manager.broadcast_to_chat(chat_guid, send_message_schema.model_dump_json()) - await websocket.send_text(send_status_replay.model_dump_json()) @socket_manager.handler('chat.message_read') @@ -90,16 +80,23 @@ async def read_message( websocket: WebSocket, db_session: AsyncSession, current_user: User, + chats: dict, incoming_message: schema.WSMessage, ): last_read_message_id: str | None = incoming_message.content if not last_read_message_id: return - db_crud = CRUD(db_session, ReadStatus) + db_crud = CRUD(db_session, ChatParticipant) + chat_guid: str = str(incoming_message.chat_guid) + chat: Chat | None = await get_chat(chat_guid, current_user, db_session) + if chat is None: + await socket_manager.send_error('聊天尚未添加', websocket) + return + condition = { 'user_id': current_user.user_id, - 'chat_guid': incoming_message.chat_guid, + 'chat_id': chat.id, } result = await db_crud.update(filters=condition, data={'last_read_message_id': last_read_message_id}) if result is None: diff --git a/core/routers/chat/routes.py b/core/routers/chat/routes.py index fc21a8c..b573377 100644 --- a/core/routers/chat/routes.py +++ b/core/routers/chat/routes.py @@ -17,7 +17,7 @@ from sqlalchemy.orm import selectinload from core.base import schema as base_schema from core.db.crud import CRUD from core.db.engine import get_async_session -from core.db.models import Message, Chat, User, ReadStatus, ChatParticipant +from core.db.models import Message, Chat, User, ChatParticipant from core.routers.auth.schema import LoginUser from core.routers.auth.services import get_current_user_form_http, get_current_user_form_ws from core.base.exceptions import WebsocketTooManyRequests @@ -100,10 +100,14 @@ async def get_single_chat_messages( if not chat: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='您访问的聊天不存在') + read_status = await CRUD(db_session, ChatParticipant).get( + filters={'chat_id': chat.id, 'user_id': login_user.user_id}) + # 组装查询条件 conditions = and_( Message.chat_id == chat.id, - Message.id <= last_read_message_id + Message.id > (read_status.last_del_message_id or 0), + Message.id <= last_read_message_id, ) # 记录总数查询 @@ -152,11 +156,13 @@ async def create_new_chat( if payload.recipient_user_id == login_user.user_id: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='操作禁止') - recipient_user: User | None = await CRUD(db_session, User).get(filters={'user_id': payload.recipient_user_id}) + db_crud = CRUD(db_session, User) + + recipient_user: User | None = await db_crud.get(filters={'user_id': payload.recipient_user_id}) if recipient_user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='收件人不存在') - initiator_user: User = await CRUD(db_session, User).get(filters={'user_id': login_user.user_id}) + initiator_user: User = await db_crud.get(filters={'user_id': login_user.user_id}) # 查询是否已存在 对应且未关闭的聊天 query = ( select(Chat) @@ -170,7 +176,7 @@ async def create_new_chat( (ChatParticipant.chat_id == Chat.id) & (ChatParticipant.user_id == recipient_user.user_id) ), - Chat.is_deleted == False + # Chat.is_deleted == False ) ).order_by(Chat.created_at.desc()) ) @@ -184,16 +190,16 @@ async def create_new_chat( chat.users.append(recipient_user) db_session.add(chat) await db_session.flush() - - initiator_read_status = ReadStatus(chat_id=chat.id, user_id=initiator_user.user_id, last_read_message_id=0) - recipient_read_status = ReadStatus(chat_id=chat.id, user_id=recipient_user.user_id, last_read_message_id=0) - db_session.add_all([initiator_read_status, recipient_read_status]) await db_session.commit() except Exception as exc_info: await db_session.rollback() raise exc_info + if chat.is_deleted is True: + await db_crud.updates(model=Chat, data={'is_deleted': False}, filters={'id': chat.id}) + await db_crud.updates(model=ChatParticipant, data={'is_deleted': False}, filters={'chat_id': chat.id}) + return { 'data': { 'chat_guid': chat.guid, @@ -219,12 +225,21 @@ async def delete_single_chat( Chat.id.in_(subquery), Chat.guid == chat_guid )) + .options(selectinload(Chat.chat_participants)) ) result = await db_session.execute(query) chat = result.scalar_one_or_none() if chat is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='聊天不存在') + + deleted_status = [ + chat_participant.is_deleted + for chat_participant in chat.chat_participants if chat_participant.user_id != login_user.user_id + ] + if all(deleted_status): + await db_session.delete(chat) + await db_session.commit() else: db_curd = CRUD(db_session, Chat) await db_curd.update( @@ -232,10 +247,21 @@ async def delete_single_chat( filters={'id': chat.id}, ) + message_query = ( + select(Message).where(Message.chat_id == chat.id).order_by(Message.id.desc()).limit(1) + ) + message_result = await db_session.execute(message_query) + message: Message | None = message_result.scalar() + + condition = {'user_id': login_user.user_id, 'chat_id': chat.id} + data = {'is_deleted': True} + if message: + data['last_del_message_id'] = message.id + await db_curd.update( model=ChatParticipant, - data={'is_deleted': True}, - filters={'user_id': login_user.user_id, 'chat_id': chat.id}, + data=data, + filters=condition, ) return {'message': '聊天已删除'} @@ -287,6 +313,15 @@ async def get_chat_websocket( incoming_message=message, ) + send_status_replay = schema.WSMessageOut( + type='chat.send_status_reply', + message_id=-1, + message_guid=message.message_guid, + message_type=message.message_type, + chat_guid=message.chat_guid, + ) + await websocket.send_text(send_status_replay.model_dump_json(exclude_none=True)) + except (JSONDecodeError, AttributeError) as exc: logger.exception(f'Websocket error, detail: {exc}') await socket_manager.send_error('Wrong message format', websocket) diff --git a/core/routers/chat/services.py b/core/routers/chat/services.py index 4061327..842117a 100644 --- a/core/routers/chat/services.py +++ b/core/routers/chat/services.py @@ -11,7 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased, selectinload from core.base.exceptions import WebsocketTooManyRequests -from core.db.models import User, Chat, ReadStatus, Message, ChatParticipant +from core.db.models import User, Chat, Message, ChatParticipant from core.routers.auth.schema import LoginUser from core.utils.helper import format_date @@ -44,7 +44,7 @@ async def get_user_active_direct_chats( async def get_chat_unread_messages_count(db_session: AsyncSession, login_user: LoginUser, chats): """查询未读消息的预览""" # 查询消息情况 - read_status_alias = aliased(ReadStatus) + read_status_alias = aliased(ChatParticipant) query = ( select(Message.chat_id, func.count().label("message_count")) .join(