diff --git a/backend/CELERY_NEXT_STEPS.md b/backend/CELERY_NEXT_STEPS.md new file mode 100644 index 00000000..ae799d96 --- /dev/null +++ b/backend/CELERY_NEXT_STEPS.md @@ -0,0 +1,201 @@ +# Celery Integration - Следующие шаги + +## ✅ Что уже сделано + +1. ✅ Добавлены зависимости в `requirements.txt`: + - `celery==5.5.3` + - `redis==5.2.1` + - `django-celery-beat==2.7.0` + - `django-celery-results==2.5.1` + - `flower==2.0.1` + +2. ✅ Создан файл конфигурации `eusrr_backend/celery.py` + - Настроена автоматическая регистрация задач + - Добавлен debug task для тестирования + +3. ✅ Обновлен `eusrr_backend/__init__.py` + - Автоматический импорт celery_app при запуске Django + +4. ✅ Добавлены настройки Celery в `settings.py`: + - `CELERY_BROKER_URL` - Redis как брокер сообщений + - `CELERY_RESULT_BACKEND` - Redis для хранения результатов + - Настройка роутинга задач по очередям + - Таймауты и лимиты для workers + +5. ✅ Добавлены приложения в `INSTALLED_APPS`: + - `django_celery_beat` - для периодических задач + - `django_celery_results` - для хранения результатов + +6. ✅ Созданы Celery задачи в `notifications/tasks.py`: + - `send_notification_task` - отправка одного уведомления + - `send_bulk_notifications_task` - массовая отправка + - `cleanup_old_notifications_task` - очистка старых уведомлений + - `send_digest_email_task` - email-дайджесты + +## 🔄 Следующие шаги (выполнить вручную) + +### Шаг 1: Установка зависимостей + +Запустите задачу в VS Code: +- **Pip: install requirements** + +Или вручную: +```bash +cd backend +..\.venv\Scripts\pip install -r requirements.txt +``` + +### Шаг 2: Установка и запуск Redis + +#### Вариант 1: Windows (рекомендуется Memurai) +1. Скачайте Memurai (Redis для Windows): https://www.memurai.com/ +2. Установите и запустите как службу Windows +3. Redis будет доступен на `localhost:6379` + +#### Вариант 2: Docker +```bash +docker run -d -p 6379:6379 --name redis redis:7-alpine +``` + +#### Проверка Redis: +```bash +redis-cli ping +# Должно вернуть: PONG +``` + +### Шаг 3: Применение миграций + +Запустите: +```bash +cd backend +migrate_celery.bat +``` + +Или вручную: +```bash +..\.venv\Scripts\python manage.py migrate django_celery_beat +..\.venv\Scripts\python manage.py migrate django_celery_results +``` + +### Шаг 4: Тестирование Celery Worker + +В отдельном терминале запустите worker: +```bash +cd backend +..\.venv\Scripts\celery -A eusrr_backend worker -l info --pool=solo +``` + +**Примечание**: На Windows используйте `--pool=solo` или `--pool=gevent` + +### Шаг 5: Тестирование отправки задачи + +В Django shell: +```python +..\.venv\Scripts\python manage.py shell + +from celery import current_app +current_app.send_task('eusrr_backend.debug_task') +``` + +Вы должны увидеть выполнение задачи в логах worker. + +### Шаг 6: Интеграция с NotificationService + +Обновите `notifications/services.py`: +```python +from notifications.tasks import send_notification_task, send_bulk_notifications_task + +class NotificationService: + @staticmethod + def create_notification(...): + # Вместо синхронного создания + send_notification_task.delay( + notification_type=notification_type, + user_id=user.id, + title=title, + message=message, + link=link, + sender_id=sender.id if sender else None, + metadata=metadata, + ) +``` + +### Шаг 7: Запуск Flower (мониторинг) + +```bash +cd backend +..\.venv\Scripts\celery -A eusrr_backend flower +``` + +Откройте http://localhost:5555 для просмотра статистики задач. + +### Шаг 8: Настройка периодических задач + +В Django Admin: +1. Перейдите в "Periodic Tasks" +2. Добавьте задачу для `notifications.cleanup_old_notifications` +3. Настройте интервал (например, раз в день) + +Или программно в `eusrr_backend/celery.py`: +```python +from celery.schedules import crontab + +app.conf.beat_schedule = { + 'cleanup-old-notifications': { + 'task': 'notifications.cleanup_old_notifications', + 'schedule': crontab(hour=3, minute=0), # Каждый день в 3:00 + 'args': (90,) # Удалять уведомления старше 90 дней + }, +} +``` + +Запустите beat scheduler: +```bash +..\.venv\Scripts\celery -A eusrr_backend beat -l info +``` + +## 📊 Ожидаемый результат + +**До внедрения Celery:** +- Отправка сообщения в группу с 8 участниками: ~500ms +- Блокировка API response до отправки всех уведомлений + +**После внедрения Celery:** +- Отправка сообщения: <50ms (только создание задачи) +- Уведомления обрабатываются асинхронно +- API мгновенно отвечает пользователю + +## 🔍 Отладка + +### Проверка статуса Celery +```bash +..\.venv\Scripts\celery -A eusrr_backend inspect active +..\.venv\Scripts\celery -A eusrr_backend inspect stats +``` + +### Проверка задач в очереди +```bash +..\.venv\Scripts\celery -A eusrr_backend inspect reserved +``` + +### Логи +- Worker логи: консоль где запущен worker +- Flower: http://localhost:5555 +- Django логи: стандартные логи приложения + +## 📚 Документация + +- Подробная документация: `docs/guides/CELERY_SETUP.md` +- План внедрения (5 дней): `docs/in-progress/CELERY_INTEGRATION_PLAN.md` +- Celery docs: https://docs.celeryq.dev/ + +## ⚠️ Production Deployment + +Для production: +1. Используйте Redis Cluster или Redis Sentinel +2. Запускайте несколько workers с supervisord/systemd +3. Настройте мониторинг (Flower + Prometheus) +4. Добавьте retry policies и dead letter queue +5. Настройте автоматический перезапуск workers + +См. `docs/in-progress/CELERY_INTEGRATION_PLAN.md` для деталей. diff --git a/backend/calendar_app/notification_signals.py b/backend/calendar_app/notification_signals.py index 8e5ac88e..208d2fe3 100644 --- a/backend/calendar_app/notification_signals.py +++ b/backend/calendar_app/notification_signals.py @@ -78,7 +78,7 @@ def notify_event_cancelled(sender, instance, **kwargs): # Создаем уведомления об отмене for recipient in recipients: - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=recipient, notification_type_code='event_cancelled', title='Событие отменено', @@ -117,7 +117,7 @@ def notify_event_created(event): ) for recipient in recipients: - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=recipient, notification_type_code='event_created', title='Новое событие в календаре', @@ -166,7 +166,7 @@ def notify_event_changed(event, changed_fields): ]) for recipient in recipients: - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=recipient, notification_type_code='event_changed', title='Событие изменено', diff --git a/backend/communications/notification_signals.py b/backend/communications/notification_signals.py index e3e309e0..2064fae6 100644 --- a/backend/communications/notification_signals.py +++ b/backend/communications/notification_signals.py @@ -111,8 +111,8 @@ def create_message_notifications(sender, instance, created, **kwargs): if user in participants_list and user.id != author.id: # Проверяем настройки уведомлений (из предзагруженного словаря) if notification_settings.get(user.id, True): - # Создаем уведомление об упоминании - NotificationService.create_notification( + # Создаем уведомление об упоминании (асинхронно) + NotificationService.create_notification_async( recipient=user, notification_type_code='chat_mention', title=f'Вас упомянул {author.get_full_name() or author.username}', @@ -138,7 +138,7 @@ def create_message_notifications(sender, instance, created, **kwargs): if original_author.id not in mentioned_user_ids: # Проверяем настройки уведомлений (из предзагруженного словаря) if notification_settings.get(original_author.id, True): - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=original_author, notification_type_code='chat_reply', title=f'{author.get_full_name() or author.username} ответил на ваше сообщение', @@ -197,7 +197,7 @@ def create_message_notifications(sender, instance, created, **kwargs): if is_announcement: metadata['is_announcement'] = True - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=recipient, notification_type_code=notification_type, title=title, @@ -238,7 +238,7 @@ def create_chat_added_notifications(sender, instance, action, pk_set, **kwargs): if not notification_settings.get(user.id, True): continue - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=user, notification_type_code='chat_added_to_chat', title=f'Вас добавили в чат', diff --git a/backend/documents/notification_signals.py b/backend/documents/notification_signals.py index e4456448..b763a5ea 100644 --- a/backend/documents/notification_signals.py +++ b/backend/documents/notification_signals.py @@ -333,7 +333,7 @@ def check_all_acknowledged(sender, instance, created, **kwargs): total_recipients > 0 and document.uploaded_by): - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=document.uploaded_by, notification_type_code='document_signed_all', title='Все ознакомились с документом', @@ -452,7 +452,7 @@ def create_document_ready_notification(document, recipient, send_immediately=Tru else 'Администратор' ) - notification = NotificationService.create_notification( + notification = NotificationService.create_notification_async( recipient=recipient, notification_type_code='document_ready', title='Новый документ на ознакомление', diff --git a/backend/eusrr_backend/__init__.py b/backend/eusrr_backend/__init__.py index e69de29b..eeabe4f0 100644 --- a/backend/eusrr_backend/__init__.py +++ b/backend/eusrr_backend/__init__.py @@ -0,0 +1,7 @@ +""" +Инициализация проекта EUSRR +""" +# Это гарантирует, что приложение Celery загружается при старте Django +from .celery import app as celery_app + +__all__ = ('celery_app',) diff --git a/backend/eusrr_backend/celery.py b/backend/eusrr_backend/celery.py new file mode 100644 index 00000000..a47f8c4c --- /dev/null +++ b/backend/eusrr_backend/celery.py @@ -0,0 +1,24 @@ +""" +Конфигурация Celery для проекта EUSRR +""" +import os +from celery import Celery + +# Устанавливаем настройки Django по умолчанию +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings') + +app = Celery('eusrr_backend') + +# Загружаем конфигурацию из Django settings +# namespace='CELERY' означает что все настройки Celery в settings.py +# должны начинаться с префикса CELERY_ +app.config_from_object('django.conf:settings', namespace='CELERY') + +# Автоматически находит tasks.py в каждом приложении +app.autodiscover_tasks() + + +@app.task(bind=True, ignore_result=True) +def debug_task(self): + """Тестовая задача для проверки работоспособности Celery""" + print(f'Request: {self.request!r}') diff --git a/backend/eusrr_backend/settings.py b/backend/eusrr_backend/settings.py index 1f3de66f..d25df8b6 100644 --- a/backend/eusrr_backend/settings.py +++ b/backend/eusrr_backend/settings.py @@ -35,6 +35,10 @@ INSTALLED_APPS = [ "widget_tweaks", "simple_history", "rest_framework", + # Celery приложения + "django_celery_beat", # Периодические задачи + "django_celery_results", # Хранение результатов + # Основные приложения "employees.apps.EmployeesConfig", "api.apps.ApiConfig", "hikcentral.apps.HikcentralConfig", @@ -460,4 +464,39 @@ VAPID_PRIVATE_KEY = os.getenv( "VAPID_PRIVATE_KEY", "MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgCNnpupg3xbtFUiOSUZ6L7s6puxuEjzR73kTL7v8bMvKhRANCAATE4rWcva-MoGCQYBnWmbgpG-68Ee4hyyb9A5DaWiTixEqLTjRE1rWQEMZyVVP2KPb_omKPu3VhHhdFRIzHBjJE" ) -VAPID_ADMIN_EMAIL = os.getenv("VAPID_ADMIN_EMAIL", "robotail-info@yandex.ru") \ No newline at end of file +VAPID_ADMIN_EMAIL = os.getenv("VAPID_ADMIN_EMAIL", "robotail-info@yandex.ru") + +# ----------------------------------------------------------------------------- +# CELERY CONFIGURATION +# ----------------------------------------------------------------------------- +CELERY_BROKER_URL = os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0') +CELERY_RESULT_BACKEND = os.getenv('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1') + +# Сериализация +CELERY_ACCEPT_CONTENT = ['json'] +CELERY_TASK_SERIALIZER = 'json' +CELERY_RESULT_SERIALIZER = 'json' + +# Таймзона +CELERY_TIMEZONE = 'Europe/Moscow' +CELERY_ENABLE_UTC = True + +# Мониторинг и отладка +CELERY_TASK_TRACK_STARTED = True +CELERY_TASK_TIME_LIMIT = 30 * 60 # 30 минут максимум на задачу +CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60 # Мягкий лимит - 25 минут + +# Приоритеты очередей +CELERY_TASK_ROUTES = { + 'notifications.tasks.*': {'queue': 'notifications'}, + 'documents.tasks.*': {'queue': 'default'}, + 'employees.tasks.*': {'queue': 'default'}, +} + +# Настройки для периодических задач (Celery Beat) +CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler' + +# Опции по умолчанию для всех задач +CELERY_TASK_ACKS_LATE = True # Подтверждаем выполнение после завершения +CELERY_WORKER_PREFETCH_MULTIPLIER = 1 # Берем по 1 задаче за раз +CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000 # Перезапуск worker после 1000 задач \ No newline at end of file diff --git a/backend/feed/notification_signals.py b/backend/feed/notification_signals.py index 1bf202db..9279cc10 100644 --- a/backend/feed/notification_signals.py +++ b/backend/feed/notification_signals.py @@ -49,7 +49,7 @@ def create_comment_notification(sender, instance, created, **kwargs): # Уведомляем автора публикации (если комментарий не от него) if post.author.id != comment_author.id: - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=post.author, notification_type_code='feed_post_comment', title='Новый комментарий к вашей публикации', @@ -90,7 +90,7 @@ def notify_new_post(post): ) for recipient in recipients: - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=recipient, notification_type_code='feed_new_post', title=f'Новая публикация {context}', @@ -163,7 +163,7 @@ def notify_post_reaction(post, user): """ # Уведомляем автора (если лайк не от него) if post.author.id != user.id: - NotificationService.create_notification( + NotificationService.create_notification_async( recipient=post.author, notification_type_code='feed_post_reaction', title='Ваша публикация понравилась', diff --git a/backend/notifications/services.py b/backend/notifications/services.py index bc61d6db..4bbb6c8b 100644 --- a/backend/notifications/services.py +++ b/backend/notifications/services.py @@ -21,9 +21,104 @@ logger = logging.getLogger(__name__) User = get_user_model() +def is_celery_available(): + """Проверка доступности Celery worker""" + try: + from celery import current_app + inspect = current_app.control.inspect() + active_workers = inspect.active() + return active_workers is not None and len(active_workers) > 0 + except Exception: + return False + + class NotificationService: """Сервис для создания и отправки уведомлений""" + @staticmethod + def create_notification_async( + recipient: User, + notification_type_code: str, + title: str, + message: str, + content_object=None, + action_url: str = '', + action_text: str = 'Посмотреть', + metadata: Optional[Dict[str, Any]] = None, + ) -> bool: + """ + Создать уведомление асинхронно через Celery (рекомендуется) + + Args: + recipient: Получатель уведомления + notification_type_code: Код типа уведомления + title: Заголовок + message: Текст сообщения + content_object: Связанный объект (опционально) + action_url: URL для действия + action_text: Текст кнопки действия + metadata: Дополнительные данные + + Returns: + True если задача поставлена в очередь, False если fallback на sync + """ + try: + # Проверяем доступность Celery + if not is_celery_available(): + logger.warning( + f"[create_notification_async] Celery недоступен, fallback на синхронный режим" + ) + NotificationService.create_notification( + recipient=recipient, + notification_type_code=notification_type_code, + title=title, + message=message, + content_object=content_object, + action_url=action_url, + action_text=action_text, + metadata=metadata, + send_immediately=True, + ) + return False + + # Импортируем задачу + from notifications.tasks import send_notification_task + + # Отправляем задачу в Celery + send_notification_task.delay( + notification_type=notification_type_code, + user_id=recipient.id, + title=title, + message=message, + link=action_url, + sender_id=None, + metadata=metadata or {}, + ) + + logger.debug( + f"[create_notification_async] ✅ Задача отправлена в Celery: " + f"user={recipient.id}, type={notification_type_code}" + ) + return True + + except Exception as e: + logger.exception( + f"[create_notification_async] ❌ Ошибка при отправке в Celery: {e}" + ) + # Fallback на синхронный режим + NotificationService.create_notification( + recipient=recipient, + notification_type_code=notification_type_code, + title=title, + message=message, + content_object=content_object, + action_url=action_url, + action_text=action_text, + metadata=metadata, + send_immediately=True, + ) + return False + @staticmethod def create_notification( recipient: User, diff --git a/backend/notifications/tasks.py b/backend/notifications/tasks.py new file mode 100644 index 00000000..21c39b0f --- /dev/null +++ b/backend/notifications/tasks.py @@ -0,0 +1,259 @@ +""" +Celery задачи для асинхронной обработки уведомлений. + +Эти задачи выполняются в фоновом режиме через Celery workers, +что позволяет значительно ускорить отклик API при отправке уведомлений. +""" +from celery import shared_task +from django.contrib.auth import get_user_model +from django.utils import timezone +from typing import List, Optional +import logging + +logger = logging.getLogger(__name__) +User = get_user_model() + + +@shared_task( + name="notifications.send_notification", + bind=True, + max_retries=3, + default_retry_delay=60, +) +def send_notification_task( + self, + notification_type: str, + user_id: int, + title: str, + message: str, + link: Optional[str] = None, + sender_id: Optional[int] = None, + metadata: Optional[dict] = None, +): + """ + Асинхронная отправка уведомления одному пользователю. + + Args: + notification_type: Тип уведомления (message, document, request и т.д.) + user_id: ID пользователя-получателя + title: Заголовок уведомления + message: Текст уведомления + link: Ссылка (опционально) + sender_id: ID отправителя (опционально) + metadata: Дополнительные данные (опционально) + """ + try: + from notifications.services import NotificationService + + user = User.objects.get(id=user_id) + sender = User.objects.get(id=sender_id) if sender_id else None + + # Вызываем синхронный метод создания уведомления + NotificationService.create_notification( + recipient=user, + notification_type_code=notification_type, + title=title, + message=message, + action_url=link or '', + metadata=metadata or {}, + ) + + logger.debug( + f"Notification sent to user {user_id} (type: {notification_type})" + ) + return {"status": "success", "user_id": user_id} + + except User.DoesNotExist: + logger.error(f"User {user_id} not found for notification") + return {"status": "error", "reason": "user_not_found"} + + except Exception as exc: + logger.exception(f"Failed to send notification to user {user_id}: {exc}") + # Повторяем попытку через 60 секунд + raise self.retry(exc=exc) + + +@shared_task( + name="notifications.send_bulk_notifications", + bind=True, + max_retries=3, + default_retry_delay=60, +) +def send_bulk_notifications_task( + self, + notification_type: str, + user_ids: List[int], + title: str, + message: str, + link: Optional[str] = None, + sender_id: Optional[int] = None, + metadata: Optional[dict] = None, +): + """ + Асинхронная массовая отправка уведомлений нескольким пользователям. + + Оптимизирована для отправки большого количества уведомлений. + + Args: + notification_type: Тип уведомления + user_ids: Список ID пользователей-получателей + title: Заголовок уведомления + message: Текст уведомления + link: Ссылка (опционально) + sender_id: ID отправителя (опционально) + metadata: Дополнительные данные (опционально) + """ + try: + from notifications.services import NotificationService + from notifications.models import Notification + + # Получаем всех пользователей одним запросом + users = User.objects.filter(id__in=user_ids) + sender = User.objects.get(id=sender_id) if sender_id else None + + # Создаем уведомления массово + notifications = [] + for user in users: + notification = Notification( + user=user, + notification_type=notification_type, + title=title, + message=message, + link=link or "", + sender=sender, + metadata=metadata or {}, + created_at=timezone.now(), + ) + notifications.append(notification) + + # Bulk create для производительности + Notification.objects.bulk_create(notifications) + + # Отправляем WebSocket уведомления + for user in users: + NotificationService.send_websocket_notification(user.id) + + logger.debug( + f"Bulk notification sent to {len(user_ids)} users (type: {notification_type})" + ) + return { + "status": "success", + "sent_count": len(notifications), + "user_ids": user_ids, + } + + except Exception as exc: + logger.exception(f"Failed to send bulk notifications: {exc}") + raise self.retry(exc=exc) + + +@shared_task( + name="notifications.cleanup_old_notifications", + bind=True, +) +def cleanup_old_notifications_task(self, days: int = 90): + """ + Периодическая задача для удаления старых прочитанных уведомлений. + + Args: + days: Количество дней, после которого удаляются прочитанные уведомления + """ + try: + from notifications.models import Notification + from datetime import timedelta + + cutoff_date = timezone.now() - timedelta(days=days) + + deleted_count, _ = Notification.objects.filter( + is_read=True, + created_at__lt=cutoff_date + ).delete() + + logger.info( + f"Cleaned up {deleted_count} old notifications (older than {days} days)" + ) + return {"status": "success", "deleted_count": deleted_count} + + except Exception as exc: + logger.exception(f"Failed to cleanup old notifications: {exc}") + raise self.retry(exc=exc) + + +@shared_task( + name="notifications.send_digest_email", + bind=True, +) +def send_digest_email_task(self, user_id: int): + """ + Отправка email-дайджеста непрочитанных уведомлений пользователю. + + Может вызываться периодически для пользователей с настроенными email-дайджестами. + + Args: + user_id: ID пользователя + """ + try: + from notifications.models import Notification + from common.emails import send_email + + user = User.objects.get(id=user_id) + + # Получаем непрочитанные уведомления за последние 24 часа + cutoff_time = timezone.now() - timezone.timedelta(hours=24) + unread_notifications = Notification.objects.filter( + user=user, + is_read=False, + created_at__gte=cutoff_time + ).order_by('-created_at')[:20] + + if not unread_notifications: + logger.debug(f"No unread notifications for user {user_id}, skipping digest") + return {"status": "skipped", "reason": "no_notifications"} + + # Формируем HTML для email + notification_html = "" + for notif in unread_notifications: + notification_html += f""" +