refactor(websocket): удалены старые WebSocket consumers
Удалены файлы старой архитектуры: - backend/communications/user_consumer.py - backend/communications/routing.py - backend/notifications/consumers.py - backend/notifications/routing.py Теперь вся WebSocket логика в backend/realtime/ При необходимости можно откатиться через git revert
This commit is contained in:
parent
0341751222
commit
825dd5f9e2
@ -1,10 +0,0 @@
|
||||
# backend\communications\routing.py
|
||||
from django.urls import re_path
|
||||
|
||||
from .user_consumer import UserConsumer
|
||||
|
||||
websocket_urlpatterns = [
|
||||
# Универсальный WebSocket для пользователя
|
||||
# Обрабатывает: чаты, уведомления, бейджи, онлайн-статус и др.
|
||||
re_path(r"^ws/$", UserConsumer.as_asgi(), name="ws_user"),
|
||||
]
|
||||
@ -1,738 +0,0 @@
|
||||
# communications/user_consumer.py
|
||||
"""
|
||||
WebSocket Consumer для пользователя.
|
||||
Один WebSocket на пользователя для всех real-time операций:
|
||||
- Чаты и сообщения
|
||||
- Уведомления
|
||||
- Обновления бейджей
|
||||
- Онлайн-статус
|
||||
- Календарь и другие real-time события
|
||||
"""
|
||||
import asyncio
|
||||
from channels.db import database_sync_to_async
|
||||
from channels.generic.websocket import AsyncJsonWebsocketConsumer
|
||||
from django.contrib.auth.models import AnonymousUser
|
||||
from django.db.models import Q
|
||||
from django.utils import timezone
|
||||
|
||||
from .models import Chat, ChatMembership, ChatReadState, Message, MessageReaction
|
||||
from .consumers import serialize_message # Используем существующую функцию
|
||||
|
||||
|
||||
class UserConsumer(AsyncJsonWebsocketConsumer):
|
||||
"""
|
||||
WebSocket Consumer для пользователя.
|
||||
Управляет всеми real-time обновлениями:
|
||||
- Чаты: обновления списка (бейдж, карточки), сообщения в активном чате
|
||||
- Реакции, редактирование, удаление сообщений
|
||||
- Индикатор "печатает..."
|
||||
- Уведомления (будущее)
|
||||
- Бейджи в sidebar (будущее)
|
||||
- Онлайн-статус (будущее)
|
||||
"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.ping_task = None
|
||||
self.ping_interval = 30 # Ping каждые 30 секунд
|
||||
self.active_chat_id = None # ID открытого чата
|
||||
self.subscribed_chats = set() # Чаты, на которые подписан
|
||||
self.user = None
|
||||
|
||||
async def connect(self):
|
||||
"""Подключение пользователя - подписка на все его чаты"""
|
||||
self.user = self.scope.get("user")
|
||||
|
||||
if not self.user or isinstance(self.user, AnonymousUser) or not self.user.is_authenticated:
|
||||
await self.close(code=4401)
|
||||
return
|
||||
|
||||
# Получаем все чаты пользователя
|
||||
chat_ids = await self._get_available_chat_ids(self.user)
|
||||
|
||||
# Подписываемся на группы всех чатов
|
||||
for chat_id in chat_ids:
|
||||
group_name = f"chat_{chat_id}"
|
||||
await self.channel_layer.group_add(group_name, self.channel_name)
|
||||
self.subscribed_chats.add(chat_id)
|
||||
|
||||
# Подписываемся на личный канал пользователя (для уведомлений)
|
||||
user_channel = f"user_{self.user.id}"
|
||||
await self.channel_layer.group_add(user_channel, self.channel_name)
|
||||
|
||||
await self.accept()
|
||||
|
||||
# Запускаем ping цикл для keepalive
|
||||
self.ping_task = asyncio.create_task(self._ping_loop())
|
||||
|
||||
print(f"[UserWS] User {self.user.id} connected, subscribed to {len(self.subscribed_chats)} chats")
|
||||
|
||||
async def disconnect(self, code):
|
||||
"""Отключение - отписка от всех групп"""
|
||||
# Останавливаем ping цикл
|
||||
if self.ping_task:
|
||||
self.ping_task.cancel()
|
||||
try:
|
||||
await self.ping_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
# Сбрасываем статус "набирает текст" если был активный чат
|
||||
if self.active_chat_id and self.user:
|
||||
chat = await self._get_chat(self.active_chat_id)
|
||||
if chat:
|
||||
await self._set_typing_status(chat, self.user, False)
|
||||
|
||||
# Отписываемся от всех групп чатов
|
||||
for chat_id in self.subscribed_chats:
|
||||
group_name = f"chat_{chat_id}"
|
||||
await self.channel_layer.group_discard(group_name, self.channel_name)
|
||||
|
||||
# Отписываемся от личного канала
|
||||
if self.user:
|
||||
user_channel = f"user_{self.user.id}"
|
||||
await self.channel_layer.group_discard(user_channel, self.channel_name)
|
||||
|
||||
print(f"[UserWS] User {self.user.id if self.user else 'unknown'} disconnected")
|
||||
|
||||
async def _ping_loop(self):
|
||||
"""Отправка ping каждые N секунд для keepalive"""
|
||||
try:
|
||||
while True:
|
||||
await asyncio.sleep(self.ping_interval)
|
||||
await self.send_json({
|
||||
"type": "ping",
|
||||
"timestamp": timezone.now().isoformat()
|
||||
})
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"[UserWS] Ping loop error: {e}")
|
||||
|
||||
async def receive_json(self, content, **kwargs):
|
||||
"""Обработка входящих сообщений от клиента"""
|
||||
if not isinstance(content, dict):
|
||||
return
|
||||
|
||||
action = content.get("action", "")
|
||||
|
||||
print(f"[UserWS] receive_json: action={action}, user={self.user.id}, active_chat={self.active_chat_id}")
|
||||
|
||||
# Управление активным чатом
|
||||
if action == "open_chat":
|
||||
await self._handle_open_chat(content)
|
||||
elif action == "close_chat":
|
||||
await self._handle_close_chat(content)
|
||||
|
||||
# Работа с сообщениями (только для активного чата)
|
||||
elif action == "send_message":
|
||||
await self._handle_send_message(content)
|
||||
elif action == "edit_message":
|
||||
await self._handle_edit_message(content)
|
||||
elif action == "delete_message":
|
||||
await self._handle_delete_message(content)
|
||||
|
||||
# Реакции
|
||||
elif action == "add_reaction":
|
||||
await self._handle_add_reaction(content)
|
||||
elif action == "remove_reaction":
|
||||
await self._handle_remove_reaction(content)
|
||||
|
||||
# Индикатор печати
|
||||
elif action == "typing":
|
||||
await self._handle_typing(content)
|
||||
elif action == "stop_typing":
|
||||
await self._handle_stop_typing()
|
||||
|
||||
# Отметка прочитанного
|
||||
elif action == "mark_read":
|
||||
await self._handle_mark_read(content)
|
||||
|
||||
# Голосования
|
||||
elif action == "vote_poll":
|
||||
await self._handle_vote_poll(content)
|
||||
|
||||
# ==================== Управление активным чатом ====================
|
||||
|
||||
async def _handle_open_chat(self, content):
|
||||
"""Открытие чата - установка активного чата и отправка начальной истории"""
|
||||
chat_id = content.get("chat_id")
|
||||
if not chat_id:
|
||||
return
|
||||
|
||||
chat_id = int(chat_id)
|
||||
|
||||
# Проверяем доступ
|
||||
chat = await self._get_chat(chat_id)
|
||||
if not chat or not await self._user_can_access(chat, self.user):
|
||||
await self.send_json({
|
||||
"type": "error",
|
||||
"error": "Chat not found or access denied",
|
||||
"chat_id": chat_id
|
||||
})
|
||||
return
|
||||
|
||||
# Устанавливаем активный чат
|
||||
self.active_chat_id = chat_id
|
||||
|
||||
# Отправляем начальную историю (если запрошена)
|
||||
if content.get("load_history", False):
|
||||
await self._send_initial_messages(chat_id)
|
||||
|
||||
# Помечаем как прочитанное
|
||||
await self._mark_read(chat, self.user)
|
||||
|
||||
await self.send_json({
|
||||
"type": "chat_opened",
|
||||
"chat_id": chat_id
|
||||
})
|
||||
|
||||
print(f"[UserWS] User {self.user.id} opened chat {chat_id}")
|
||||
|
||||
async def _handle_close_chat(self, content):
|
||||
"""Закрытие чата - сброс активного чата"""
|
||||
chat_id = content.get("chat_id")
|
||||
|
||||
if chat_id and int(chat_id) == self.active_chat_id:
|
||||
# Сбрасываем индикатор печати
|
||||
if self.active_chat_id:
|
||||
chat = await self._get_chat(self.active_chat_id)
|
||||
if chat:
|
||||
await self._set_typing_status(chat, self.user, False)
|
||||
|
||||
self.active_chat_id = None
|
||||
|
||||
await self.send_json({
|
||||
"type": "chat_closed",
|
||||
"chat_id": chat_id
|
||||
})
|
||||
|
||||
print(f"[UserWS] User {self.user.id} closed chat {chat_id}")
|
||||
|
||||
# ==================== Обработчики событий из channel layer ====================
|
||||
|
||||
async def chat_message(self, event):
|
||||
"""
|
||||
Новое сообщение в чате.
|
||||
Отправляем полные данные для активного чата,
|
||||
компактные - для обновления списка.
|
||||
"""
|
||||
chat_id = event.get("chat_id")
|
||||
payload = event.get("payload", {})
|
||||
|
||||
# Для активного чата - полное сообщение
|
||||
if chat_id == self.active_chat_id:
|
||||
await self.send_json({
|
||||
"type": "new_message",
|
||||
"chat_id": chat_id,
|
||||
"message": payload
|
||||
})
|
||||
|
||||
# Для списка чатов - компактное обновление
|
||||
await self.send_json({
|
||||
"type": "list_update",
|
||||
"chat_id": chat_id,
|
||||
"message": payload
|
||||
})
|
||||
|
||||
async def chat_message_edited(self, event):
|
||||
"""Сообщение отредактировано"""
|
||||
chat_id = event.get("chat_id")
|
||||
payload = event.get("payload", {})
|
||||
|
||||
# Для активного чата - обновляем сообщение
|
||||
if chat_id == self.active_chat_id:
|
||||
await self.send_json({
|
||||
"type": "message_updated",
|
||||
"chat_id": chat_id,
|
||||
"message": payload
|
||||
})
|
||||
|
||||
# Для списка - обновляем если это последнее сообщение
|
||||
await self.send_json({
|
||||
"type": "message_edited",
|
||||
"chat_id": chat_id,
|
||||
"message": payload
|
||||
})
|
||||
|
||||
async def chat_message_deleted(self, event):
|
||||
"""Сообщение удалено"""
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
chat_id = event.get("chat_id")
|
||||
message_id = event.get("message_id")
|
||||
|
||||
logger.info(
|
||||
"[USER_WS_DELETE] Received event: chat_id=%s, message_id=%s",
|
||||
chat_id, message_id
|
||||
)
|
||||
logger.info(
|
||||
"[USER_WS_DELETE] Active chat: %s, matches=%s",
|
||||
self.active_chat_id, chat_id == self.active_chat_id
|
||||
)
|
||||
|
||||
if chat_id == self.active_chat_id:
|
||||
logger.info(
|
||||
"[USER_WS_DELETE] Sending to client: message_id=%s",
|
||||
message_id
|
||||
)
|
||||
await self.send_json({
|
||||
"type": "message_deleted",
|
||||
"chat_id": chat_id,
|
||||
"message_id": message_id
|
||||
})
|
||||
logger.info(
|
||||
"[USER_WS_DELETE] Sent to client successfully"
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
"[USER_WS_DELETE] Skipping - not active chat"
|
||||
)
|
||||
|
||||
async def chat_reaction_added(self, event):
|
||||
"""Реакция добавлена"""
|
||||
chat_id = event.get("chat_id")
|
||||
|
||||
if chat_id == self.active_chat_id:
|
||||
await self.send_json({
|
||||
"type": "reaction_added",
|
||||
"chat_id": chat_id,
|
||||
"message_id": event.get("message_id"),
|
||||
"emoji": event.get("emoji"),
|
||||
"user_id": event.get("user_id"),
|
||||
"user_name": event.get("user_name"),
|
||||
"reactions_summary": event.get("reactions_summary")
|
||||
})
|
||||
|
||||
async def chat_reaction_removed(self, event):
|
||||
"""Реакция удалена"""
|
||||
chat_id = event.get("chat_id")
|
||||
|
||||
if chat_id == self.active_chat_id:
|
||||
await self.send_json({
|
||||
"type": "reaction_removed",
|
||||
"chat_id": chat_id,
|
||||
"message_id": event.get("message_id"),
|
||||
"emoji": event.get("emoji"),
|
||||
"user_id": event.get("user_id"),
|
||||
"reactions_summary": event.get("reactions_summary")
|
||||
})
|
||||
|
||||
async def chat_user_typing(self, event):
|
||||
"""Пользователь печатает"""
|
||||
chat_id = event.get("chat_id")
|
||||
user_id = event.get("user_id")
|
||||
|
||||
# Показываем только в активном чате и не показываем свой статус
|
||||
if chat_id == self.active_chat_id and user_id != self.user.id:
|
||||
await self.send_json({
|
||||
"type": "typing_start",
|
||||
"chat_id": chat_id,
|
||||
"user_id": user_id,
|
||||
"user_name": event.get("user_name")
|
||||
})
|
||||
|
||||
async def chat_user_stopped_typing(self, event):
|
||||
"""Пользователь перестал печатать"""
|
||||
chat_id = event.get("chat_id")
|
||||
user_id = event.get("user_id")
|
||||
|
||||
if chat_id == self.active_chat_id and user_id != self.user.id:
|
||||
await self.send_json({
|
||||
"type": "typing_stop",
|
||||
"chat_id": chat_id,
|
||||
"user_id": user_id
|
||||
})
|
||||
|
||||
async def chat_poll_update(self, event):
|
||||
"""Обновление голосования"""
|
||||
chat_id = event.get("chat_id")
|
||||
payload = event.get("payload", {})
|
||||
|
||||
if chat_id == self.active_chat_id:
|
||||
await self.send_json({
|
||||
"type": "poll_update",
|
||||
"chat_id": chat_id,
|
||||
"poll_id": payload.get("poll_id"),
|
||||
"message_id": payload.get("message_id"),
|
||||
"results": payload.get("results")
|
||||
})
|
||||
|
||||
# ==================== Обработка действий пользователя ====================
|
||||
|
||||
async def _handle_send_message(self, content):
|
||||
"""Отправка нового сообщения"""
|
||||
if not self.active_chat_id:
|
||||
await self.send_json({
|
||||
"type": "error",
|
||||
"error": "No active chat"
|
||||
})
|
||||
return
|
||||
|
||||
text = str(content.get("content") or content.get("text") or "").strip()
|
||||
if not text:
|
||||
return
|
||||
|
||||
chat = await self._get_chat(self.active_chat_id)
|
||||
if not chat:
|
||||
return
|
||||
|
||||
# Проверка прав на отправку
|
||||
if chat.type == "announcement" and chat.created_by_id != self.user.id:
|
||||
await self.send_json({
|
||||
"type": "error",
|
||||
"error": "Только автор может публиковать в это объявление"
|
||||
})
|
||||
return
|
||||
|
||||
# Создаем сообщение
|
||||
msg = await self._create_message(chat, self.user, text)
|
||||
|
||||
if msg:
|
||||
# Сериализуем и отправляем всем в группе
|
||||
msg_data = await database_sync_to_async(serialize_message)(msg)
|
||||
|
||||
await self.channel_layer.group_send(
|
||||
f"chat_{chat.id}",
|
||||
{
|
||||
"type": "chat_message",
|
||||
"chat_id": chat.id,
|
||||
"payload": msg_data
|
||||
}
|
||||
)
|
||||
|
||||
async def _handle_edit_message(self, content):
|
||||
"""Редактирование сообщения"""
|
||||
if not self.active_chat_id:
|
||||
return
|
||||
|
||||
message_id = content.get("message_id")
|
||||
new_content = content.get("content", "").strip()
|
||||
|
||||
if not message_id or not new_content:
|
||||
return
|
||||
|
||||
msg = await self._get_message(message_id)
|
||||
if not msg or msg.author_id != self.user.id:
|
||||
return
|
||||
|
||||
# Обновляем сообщение
|
||||
await self._update_message(msg, new_content)
|
||||
|
||||
# Сериализуем и отправляем
|
||||
msg_data = await database_sync_to_async(serialize_message)(msg)
|
||||
|
||||
await self.channel_layer.group_send(
|
||||
f"chat_{msg.chat_id}",
|
||||
{
|
||||
"type": "chat_message_edited",
|
||||
"chat_id": msg.chat_id,
|
||||
"payload": msg_data
|
||||
}
|
||||
)
|
||||
|
||||
async def _handle_delete_message(self, content):
|
||||
"""Удаление сообщения"""
|
||||
if not self.active_chat_id:
|
||||
return
|
||||
|
||||
message_id = content.get("message_id")
|
||||
if not message_id:
|
||||
return
|
||||
|
||||
msg = await self._get_message(message_id)
|
||||
if not msg or msg.author_id != self.user.id:
|
||||
return
|
||||
|
||||
chat_id = msg.chat_id
|
||||
|
||||
# Помечаем как удаленное
|
||||
await self._soft_delete_message(msg)
|
||||
|
||||
await self.channel_layer.group_send(
|
||||
f"chat_{chat_id}",
|
||||
{
|
||||
"type": "chat_message_deleted",
|
||||
"chat_id": chat_id,
|
||||
"message_id": message_id
|
||||
}
|
||||
)
|
||||
|
||||
async def _handle_add_reaction(self, content):
|
||||
"""Добавление реакции"""
|
||||
message_id = content.get("message_id")
|
||||
emoji = content.get("emoji")
|
||||
|
||||
if not message_id or not emoji:
|
||||
return
|
||||
|
||||
msg = await self._get_message(message_id)
|
||||
if not msg:
|
||||
return
|
||||
|
||||
# Создаем реакцию
|
||||
created = await self._add_reaction(msg, self.user, emoji)
|
||||
|
||||
if created:
|
||||
user_name = await database_sync_to_async(
|
||||
lambda: self.user.get_full_name() or self.user.username
|
||||
)()
|
||||
|
||||
await self.channel_layer.group_send(
|
||||
f"chat_{msg.chat_id}",
|
||||
{
|
||||
"type": "chat_reaction_added",
|
||||
"chat_id": msg.chat_id,
|
||||
"message_id": message_id,
|
||||
"emoji": emoji,
|
||||
"user_id": self.user.id,
|
||||
"user_name": user_name
|
||||
}
|
||||
)
|
||||
|
||||
async def _handle_remove_reaction(self, content):
|
||||
"""Удаление реакции"""
|
||||
message_id = content.get("message_id")
|
||||
emoji = content.get("emoji")
|
||||
|
||||
if not message_id or not emoji:
|
||||
return
|
||||
|
||||
msg = await self._get_message(message_id)
|
||||
if not msg:
|
||||
return
|
||||
|
||||
# Удаляем реакцию
|
||||
deleted = await self._remove_reaction(msg, self.user, emoji)
|
||||
|
||||
if deleted:
|
||||
await self.channel_layer.group_send(
|
||||
f"chat_{msg.chat_id}",
|
||||
{
|
||||
"type": "chat_reaction_removed",
|
||||
"chat_id": msg.chat_id,
|
||||
"message_id": message_id,
|
||||
"emoji": emoji,
|
||||
"user_id": self.user.id
|
||||
}
|
||||
)
|
||||
|
||||
async def _handle_typing(self, content):
|
||||
"""Индикатор печати"""
|
||||
if not self.active_chat_id:
|
||||
return
|
||||
|
||||
chat = await self._get_chat(self.active_chat_id)
|
||||
if not chat:
|
||||
return
|
||||
|
||||
await self._set_typing_status(chat, self.user, True)
|
||||
|
||||
user_name = await database_sync_to_async(
|
||||
lambda: self.user.get_full_name() or self.user.username
|
||||
)()
|
||||
|
||||
await self.channel_layer.group_send(
|
||||
f"chat_{chat.id}",
|
||||
{
|
||||
"type": "chat_user_typing",
|
||||
"chat_id": chat.id,
|
||||
"user_id": self.user.id,
|
||||
"user_name": user_name
|
||||
}
|
||||
)
|
||||
|
||||
async def _handle_stop_typing(self):
|
||||
"""Остановка индикатора печати"""
|
||||
if not self.active_chat_id:
|
||||
return
|
||||
|
||||
chat = await self._get_chat(self.active_chat_id)
|
||||
if not chat:
|
||||
return
|
||||
|
||||
await self._set_typing_status(chat, self.user, False)
|
||||
|
||||
await self.channel_layer.group_send(
|
||||
f"chat_{chat.id}",
|
||||
{
|
||||
"type": "chat_user_stopped_typing",
|
||||
"chat_id": chat.id,
|
||||
"user_id": self.user.id
|
||||
}
|
||||
)
|
||||
|
||||
async def _handle_mark_read(self, content):
|
||||
"""Отметка сообщений как прочитанных"""
|
||||
chat_id = content.get("chat_id")
|
||||
if not chat_id:
|
||||
return
|
||||
|
||||
chat = await self._get_chat(int(chat_id))
|
||||
if not chat:
|
||||
return
|
||||
|
||||
await self._mark_read(chat, self.user)
|
||||
|
||||
async def _handle_vote_poll(self, content):
|
||||
"""Голосование в опросе"""
|
||||
# TODO: Реализовать логику голосования
|
||||
pass
|
||||
|
||||
# ==================== Вспомогательные методы ====================
|
||||
|
||||
async def _send_initial_messages(self, chat_id, limit=50):
|
||||
"""Отправка начальной истории сообщений"""
|
||||
messages = await self._get_recent_messages(chat_id, limit)
|
||||
|
||||
await self.send_json({
|
||||
"type": "initial_messages",
|
||||
"chat_id": chat_id,
|
||||
"messages": messages
|
||||
})
|
||||
|
||||
@database_sync_to_async
|
||||
def _get_available_chat_ids(self, user):
|
||||
"""Получить ID всех доступных чатов пользователя"""
|
||||
departments = user.departments.all()
|
||||
|
||||
# Получаем ID через membership
|
||||
membership_chat_ids = list(
|
||||
ChatMembership.objects.filter(user=user).values_list('chat_id', flat=True)
|
||||
)
|
||||
|
||||
chat_ids = list(
|
||||
Chat.objects.filter(
|
||||
Q(type="global")
|
||||
| Q(type="department", department__in=departments)
|
||||
| Q(type="private", participants=user)
|
||||
| Q(id__in=membership_chat_ids)
|
||||
).values_list("id", flat=True).distinct()
|
||||
)
|
||||
|
||||
return chat_ids
|
||||
|
||||
@database_sync_to_async
|
||||
def _get_chat(self, chat_id):
|
||||
"""Получить чат по ID"""
|
||||
try:
|
||||
return Chat.objects.get(pk=chat_id)
|
||||
except Chat.DoesNotExist:
|
||||
return None
|
||||
|
||||
@database_sync_to_async
|
||||
def _get_message(self, message_id):
|
||||
"""Получить сообщение по ID"""
|
||||
try:
|
||||
return Message.objects.select_related('author', 'chat').get(pk=message_id)
|
||||
except Message.DoesNotExist:
|
||||
return None
|
||||
|
||||
@database_sync_to_async
|
||||
def _user_can_access(self, chat, user):
|
||||
"""Проверка доступа пользователя к чату"""
|
||||
if chat.type == "global":
|
||||
return True
|
||||
|
||||
if chat.type == "private":
|
||||
return chat.participants.filter(pk=user.pk).exists()
|
||||
|
||||
if chat.type == "group":
|
||||
return (
|
||||
chat.participants.filter(pk=user.pk).exists()
|
||||
or ChatMembership.objects.filter(chat=chat, user=user).exists()
|
||||
)
|
||||
|
||||
if chat.type == "department" and chat.department_id:
|
||||
return chat.get_participants.filter(pk=user.pk).exists()
|
||||
|
||||
if chat.type in ("channel", "announcement"):
|
||||
if chat.include_all_employees:
|
||||
return user.is_active
|
||||
return (
|
||||
chat.participants.filter(pk=user.pk).exists()
|
||||
or ChatMembership.objects.filter(chat=chat, user=user).exists()
|
||||
)
|
||||
|
||||
return False
|
||||
|
||||
@database_sync_to_async
|
||||
def _create_message(self, chat, user, content):
|
||||
"""Создать новое сообщение"""
|
||||
try:
|
||||
msg = Message.objects.create(
|
||||
chat=chat,
|
||||
author=user,
|
||||
content=content
|
||||
)
|
||||
return msg
|
||||
except Exception as e:
|
||||
print(f"Error creating message: {e}")
|
||||
return None
|
||||
|
||||
@database_sync_to_async
|
||||
def _update_message(self, msg, new_content):
|
||||
"""Обновить содержимое сообщения"""
|
||||
msg.content = new_content
|
||||
msg.is_edited = True
|
||||
msg.edited_at = timezone.now()
|
||||
msg.save(update_fields=['content', 'is_edited', 'edited_at'])
|
||||
|
||||
@database_sync_to_async
|
||||
def _soft_delete_message(self, msg):
|
||||
"""Мягкое удаление сообщения"""
|
||||
msg.is_deleted = True
|
||||
msg.content = "[Сообщение удалено]"
|
||||
msg.save(update_fields=['is_deleted', 'content'])
|
||||
|
||||
@database_sync_to_async
|
||||
def _add_reaction(self, msg, user, emoji):
|
||||
"""Добавить реакцию"""
|
||||
reaction, created = MessageReaction.objects.get_or_create(
|
||||
message=msg,
|
||||
user=user,
|
||||
emoji=emoji
|
||||
)
|
||||
return created
|
||||
|
||||
@database_sync_to_async
|
||||
def _remove_reaction(self, msg, user, emoji):
|
||||
"""Удалить реакцию"""
|
||||
deleted_count, _ = MessageReaction.objects.filter(
|
||||
message=msg,
|
||||
user=user,
|
||||
emoji=emoji
|
||||
).delete()
|
||||
return deleted_count > 0
|
||||
|
||||
@database_sync_to_async
|
||||
def _mark_read(self, chat, user):
|
||||
"""Отметить чат как прочитанный"""
|
||||
ChatReadState.objects.update_or_create(
|
||||
chat=chat,
|
||||
user=user,
|
||||
defaults={"last_read_at": timezone.now()}
|
||||
)
|
||||
|
||||
@database_sync_to_async
|
||||
def _set_typing_status(self, chat, user, is_typing):
|
||||
"""Установить статус печати (можно кэшировать в Redis)"""
|
||||
# TODO: Реализовать через Redis для производительности
|
||||
pass
|
||||
|
||||
@database_sync_to_async
|
||||
def _get_recent_messages(self, chat_id, limit=50):
|
||||
"""Получить последние сообщения чата"""
|
||||
messages = Message.objects.filter(
|
||||
chat_id=chat_id
|
||||
).select_related(
|
||||
'author'
|
||||
).prefetch_related(
|
||||
'attachments', 'reactions'
|
||||
).order_by('-created_at')[:limit]
|
||||
|
||||
# Реверсируем порядок (старые -> новые)
|
||||
messages = list(reversed(messages))
|
||||
|
||||
return [serialize_message(msg) for msg in messages]
|
||||
@ -1,151 +0,0 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from channels.generic.websocket import AsyncWebsocketConsumer
|
||||
from channels.db import database_sync_to_async
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.utils import timezone
|
||||
|
||||
User = get_user_model()
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class NotificationConsumer(AsyncWebsocketConsumer):
|
||||
"""WebSocket consumer для уведомлений в реальном времени"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.ping_task = None
|
||||
self.ping_interval = 20 # Ping каждые 20 секунд
|
||||
|
||||
async def connect(self):
|
||||
"""Подключение к WebSocket"""
|
||||
self.user = self.scope['user']
|
||||
|
||||
if self.user.is_anonymous:
|
||||
await self.close()
|
||||
return
|
||||
|
||||
# Добавить в группу уведомлений пользователя
|
||||
self.notification_group_name = f'notifications_{self.user.id}'
|
||||
await self.channel_layer.group_add(
|
||||
self.notification_group_name,
|
||||
self.channel_name
|
||||
)
|
||||
|
||||
await self.accept()
|
||||
|
||||
logger.info(f'NotificationConsumer connected: user={self.user.id}, group={self.notification_group_name}')
|
||||
|
||||
# Отправить текущее количество непрочитанных
|
||||
unread_count = await self.get_unread_count()
|
||||
await self.send(text_data=json.dumps({
|
||||
'type': 'unread_count',
|
||||
'count': unread_count
|
||||
}))
|
||||
|
||||
# Запускаем ping цикл для keepalive
|
||||
self.ping_task = asyncio.create_task(self._ping_loop())
|
||||
|
||||
async def disconnect(self, close_code):
|
||||
"""Отключение от WebSocket"""
|
||||
# Останавливаем ping цикл
|
||||
if self.ping_task:
|
||||
self.ping_task.cancel()
|
||||
try:
|
||||
await self.ping_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
if hasattr(self, 'notification_group_name'):
|
||||
await self.channel_layer.group_discard(
|
||||
self.notification_group_name,
|
||||
self.channel_name
|
||||
)
|
||||
|
||||
async def _ping_loop(self):
|
||||
"""Отправка ping каждые N секунд для keepalive"""
|
||||
try:
|
||||
while True:
|
||||
await asyncio.sleep(self.ping_interval)
|
||||
await self.send(text_data=json.dumps({
|
||||
'type': 'ping',
|
||||
'timestamp': timezone.now().isoformat()
|
||||
}))
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.error(f"Ping loop error: {e}")
|
||||
|
||||
async def receive(self, text_data):
|
||||
"""Получение сообщения от клиента"""
|
||||
try:
|
||||
data = json.loads(text_data)
|
||||
message_type = data.get('type')
|
||||
|
||||
if message_type == 'mark_read':
|
||||
notification_id = data.get('notification_id')
|
||||
if notification_id:
|
||||
await self.mark_notification_as_read(notification_id)
|
||||
|
||||
elif message_type == 'mark_all_read':
|
||||
await self.mark_all_notifications_as_read()
|
||||
|
||||
except json.JSONDecodeError:
|
||||
await self.send(text_data=json.dumps({
|
||||
'type': 'error',
|
||||
'message': 'Invalid JSON'
|
||||
}))
|
||||
|
||||
async def notification_new(self, event):
|
||||
"""Отправить новое уведомление клиенту"""
|
||||
logger.info(f'Sending new notification to user {self.user.id}: {event.get("notification", {}).get("title", "No title")}')
|
||||
await self.send(text_data=json.dumps({
|
||||
'type': 'new_notification',
|
||||
'notification': event['notification']
|
||||
}))
|
||||
|
||||
async def notification_count_update(self, event):
|
||||
"""Обновить счетчик непрочитанных"""
|
||||
await self.send(text_data=json.dumps({
|
||||
'type': 'unread_count',
|
||||
'count': event['count']
|
||||
}))
|
||||
|
||||
@database_sync_to_async
|
||||
def get_unread_count(self):
|
||||
"""Получить количество непрочитанных уведомлений"""
|
||||
from .models import Notification
|
||||
return Notification.objects.filter(
|
||||
recipient=self.user,
|
||||
is_read=False,
|
||||
is_archived=False
|
||||
).count()
|
||||
|
||||
@database_sync_to_async
|
||||
def mark_notification_as_read(self, notification_id):
|
||||
"""Отметить уведомление как прочитанное"""
|
||||
from .models import Notification
|
||||
try:
|
||||
notification = Notification.objects.get(
|
||||
id=notification_id,
|
||||
recipient=self.user
|
||||
)
|
||||
notification.mark_as_read()
|
||||
return True
|
||||
except Notification.DoesNotExist:
|
||||
return False
|
||||
|
||||
@database_sync_to_async
|
||||
def mark_all_notifications_as_read(self):
|
||||
"""Отметить все уведомления как прочитанные"""
|
||||
from .models import Notification
|
||||
from django.utils import timezone
|
||||
|
||||
Notification.objects.filter(
|
||||
recipient=self.user,
|
||||
is_read=False
|
||||
).update(
|
||||
is_read=True,
|
||||
read_at=timezone.now()
|
||||
)
|
||||
@ -1,6 +0,0 @@
|
||||
from django.urls import re_path
|
||||
from . import consumers
|
||||
|
||||
websocket_urlpatterns = [
|
||||
re_path(r'^ws/notifications/$', consumers.NotificationConsumer.as_asgi()),
|
||||
]
|
||||
Loading…
x
Reference in New Issue
Block a user