feat: полностью асинхронная обработка уведомлений
- Добавлены Celery задачи для календаря, ленты и заявок - Обновлены сигналы для запуска задач через transaction.on_commit - Signal handlers теперь сразу возвращают управление - Вся обработка (циклы, отправка) происходит в фоне Аналогично сообщениям в чатах: - process_event_notifications_task (календарь) - process_post_notifications_task (лента) - process_request_notifications_task (заявки) Fixes: Блокировка транзакций при массовой отправке уведомлений
This commit is contained in:
parent
1afe904b0d
commit
a0a4df8aad
@ -22,21 +22,48 @@ Employee = get_user_model()
|
||||
@receiver(post_save, sender=CalendarEvent)
|
||||
def create_event_notifications(sender, instance, created, **kwargs):
|
||||
"""
|
||||
Создает уведомления при создании или изменении события.
|
||||
Создает уведомления при создании или изменении события асинхронно.
|
||||
|
||||
Уведомления отправляются:
|
||||
- При создании - всем сотрудникам компании или отдела
|
||||
- При изменении - участникам события
|
||||
"""
|
||||
event = instance
|
||||
from django.db import transaction
|
||||
from notifications.tasks import process_event_notifications_task
|
||||
|
||||
if created:
|
||||
# Новое событие - уведомляем всех участников
|
||||
notify_event_created(event)
|
||||
else:
|
||||
# Проверяем изменение ключевых полей
|
||||
if hasattr(event, '_changed_fields'):
|
||||
notify_event_changed(event, event._changed_fields)
|
||||
def send_task():
|
||||
"""Отложенная отправка задачи после commit транзакции"""
|
||||
try:
|
||||
if created:
|
||||
# Новое событие
|
||||
process_event_notifications_task.delay(
|
||||
event_id=instance.id,
|
||||
action='created'
|
||||
)
|
||||
else:
|
||||
# Изменение события
|
||||
changed_fields = getattr(instance, '_changed_fields', [])
|
||||
if changed_fields:
|
||||
process_event_notifications_task.delay(
|
||||
event_id=instance.id,
|
||||
action='updated',
|
||||
changed_fields=changed_fields
|
||||
)
|
||||
except Exception as e:
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.warning(
|
||||
f"Failed to queue event notifications task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
if created:
|
||||
notify_event_created(instance)
|
||||
else:
|
||||
if hasattr(instance, '_changed_fields'):
|
||||
notify_event_changed(instance, instance._changed_fields)
|
||||
|
||||
transaction.on_commit(send_task)
|
||||
|
||||
|
||||
@receiver(pre_save, sender=CalendarEvent)
|
||||
@ -69,33 +96,45 @@ def track_event_changes(sender, instance, **kwargs):
|
||||
@receiver(pre_delete, sender=CalendarEvent)
|
||||
def notify_event_cancelled(sender, instance, **kwargs):
|
||||
"""
|
||||
Создает уведомления при удалении (отмене) события.
|
||||
Создает уведомления при удалении (отмене) события асинхронно.
|
||||
"""
|
||||
event = instance
|
||||
from notifications.tasks import process_event_notifications_task
|
||||
|
||||
# Определяем получателей
|
||||
recipients = get_event_recipients(event)
|
||||
|
||||
# Создаем уведомления об отмене
|
||||
for recipient in recipients:
|
||||
NotificationService.create_notification_async(
|
||||
recipient=recipient,
|
||||
notification_type_code='event_cancelled',
|
||||
title='Событие отменено',
|
||||
message=(
|
||||
f'Событие "{event.title}" '
|
||||
f'({event.start_date.strftime("%d.%m.%Y")}) отменено'
|
||||
),
|
||||
action_url=get_calendar_url(event),
|
||||
metadata={
|
||||
'event_title': event.title,
|
||||
'event_date': event.start_date.isoformat(),
|
||||
'department_id': (
|
||||
event.department.id if event.department else None
|
||||
),
|
||||
'is_company_event': event.is_company,
|
||||
}
|
||||
try:
|
||||
# Запускаем задачу синхронно, т.к. pre_delete выполняется до удаления
|
||||
# и объект ещё существует в БД
|
||||
process_event_notifications_task.apply_async(
|
||||
args=[instance.id, 'cancelled'],
|
||||
countdown=1 # Небольшая задержка чтобы удаление успело произойти
|
||||
)
|
||||
except Exception as e:
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.warning(
|
||||
f"Failed to queue event cancellation task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
recipients = get_event_recipients(instance)
|
||||
for recipient in recipients:
|
||||
NotificationService.create_notification(
|
||||
recipient=recipient,
|
||||
notification_type_code='event_cancelled',
|
||||
title='Событие отменено',
|
||||
message=(
|
||||
f'Событие "{instance.title}" '
|
||||
f'({instance.start_date.strftime("%d.%m.%Y")}) отменено'
|
||||
),
|
||||
action_url=get_calendar_url(instance),
|
||||
metadata={
|
||||
'event_title': instance.title,
|
||||
'event_date': instance.start_date.isoformat(),
|
||||
'department_id': (
|
||||
instance.department.id if instance.department else None
|
||||
),
|
||||
'is_company_event': instance.is_company,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ===== Вспомогательные функции =====
|
||||
|
||||
@ -21,7 +21,7 @@ Employee = get_user_model()
|
||||
@receiver(post_save, sender=Post)
|
||||
def create_post_notification(sender, instance, created, **kwargs):
|
||||
"""
|
||||
Создает уведомления при создании новой публикации.
|
||||
Создает уведомления при создании новой публикации асинхронно.
|
||||
|
||||
Уведомления отправляются в зависимости от типа:
|
||||
- TYPE_COMPANY - всем сотрудникам
|
||||
@ -31,8 +31,27 @@ def create_post_notification(sender, instance, created, **kwargs):
|
||||
if not created:
|
||||
return
|
||||
|
||||
post = instance
|
||||
notify_new_post(post)
|
||||
from django.db import transaction
|
||||
from notifications.tasks import process_post_notifications_task
|
||||
|
||||
def send_task():
|
||||
"""Отложенная отправка задачи после commit транзакции"""
|
||||
try:
|
||||
process_post_notifications_task.delay(
|
||||
post_id=instance.id,
|
||||
action='created'
|
||||
)
|
||||
except Exception as e:
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.warning(
|
||||
f"Failed to queue post notifications task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
notify_new_post(instance)
|
||||
|
||||
transaction.on_commit(send_task)
|
||||
|
||||
|
||||
@receiver(post_save, sender=Comment)
|
||||
|
||||
@ -428,3 +428,307 @@ def process_message_notifications_task(self, message_id: int):
|
||||
except Exception as exc:
|
||||
logger.exception(f"Failed to process notifications for message {message_id}: {exc}")
|
||||
raise self.retry(exc=exc)
|
||||
|
||||
|
||||
@shared_task(
|
||||
name="calendar.process_event_notifications",
|
||||
bind=True,
|
||||
max_retries=3,
|
||||
default_retry_delay=30,
|
||||
)
|
||||
def process_event_notifications_task(
|
||||
self,
|
||||
event_id: int,
|
||||
action: str, # 'created', 'updated', 'cancelled'
|
||||
changed_fields: list = None
|
||||
):
|
||||
"""
|
||||
Обрабатывает уведомления для событий календаря асинхронно.
|
||||
|
||||
Args:
|
||||
event_id: ID события
|
||||
action: Тип действия ('created', 'updated', 'cancelled')
|
||||
changed_fields: Список изменённых полей (для updated)
|
||||
"""
|
||||
try:
|
||||
from calendar_app.models import CalendarEvent
|
||||
from notifications.services import NotificationService
|
||||
|
||||
event = CalendarEvent.objects.select_related(
|
||||
'department', 'creator'
|
||||
).prefetch_related('participants').get(id=event_id)
|
||||
|
||||
# Определяем получателей
|
||||
recipients = set()
|
||||
|
||||
if event.is_company:
|
||||
from employees.models import Employee
|
||||
recipients = set(Employee.objects.filter(is_active=True).exclude(id=event.creator.id))
|
||||
elif event.department:
|
||||
recipients = set(event.department.employees.filter(is_active=True).exclude(id=event.creator.id))
|
||||
else:
|
||||
recipients = set(event.participants.exclude(id=event.creator.id))
|
||||
|
||||
if not recipients:
|
||||
return {"status": "success", "notifications_sent": 0}
|
||||
|
||||
# Определяем тип и текст уведомления
|
||||
if action == 'created':
|
||||
notification_type = 'event_created'
|
||||
title = 'Новое событие в календаре'
|
||||
message = f'"{event.title}" ({event.start_date.strftime("%d.%m.%Y")})'
|
||||
elif action == 'cancelled':
|
||||
notification_type = 'event_cancelled'
|
||||
title = 'Событие отменено'
|
||||
message = f'Событие "{event.title}" ({event.start_date.strftime("%d.%m.%Y")}) отменено'
|
||||
else: # updated
|
||||
notification_type = 'event_updated'
|
||||
title = 'Событие изменено'
|
||||
message = f'Изменено событие "{event.title}"'
|
||||
|
||||
# URL календаря
|
||||
if event.department:
|
||||
action_url = f'/calendar/?department={event.department.id}'
|
||||
else:
|
||||
action_url = '/calendar/'
|
||||
|
||||
notifications_count = 0
|
||||
|
||||
# Отправляем уведомления
|
||||
for recipient in recipients:
|
||||
NotificationService.create_notification(
|
||||
recipient=recipient,
|
||||
notification_type_code=notification_type,
|
||||
title=title,
|
||||
message=message,
|
||||
content_object=event,
|
||||
action_url=action_url,
|
||||
metadata={
|
||||
'event_id': event.id,
|
||||
'event_title': event.title,
|
||||
'event_date': event.start_date.isoformat(),
|
||||
'department_id': event.department.id if event.department else None,
|
||||
'is_company_event': event.is_company,
|
||||
}
|
||||
)
|
||||
notifications_count += 1
|
||||
|
||||
logger.info(
|
||||
f"Processed event {event_id} ({action}): sent {notifications_count} notifications"
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"event_id": event_id,
|
||||
"action": action,
|
||||
"notifications_sent": notifications_count
|
||||
}
|
||||
|
||||
except CalendarEvent.DoesNotExist:
|
||||
logger.error(f"CalendarEvent {event_id} not found")
|
||||
return {"status": "error", "reason": "event_not_found"}
|
||||
|
||||
except Exception as exc:
|
||||
logger.exception(f"Failed to process event notifications: {exc}")
|
||||
raise self.retry(exc=exc)
|
||||
|
||||
|
||||
@shared_task(
|
||||
name="feed.process_post_notifications",
|
||||
bind=True,
|
||||
max_retries=3,
|
||||
default_retry_delay=30,
|
||||
)
|
||||
def process_post_notifications_task(self, post_id: int, action: str = 'created'):
|
||||
"""
|
||||
Обрабатывает уведомления для публикаций асинхронно.
|
||||
|
||||
Args:
|
||||
post_id: ID публикации
|
||||
action: Тип действия ('created', 'comment')
|
||||
"""
|
||||
try:
|
||||
from feed.models import Post
|
||||
from notifications.services import NotificationService
|
||||
|
||||
post = Post.objects.select_related(
|
||||
'author', 'department'
|
||||
).get(id=post_id)
|
||||
|
||||
# Определяем получателей
|
||||
recipients = set()
|
||||
|
||||
if post.type == 'company':
|
||||
from employees.models import Employee
|
||||
recipients = set(Employee.objects.filter(is_active=True).exclude(id=post.author.id))
|
||||
elif post.type == 'department' and post.department:
|
||||
recipients = set(post.department.employees.filter(is_active=True).exclude(id=post.author.id))
|
||||
|
||||
if not recipients:
|
||||
return {"status": "success", "notifications_sent": 0}
|
||||
|
||||
# Определяем контекст
|
||||
if post.type == 'company':
|
||||
context = 'компании'
|
||||
elif post.type == 'department' and post.department:
|
||||
context = f'отдела {post.department.name}'
|
||||
else:
|
||||
context = f'от {post.author.get_full_name() or post.author.username}'
|
||||
|
||||
notifications_count = 0
|
||||
|
||||
# Отправляем уведомления
|
||||
for recipient in recipients:
|
||||
NotificationService.create_notification(
|
||||
recipient=recipient,
|
||||
notification_type_code='feed_new_post',
|
||||
title=f'Новая публикация {context}',
|
||||
message=post.title,
|
||||
content_object=post,
|
||||
action_url=f'/post/{post.id}/',
|
||||
metadata={
|
||||
'post_id': post.id,
|
||||
'post_title': post.title,
|
||||
'author_id': post.author.id,
|
||||
'post_type': post.type,
|
||||
}
|
||||
)
|
||||
notifications_count += 1
|
||||
|
||||
logger.info(
|
||||
f"Processed post {post_id}: sent {notifications_count} notifications"
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"post_id": post_id,
|
||||
"notifications_sent": notifications_count
|
||||
}
|
||||
|
||||
except Post.DoesNotExist:
|
||||
logger.error(f"Post {post_id} not found")
|
||||
return {"status": "error", "reason": "post_not_found"}
|
||||
|
||||
except Exception as exc:
|
||||
logger.exception(f"Failed to process post notifications: {exc}")
|
||||
raise self.retry(exc=exc)
|
||||
|
||||
|
||||
@shared_task(
|
||||
name="requests.process_request_notifications",
|
||||
bind=True,
|
||||
max_retries=3,
|
||||
default_retry_delay=30,
|
||||
)
|
||||
def process_request_notifications_task(
|
||||
self,
|
||||
request_id: int,
|
||||
action: str, # 'created', 'approved', 'rejected', 'comment'
|
||||
comment_id: int = None
|
||||
):
|
||||
"""
|
||||
Обрабатывает уведомления для заявлений асинхронно.
|
||||
|
||||
Args:
|
||||
request_id: ID заявления
|
||||
action: Тип действия
|
||||
comment_id: ID комментария (если action='comment')
|
||||
"""
|
||||
try:
|
||||
from requests_app.models import Request, RequestComment
|
||||
from notifications.services import NotificationService
|
||||
|
||||
request_obj = Request.objects.select_related(
|
||||
'employee'
|
||||
).prefetch_related('departments').get(id=request_id)
|
||||
|
||||
recipients = set()
|
||||
|
||||
if action == 'comment' and comment_id:
|
||||
comment = RequestComment.objects.select_related('author').get(id=comment_id)
|
||||
|
||||
# Уведомляем автора заявления
|
||||
recipients.add(request_obj.employee)
|
||||
|
||||
# Уведомляем руководителей отделов
|
||||
for dept in request_obj.departments.all():
|
||||
dept_heads = dept.employees.filter(
|
||||
departments_links__role='head',
|
||||
departments_links__is_active=True,
|
||||
is_active=True
|
||||
)
|
||||
recipients.update(dept_heads)
|
||||
|
||||
# Исключаем автора комментария
|
||||
recipients.discard(comment.author)
|
||||
|
||||
notification_type = 'request_comment'
|
||||
title = f"💬 Новый комментарий к заявлению от {request_obj.employee.get_full_name()}"
|
||||
message = f"{comment.author.get_full_name()} прокомментировал: {comment.text[:100]}"
|
||||
|
||||
elif action == 'approved':
|
||||
recipients = {request_obj.employee}
|
||||
notification_type = 'request_approved'
|
||||
title = f"✅ Заявление одобрено"
|
||||
message = f'Ваше заявление "{request_obj.get_type_display()}" одобрено'
|
||||
|
||||
elif action == 'rejected':
|
||||
recipients = {request_obj.employee}
|
||||
notification_type = 'request_rejected'
|
||||
title = f"❌ Заявление отклонено"
|
||||
message = f'Ваше заявление "{request_obj.get_type_display()}" отклонено'
|
||||
|
||||
else: # created
|
||||
# Уведомляем руководителей
|
||||
for dept in request_obj.departments.all():
|
||||
dept_heads = dept.employees.filter(
|
||||
departments_links__role='head',
|
||||
departments_links__is_active=True,
|
||||
is_active=True
|
||||
)
|
||||
recipients.update(dept_heads)
|
||||
|
||||
notification_type = 'request_new'
|
||||
title = f"📋 Новое заявление от {request_obj.employee.get_full_name()}"
|
||||
message = f'Тип: {request_obj.get_type_display()}'
|
||||
|
||||
if not recipients:
|
||||
return {"status": "success", "notifications_sent": 0}
|
||||
|
||||
notifications_count = 0
|
||||
|
||||
# Отправляем уведомления
|
||||
for recipient in recipients:
|
||||
NotificationService.create_notification(
|
||||
recipient=recipient,
|
||||
notification_type_code=notification_type,
|
||||
title=title,
|
||||
message=message,
|
||||
content_object=request_obj,
|
||||
action_url=f'/requests/{request_obj.id}/',
|
||||
metadata={
|
||||
'request_id': request_obj.id,
|
||||
'request_type': request_obj.type,
|
||||
'employee_id': request_obj.employee.id,
|
||||
}
|
||||
)
|
||||
notifications_count += 1
|
||||
|
||||
logger.info(
|
||||
f"Processed request {request_id} ({action}): sent {notifications_count} notifications"
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"request_id": request_id,
|
||||
"action": action,
|
||||
"notifications_sent": notifications_count
|
||||
}
|
||||
|
||||
except Request.DoesNotExist:
|
||||
logger.error(f"Request {request_id} not found")
|
||||
return {"status": "error", "reason": "request_not_found"}
|
||||
|
||||
except Exception as exc:
|
||||
logger.exception(f"Failed to process request notifications: {exc}")
|
||||
raise self.retry(exc=exc)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user