Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Полный переход на новую архитектуру уведомлений v2.0

Browse Source 
BREAKING CHANGES:
- Удалена старая система (6 моделей → 2 модели)
- models_new.py → models.py (Notification, UserChannelPreferences)
- signals_new.py → signals.py (notify.send API)
- admin_new.py → admin.py

Новая архитектура:
- Все отправки через Celery (полностью асинхронно)
- Модульная структура tasks/ (base, email, push, websocket)
- Каждый канал - отдельный класс задачи наследующий BaseNotificationTask
- Rate limiting, retry, приоритеты из коробки

Обратная совместимость:
- Создан services.py facade для старого API
- NotificationService.create_notification() работает через notify.send()
- Старый код продолжает работать без изменений

Удалено:
- Старые models.py, signals.py, admin.py, tasks.py (→ .old)
- telegram_bot.py, telegram_models.py (Telegram отключен)
- test_views.py, views.py, urls.py (устаревшие)

Создано:
- tasks/ модуль с классовой архитектурой (как senders/)
- tasks/base.py - BaseNotificationTask
- tasks/email.py - EmailNotificationTask + DigestEmailTask
- tasks/push.py - PushNotificationTask
- tasks/websocket.py - WebSocketNotificationTask

Обновлено:
- apps.py - загружает signals, channels, tasks
- channels.py - все отправки через Celery задачи
- Все импорты models_new → models
This commit is contained in:
lizerk 2026-03-09 16:36:14 +03:00
parent d12efde0d2
commit d5b4ae9f43
21 changed files with 1199 additions and 3333 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

507
backend/notifications/admin.py
View File

@ -1,351 +1,276 @@
"""
Admin для упрощенной системы уведомлений
"""
from django.contrib import admin
from django.utils.html import format_html
from .models import (
NotificationCategory,
NotificationType,
Notification,
UserNotificationSettings,
NotificationTemplate,
)
from .telegram_models import TelegramUser
@admin.register(NotificationCategory)
class NotificationCategoryAdmin(admin.ModelAdmin):
list_display = ['name', 'code', 'icon_preview', 'color', 'order', 'is_active']
list_editable = ['order', 'is_active']
list_filter = ['is_active']
search_fields = ['name', 'code', 'description']
ordering = ['order', 'name']
def icon_preview(self, obj):
return format_html('<i class="{}" style="font-size: 1.2em;"></i>', obj.icon)
icon_preview.short_description = 'Иконка'
@admin.register(NotificationType)
class NotificationTypeAdmin(admin.ModelAdmin):
list_display = [
'name',
'code',
'category',
'priority',
'default_enabled',
'is_groupable',
'is_required',
'is_active',
]
list_filter = [
'category',
'priority',
'default_enabled',
'is_groupable',
'is_required',
'is_active',
]
search_fields = ['name', 'code', 'description']
list_editable = ['is_active']
ordering = ['category__order', 'name']
fieldsets = (
('Основная информация', {
'fields': ('category', 'code', 'name', 'description', 'is_active')
}),
('Настройки по умолчанию', {
'fields': ('default_enabled', 'default_channels', 'priority', 'is_required')
}),
('Группировка', {
'fields': ('is_groupable', 'grouping_window_minutes'),
'classes': ('collapse',)
}),
)
from django.utils.safestring import mark_safe
from .models import Notification, UserChannelPreferences
@admin.register(Notification)
class NotificationAdmin(admin.ModelAdmin):
"""Admin для модели Notification"""
list_display = [
'id',
'recipient_name',
'title_short',
'notification_type',
'is_read',
'created_at',
'channels_sent',
'get_actor_display',
'verb',
'get_recipient_display',
'unread',
'public',
'deleted',
'emailed',
'timestamp',
]
list_filter = [
'is_read',
'is_archived',
'notification_type__category',
'notification_type',
'created_at',
'unread',
'public',
'deleted',
'emailed',
'verb',
'timestamp',
]
search_fields = [
'title',
'message',
'recipient__username',
'recipient__first_name',
'recipient__last_name',
'recipient__email',
'verb',
'description',
]
readonly_fields = [
'created_at',
'updated_at',
'read_at',
'sent_at',
'archived_at',
'get_actor_info',
'get_action_object_info',
'get_target_info',
'timestamp',
'timestamp_read',
]
date_hierarchy = 'created_at'
ordering = ['-created_at']
fieldsets = (
('Основная информация', {
'fields': ('recipient', 'notification_type', 'title', 'message', 'short_message')
}),
('Связанный объект', {
'fields': ('content_type', 'object_id'),
'classes': ('collapse',)
}),
('Действие', {
'fields': ('action_url', 'action_text')
}),
('Статус', {
'fields': ('is_read', 'read_at', 'is_archived', 'archived_at')
}),
('Доставка', {
'fields': (
'sent_web',
'sent_email',
'sent_telegram',
'sent_whatsapp',
'sent_wechat',
'sent_at',
'recipient',
'verb',
'description',
'action_url',
)
}),
('Связанные объекты', {
'fields': (
'get_actor_info',
'get_action_object_info',
'get_target_info',
),
'classes': ('collapse',)
'classes': ('collapse',),
}),
('Группировка', {
'fields': ('group_key', 'is_grouped', 'grouped_count'),
'classes': ('collapse',)
('Дополнительные данные', {
'fields': (
'data',
),
'classes': ('collapse',),
}),
('Метаданные', {
'fields': ('metadata', 'created_at', 'updated_at'),
'classes': ('collapse',)
('Статусы', {
'fields': (
'unread',
'timestamp_read',
'public',
'deleted',
'emailed',
)
}),
('Временные метки', {
'fields': (
'timestamp',
)
}),
)
def recipient_name(self, obj):
return obj.recipient.get_full_name() or obj.recipient.username
recipient_name.short_description = 'Получатель'
recipient_name.admin_order_field = 'recipient__last_name'
date_hierarchy = 'timestamp'
def title_short(self, obj):
return obj.title[:50] + '...' if len(obj.title) > 50 else obj.title
title_short.short_description = 'Заголовок'
actions = [
'mark_as_read',
'mark_as_unread',
'mark_as_deleted',
'mark_as_not_deleted',
]
def channels_sent(self, obj):
channels = []
if obj.sent_web:
channels.append('🌐 Веб')
if obj.sent_email:
channels.append('📧 Email')
if obj.sent_telegram:
channels.append('✈️ Telegram')
if obj.sent_whatsapp:
channels.append('💬 WhatsApp')
if obj.sent_wechat:
channels.append('💚 WeChat')
return ', '.join(channels) if channels else '—'
channels_sent.short_description = 'Отправлено'
def get_actor_display(self, obj):
"""Отображение актора в списке"""
if obj.actor:
return format_html(
'<span title="{}">{}</span>',
obj.actor_content_type.model,
str(obj.actor)[:50]
)
return '-'
get_actor_display.short_description = 'Актор'
actions = ['mark_as_read', 'mark_as_unread', 'archive_notifications']
def get_recipient_display(self, obj):
"""Отображение получателя"""
return format_html(
'<a href="/admin/employees/employee/{}/change/">{}</a>',
obj.recipient.id,
obj.recipient.get_short_name()
)
get_recipient_display.short_description = 'Получатель'
def get_actor_info(self, obj):
"""Подробная информация об акторе"""
if not obj.actor:
return '-'
info = f"""
<strong>Тип:</strong> {obj.actor_content_type.model}<br>
<strong>ID:</strong> {obj.actor_object_id}<br>
<strong>Объект:</strong> {obj.actor}
"""
return mark_safe(info)
get_actor_info.short_description = 'Информация об акторе'
def get_action_object_info(self, obj):
"""Подробная информация об объекте действия"""
if not obj.action_object:
return '-'
info = f"""
<strong>Тип:</strong> {obj.action_object_content_type.model}<br>
<strong>ID:</strong> {obj.action_object_object_id}<br>
<strong>Объект:</strong> {obj.action_object}
"""
return mark_safe(info)
get_action_object_info.short_description = 'Объект действия'
def get_target_info(self, obj):
"""Подробная информация о целевом объекте"""
if not obj.target:
return '-'
info = f"""
<strong>Тип:</strong> {obj.target_content_type.model}<br>
<strong>ID:</strong> {obj.target_object_id}<br>
<strong>Объект:</strong> {obj.target}
"""
return mark_safe(info)
get_target_info.short_description = 'Целевой объект'
# === Actions ===
@admin.action(description='Отметить как прочитанные')
def mark_as_read(self, request, queryset):
"""Отметить выбранные уведомления как прочитанные"""
count = 0
for notification in queryset.filter(is_read=False):
notification.mark_as_read()
count += 1
self.message_user(request, f'Отмечено как прочитанное: {count}')
mark_as_read.short_description = 'Отметить как прочитанное'
for notification in queryset:
if notification.unread:
notification.mark_as_read()
count += 1
self.message_user(request, f'Отмечено как прочитанные: {count} уведомлений')
@admin.action(description='Отметить как непрочитанные')
def mark_as_unread(self, request, queryset):
count = queryset.filter(is_read=True).update(is_read=False, read_at=None)
self.message_user(request, f'Отмечено как непрочитанное: {count}')
mark_as_unread.short_description = 'Отметить как непрочитанное'
def archive_notifications(self, request, queryset):
"""Отметить выбранные уведомления как непрочитанные"""
count = 0
for notification in queryset.filter(is_archived=False):
notification.archive()
count += 1
self.message_user(request, f'Заархивировано: {count}')
archive_notifications.short_description = 'Архивировать'
for notification in queryset:
if not notification.unread:
notification.mark_as_unread()
count += 1
self.message_user(request, f'Отмечено как непрочитанные: {count} уведомлений')
@admin.action(description='Пометить как удаленные')
def mark_as_deleted(self, request, queryset):
"""Пометить как удаленные (мягкое удаление)"""
count = queryset.filter(deleted=False).update(deleted=True)
self.message_user(request, f'Помечено как удаленные: {count} уведомлений')
@admin.action(description='Восстановить удаленные')
def mark_as_not_deleted(self, request, queryset):
"""Восстановить удаленные"""
count = queryset.filter(deleted=True).update(deleted=False)
self.message_user(request, f'Восстановлено: {count} уведомлений')
@admin.register(UserNotificationSettings)
class UserNotificationSettingsAdmin(admin.ModelAdmin):
@admin.register(UserChannelPreferences)
class UserChannelPreferencesAdmin(admin.ModelAdmin):
"""Admin для настроек каналов пользователя"""
list_display = [
'user_name',
'notification_type',
'is_enabled',
'channels_enabled',
'quiet_hours_status',
'user',
'web_enabled',
'email_enabled',
'push_enabled',
'email_frequency',
'dnd_enabled',
'get_disabled_verbs_count',
]
list_filter = [
'is_enabled',
'notification_type__category',
'send_web',
'send_email',
'send_telegram',
'quiet_hours_enabled',
'web_enabled',
'email_enabled',
'push_enabled',
'email_frequency',
'dnd_enabled',
]
search_fields = [
'user__username',
'user__first_name',
'user__last_name',
'user__email',
'notification_type__name',
]
ordering = ['user__last_name', 'notification_type__category__order']
fieldsets = (
('Основная информация', {
'fields': ('user', 'notification_type', 'is_enabled')
}),
('Каналы доставки', {
'fields': ('send_web', 'send_email', 'send_telegram', 'send_whatsapp', 'send_wechat')
}),
('Тихий режим', {
'fields': ('quiet_hours_enabled', 'quiet_start_time', 'quiet_end_time'),
'classes': ('collapse',)
}),
)
def user_name(self, obj):
return obj.user.get_full_name() or obj.user.username
user_name.short_description = 'Пользователь'
user_name.admin_order_field = 'user__last_name'
def channels_enabled(self, obj):
channels = []
if obj.send_web:
channels.append('🌐')
if obj.send_email:
channels.append('📧')
if obj.send_telegram:
channels.append('✈️')
if obj.send_whatsapp:
channels.append('💬')
if obj.send_wechat:
channels.append('💚')
return ' '.join(channels) if channels else '—'
channels_enabled.short_description = 'Каналы'
def quiet_hours_status(self, obj):
if obj.quiet_hours_enabled and obj.quiet_start_time and obj.quiet_end_time:
return f'🌙 {obj.quiet_start_time.strftime("%H:%M")} - {obj.quiet_end_time.strftime("%H:%M")}'
return '—'
quiet_hours_status.short_description = 'Тихий режим'
@admin.register(NotificationTemplate)
class NotificationTemplateAdmin(admin.ModelAdmin):
list_display = [
'notification_type',
'channel',
'is_active',
'updated_at',
]
list_filter = [
'channel',
'is_active',
'notification_type__category',
]
search_fields = [
'notification_type__name',
'title_template',
'message_template',
]
ordering = ['notification_type__category__order', 'channel']
fieldsets = (
('Основная информация', {
'fields': ('notification_type', 'channel', 'is_active')
}),
('Шаблоны', {
'fields': ('title_template', 'message_template', 'action_button_template')
}),
('Email (дополнительно)', {
'fields': ('html_template',),
'classes': ('collapse',)
}),
)
@admin.register(TelegramUser)
class TelegramUserAdmin(admin.ModelAdmin):
list_display = [
'user_name',
'telegram_username',
'telegram_id',
'is_active',
'is_blocked',
'linked_at',
]
list_filter = [
'is_active',
'is_blocked',
'linked_at',
]
search_fields = [
'user__first_name',
'user__last_name',
'user__email',
'telegram_username',
'telegram_id',
]
readonly_fields = [
'telegram_id',
'link_code',
'link_code_created_at',
'linked_at',
'last_interaction_at',
'updated_at',
]
ordering = ['-linked_at']
fieldsets = (
('Пользователь', {
'fields': ('user', 'is_active')
'fields': ('user',)
}),
('Telegram информация', {
('Каналы доставки', {
'fields': (
'telegram_id',
'telegram_username',
'first_name',
'last_name',
'is_blocked',
'web_enabled',
'email_enabled',
'push_enabled',
)
}),
('Код привязки', {
'fields': ('link_code', 'link_code_created_at'),
'classes': ('collapse',)
('Email настройки', {
'fields': (
'email_frequency',
)
}),
('Временные метки', {
'fields': ('linked_at', 'last_interaction_at', 'updated_at'),
'classes': ('collapse',)
('Режим "Не беспокоить"', {
'fields': (
'dnd_enabled',
'dnd_start_time',
'dnd_end_time',
)
}),
('Фильтры', {
'fields': (
'disabled_verbs',
),
'classes': ('collapse',),
}),
('Метаданные', {
'fields': (
'created_at',
'updated_at',
),
'classes': ('collapse',),
}),
)
def user_name(self, obj):
return obj.user.get_full_name() or obj.user.username
user_name.short_description = 'Пользователь'
user_name.admin_order_field = 'user__last_name'
readonly_fields = ['created_at', 'updated_at']
actions = ['unlink_accounts', 'reactivate_accounts']
def unlink_accounts(self, request, queryset):
count = queryset.filter(is_active=True).update(is_active=False)
self.message_user(request, f'Отвязано аккаунтов: {count}')
unlink_accounts.short_description = 'Отвязать аккаунты'
def reactivate_accounts(self, request, queryset):
count = queryset.update(is_active=True, is_blocked=False)
self.message_user(request, f'Реактивировано аккаунтов: {count}')
reactivate_accounts.short_description = 'Реактивировать'
def get_disabled_verbs_count(self, obj):
"""Количество отключенных типов"""
count = len(obj.disabled_verbs)
if count == 0:
return '-'
return format_html(
'<span title="{}">{} типов</span>',
', '.join(obj.disabled_verbs),
count
)
get_disabled_verbs_count.short_description = 'Отключено типов'

276
backend/notifications/admin_new.py
View File

@ -1,276 +0,0 @@
"""
Admin для упрощенной системы уведомлений
"""
from django.contrib import admin
from django.utils.html import format_html
from django.utils.safestring import mark_safe
from .models_new import Notification, UserChannelPreferences
@admin.register(Notification)
class NotificationAdmin(admin.ModelAdmin):
"""Admin для модели Notification"""
list_display = [
'id',
'get_actor_display',
'verb',
'get_recipient_display',
'unread',
'public',
'deleted',
'emailed',
'timestamp',
]
list_filter = [
'unread',
'public',
'deleted',
'emailed',
'verb',
'timestamp',
]
search_fields = [
'recipient__username',
'recipient__first_name',
'recipient__last_name',
'recipient__email',
'verb',
'description',
]
readonly_fields = [
'get_actor_info',
'get_action_object_info',
'get_target_info',
'timestamp',
'timestamp_read',
]
fieldsets = (
('Основная информация', {
'fields': (
'recipient',
'verb',
'description',
'action_url',
)
}),
('Связанные объекты', {
'fields': (
'get_actor_info',
'get_action_object_info',
'get_target_info',
),
'classes': ('collapse',),
}),
('Дополнительные данные', {
'fields': (
'data',
),
'classes': ('collapse',),
}),
('Статусы', {
'fields': (
'unread',
'timestamp_read',
'public',
'deleted',
'emailed',
)
}),
('Временные метки', {
'fields': (
'timestamp',
)
}),
)
date_hierarchy = 'timestamp'
actions = [
'mark_as_read',
'mark_as_unread',
'mark_as_deleted',
'mark_as_not_deleted',
]
def get_actor_display(self, obj):
"""Отображение актора в списке"""
if obj.actor:
return format_html(
'<span title="{}">{}</span>',
obj.actor_content_type.model,
str(obj.actor)[:50]
)
return '-'
get_actor_display.short_description = 'Актор'
def get_recipient_display(self, obj):
"""Отображение получателя"""
return format_html(
'<a href="/admin/employees/employee/{}/change/">{}</a>',
obj.recipient.id,
obj.recipient.get_short_name()
)
get_recipient_display.short_description = 'Получатель'
def get_actor_info(self, obj):
"""Подробная информация об акторе"""
if not obj.actor:
return '-'
info = f"""
<strong>Тип:</strong> {obj.actor_content_type.model}<br>
<strong>ID:</strong> {obj.actor_object_id}<br>
<strong>Объект:</strong> {obj.actor}
"""
return mark_safe(info)
get_actor_info.short_description = 'Информация об акторе'
def get_action_object_info(self, obj):
"""Подробная информация об объекте действия"""
if not obj.action_object:
return '-'
info = f"""
<strong>Тип:</strong> {obj.action_object_content_type.model}<br>
<strong>ID:</strong> {obj.action_object_object_id}<br>
<strong>Объект:</strong> {obj.action_object}
"""
return mark_safe(info)
get_action_object_info.short_description = 'Объект действия'
def get_target_info(self, obj):
"""Подробная информация о целевом объекте"""
if not obj.target:
return '-'
info = f"""
<strong>Тип:</strong> {obj.target_content_type.model}<br>
<strong>ID:</strong> {obj.target_object_id}<br>
<strong>Объект:</strong> {obj.target}
"""
return mark_safe(info)
get_target_info.short_description = 'Целевой объект'
# === Actions ===
@admin.action(description='Отметить как прочитанные')
def mark_as_read(self, request, queryset):
"""Отметить выбранные уведомления как прочитанные"""
count = 0
for notification in queryset:
if notification.unread:
notification.mark_as_read()
count += 1
self.message_user(request, f'Отмечено как прочитанные: {count} уведомлений')
@admin.action(description='Отметить как непрочитанные')
def mark_as_unread(self, request, queryset):
"""Отметить выбранные уведомления как непрочитанные"""
count = 0
for notification in queryset:
if not notification.unread:
notification.mark_as_unread()
count += 1
self.message_user(request, f'Отмечено как непрочитанные: {count} уведомлений')
@admin.action(description='Пометить как удаленные')
def mark_as_deleted(self, request, queryset):
"""Пометить как удаленные (мягкое удаление)"""
count = queryset.filter(deleted=False).update(deleted=True)
self.message_user(request, f'Помечено как удаленные: {count} уведомлений')
@admin.action(description='Восстановить удаленные')
def mark_as_not_deleted(self, request, queryset):
"""Восстановить удаленные"""
count = queryset.filter(deleted=True).update(deleted=False)
self.message_user(request, f'Восстановлено: {count} уведомлений')
@admin.register(UserChannelPreferences)
class UserChannelPreferencesAdmin(admin.ModelAdmin):
"""Admin для настроек каналов пользователя"""
list_display = [
'user',
'web_enabled',
'email_enabled',
'push_enabled',
'email_frequency',
'dnd_enabled',
'get_disabled_verbs_count',
]
list_filter = [
'web_enabled',
'email_enabled',
'push_enabled',
'email_frequency',
'dnd_enabled',
]
search_fields = [
'user__username',
'user__first_name',
'user__last_name',
'user__email',
]
fieldsets = (
('Пользователь', {
'fields': ('user',)
}),
('Каналы доставки', {
'fields': (
'web_enabled',
'email_enabled',
'push_enabled',
)
}),
('Email настройки', {
'fields': (
'email_frequency',
)
}),
('Режим "Не беспокоить"', {
'fields': (
'dnd_enabled',
'dnd_start_time',
'dnd_end_time',
)
}),
('Фильтры', {
'fields': (
'disabled_verbs',
),
'classes': ('collapse',),
}),
('Метаданные', {
'fields': (
'created_at',
'updated_at',
),
'classes': ('collapse',),
}),
)
readonly_fields = ['created_at', 'updated_at']
def get_disabled_verbs_count(self, obj):
"""Количество отключенных типов"""
count = len(obj.disabled_verbs)
if count == 0:
return '-'
return format_html(
'<span title="{}">{} типов</span>',
', '.join(obj.disabled_verbs),
count
)
get_disabled_verbs_count.short_description = 'Отключено типов'

8
backend/notifications/apps.py
View File

@ -7,5 +7,11 @@ class NotificationsConfig(AppConfig):
verbose_name = 'Уведомления'
def ready(self):
# Загружаем сигналы (notify.send API)
import notifications.signals # noqa
# Old signals_telegram removed - Telegram support moved to new architecture
# Загружаем обработчики каналов (post_save → Celery tasks)
import notifications.channels # noqa
# Загружаем Celery задачи
import notifications.tasks # noqa

86
backend/notifications/channels.py
View File

@ -1,30 +1,26 @@
"""
Обработчики каналов доставки уведомлений
Слушают post_save сигнал от Notification и отправляют по каналам:
- WebSocket (realtime)
- Email
- Web Push
Слушают post_save сигнал от Notification и асинхронно отправляют через Celery:
- WebSocket (realtime) - мгновенно
- Email - с retry и rate limiting
- Web Push - с автоматическим удалением неактивных устройств
"""
from django.db.models.signals import post_save
from django.dispatch import receiver
import logging
from .senders import (
EmailNotificationSender,
WebSocketNotificationSender,
PushNotificationSender,
)
logger = logging.getLogger(__name__)
@receiver(post_save, sender='notifications.Notification')
def handle_notification_channels(sender, instance, created, **kwargs):
def route_notification_to_channels(sender, instance, created, **kwargs):
"""
Главный роутер уведомлений по каналам
Вызывается при создании нового уведомления
Главный роутер уведомлений по каналам.
Вызывается при создании нового уведомления.
Асинхронно отправляет в Celery очередь для каждого канала.
"""
if not created:
return # Обрабатываем только новые уведомления
@ -37,7 +33,7 @@ def handle_notification_channels(sender, instance, created, **kwargs):
prefs = user.channel_preferences
except Exception:
# Если настроек нет - создаем с дефолтными значениями
from .models_new import UserChannelPreferences
from .models import UserChannelPreferences
prefs = UserChannelPreferences.objects.create(user=user)
# Проверяем, не отключен ли этот тип уведомлений
@ -45,59 +41,33 @@ def handle_notification_channels(sender, instance, created, **kwargs):
logger.debug(f'Notification verb "{notification.verb}" disabled for user {user.id}')
return
# Импортируем задачи
from .tasks import (
send_websocket_notification,
send_email_notification,
send_push_notification,
)
# Проверяем режим "Не беспокоить"
if prefs.is_in_dnd_period():
logger.debug(f'User {user.id} is in DND period, skipping non-web channels')
# В DND режиме отправляем только веб (без звука)
is_dnd = prefs.is_in_dnd_period()
if is_dnd:
logger.debug(f'User {user.id} is in DND period, only web notifications (silent)')
# В DND режиме только WebSocket без звука
if prefs.web_enabled:
send_websocket_notification(notification, silent=True)
send_websocket_notification.delay(notification.id, silent=True)
return
# Отправляем по каналам согласно настройкам
# Отправляем по каналам асинхронно через Celery
if prefs.web_enabled:
sender = WebSocketNotificationSender()
sender.send(notification, silent=prefs.is_in_dnd_period())
send_websocket_notification.delay(notification.id, silent=False)
if prefs.email_enabled and prefs.email_frequency == 'instant':
sender = EmailNotificationSender()
sender.send(notification)
send_email_notification.delay(notification.id)
if prefs.push_enabled:
sender = PushNotificationSender()
sender.send(notification)
def send_websocket_notification(notification, silent=False):
"""
Отправка через WebSocket (realtime)
Args:
notification: Объект Notification
silent: Если True, уведомление без звука/визуальных эффектов
DEPRECATED: Используйте WebSocketNotificationSender().send()
"""
sender = WebSocketNotificationSender()
return sender.send(notification, silent=silent)
def send_email_notification(notification):
"""
Отправка email уведомления
Args:
notification: Объект Notification
DEPRECATED: Используйте EmailNotificationSender().send()
"""
sender = EmailNotificationSender()
return sender.send(notification)
def send_push_notification(notification):
"""
Отправка Web Push уведомления
send_push_notification.delay(notification.id)
Args:
notification: Объект Notification
@ -121,7 +91,7 @@ def send_email_digest(user, frequency='daily'):
try:
from datetime import timedelta
from django.utils import timezone
from .models_new import Notification
from .models import Notification
# Определяем период
if frequency == 'daily':

751
backend/notifications/models.py
View File

@ -1,546 +1,411 @@
"""
Упрощенная универсальная система уведомлений
Основана на архитектуре django-notifications-hq, адаптирована для Django 5.2+
Ключевые изменения:
- Одна модель Notification вместо 6
- GenericForeignKey для универсальности
- Простой API через сигналы
- Multi-channel поддержка (Web, Email, Push)
"""
from django.db import models
from django.conf import settings
from django.utils import timezone
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
# Импортируем Telegram модель
from .telegram_models import TelegramUser # noqa: F401
from django.db.models.query import QuerySet
class NotificationCategory(models.Model):
"""Категории уведомлений для группировки"""
class NotificationQuerySet(QuerySet):
"""QuerySet с удобными методами фильтрации"""
CATEGORY_CHOICES = [
('communications', 'Коммуникации'),
('documents', 'Документы'),
('requests', 'Заявления'),
('calendar', 'Календарь'),
('department', 'Отдел'),
('profile', 'Профиль'),
('feed', 'Новости'),
('system', 'Система'),
]
def unread(self):
"""Непрочитанные уведомления"""
return self.filter(unread=True)
code = models.CharField(
max_length=50,
unique=True,
choices=CATEGORY_CHOICES,
verbose_name='Код'
)
name = models.CharField(max_length=100, verbose_name='Название')
description = models.TextField(blank=True, verbose_name='Описание')
icon = models.CharField(
max_length=50,
default='bi-bell',
verbose_name='Иконка',
help_text='Bootstrap icon класс'
)
color = models.CharField(
max_length=20,
default='primary',
verbose_name='Цвет',
help_text='Bootstrap цвет'
)
is_active = models.BooleanField(default=True, verbose_name='Активна')
order = models.IntegerField(default=0, verbose_name='Порядок')
def read(self):
"""Прочитанные уведомления"""
return self.filter(unread=False)
class Meta:
ordering = ['order', 'name']
verbose_name = 'Категория уведомлений'
verbose_name_plural = 'Категории уведомлений'
db_table = 'notifications_category'
def mark_all_as_read(self, recipient=None):
"""Отметить все как прочитанные"""
qs = self.unread()
if recipient:
qs = qs.filter(recipient=recipient)
return qs.update(
unread=False,
timestamp_read=timezone.now()
)
def __str__(self):
return self.name
class NotificationType(models.Model):
"""Конкретные типы уведомлений внутри категорий"""
def mark_all_as_unread(self, recipient=None):
"""Отметить все как непрочитанные"""
qs = self.read()
if recipient:
qs = qs.filter(recipient=recipient)
return qs.update(
unread=True,
timestamp_read=None
)
category = models.ForeignKey(
NotificationCategory,
on_delete=models.CASCADE,
related_name='types',
verbose_name='Категория'
)
code = models.CharField(
max_length=100,
unique=True,
verbose_name='Код',
help_text='Уникальный код типа, например: chat_new_message'
)
name = models.CharField(max_length=200, verbose_name='Название')
description = models.TextField(blank=True, verbose_name='Описание')
def deleted(self):
"""Удаленные уведомления"""
return self.filter(deleted=True)
# Настройки по умолчанию для всех пользователей
default_enabled = models.BooleanField(
default=True,
verbose_name='Включено по умолчанию'
)
default_channels = models.JSONField(
default=dict,
verbose_name='Каналы по умолчанию',
help_text='{"web": true, "email": false, "telegram": false}'
)
def active(self):
"""Активные (не удаленные) уведомления"""
return self.filter(deleted=False)
# Приоритет
PRIORITY_CHOICES = [
('low', 'Низкий'),
('normal', 'Обычный'),
('high', 'Высокий'),
('urgent', 'Срочный'),
]
priority = models.CharField(
max_length=20,
choices=PRIORITY_CHOICES,
default='normal',
verbose_name='Приоритет'
)
def public(self):
"""Публичные уведомления"""
return self.filter(public=True)
# Можно ли отключить
is_required = models.BooleanField(
default=False,
verbose_name='Обязательное',
help_text='Нельзя отключить (например, системные уведомления)'
)
# Группировка
is_groupable = models.BooleanField(
default=True,
verbose_name='Можно группировать',
help_text='Объединять несколько похожих уведомлений'
)
grouping_window_minutes = models.IntegerField(
default=5,
verbose_name='Окно группировки (минуты)',
help_text='Группировать уведомления за последние N минут'
)
is_active = models.BooleanField(default=True, verbose_name='Активен')
created_at = models.DateTimeField(auto_now_add=True, verbose_name='Создан')
updated_at = models.DateTimeField(auto_now=True, verbose_name='Обновлен')
class Meta:
ordering = ['category__order', 'name']
verbose_name = 'Тип уведомления'
verbose_name_plural = 'Типы уведомлений'
db_table = 'notifications_type'
def __str__(self):
return f'{self.category.name}: {self.name}'
def for_user(self, user):
"""Уведомления для конкретного пользователя"""
return self.filter(recipient=user).active()
class Notification(models.Model):
"""Конкретное уведомление для пользователя"""
"""
Универсальная модель уведомления
Структура по образу django-notifications-hq:
- actor performed verb on action_object at target
Примеры:
- John (actor) commented (verb) on photo (target)
- Mary (actor) liked (verb) your comment (action_object) on post (target)
- System (actor) approved (verb) your request (action_object)
"""
# === Базовые поля ===
recipient = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='notifications',
verbose_name='Получатель'
)
notification_type = models.ForeignKey(
NotificationType,
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
verbose_name='Тип'
related_name='notifications',
verbose_name='Получатель',
help_text='Кому адресовано уведомление'
)
# Заголовок и текст
title = models.CharField(max_length=255, verbose_name='Заголовок')
message = models.TextField(verbose_name='Сообщение')
short_message = models.CharField(
max_length=150,
# Кто совершил действие (GenericForeignKey)
actor_content_type = models.ForeignKey(
ContentType,
related_name='notify_actor',
on_delete=models.CASCADE,
null=True,
blank=True,
verbose_name='Краткое сообщение',
help_text='Для превью'
verbose_name='Тип актора'
)
actor_object_id = models.CharField(
max_length=255,
null=True,
blank=True,
verbose_name='ID актора'
)
actor = GenericForeignKey('actor_content_type', 'actor_object_id')
# Связь с объектом (GenericForeignKey)
content_type = models.ForeignKey(
ContentType,
on_delete=models.CASCADE,
null=True,
blank=True,
verbose_name='Тип объекта'
)
object_id = models.PositiveIntegerField(
null=True,
blank=True,
verbose_name='ID объекта'
)
content_object = GenericForeignKey('content_type', 'object_id')
# Метаданные
metadata = models.JSONField(
default=dict,
blank=True,
verbose_name='Метаданные',
help_text='Дополнительные данные для рендеринга'
)
# Ссылка действия
action_url = models.CharField(
max_length=500,
blank=True,
verbose_name='Ссылка действия'
)
action_text = models.CharField(
max_length=100,
default='Посмотреть',
verbose_name='Текст кнопки'
)
# Статусы
is_read = models.BooleanField(
default=False,
verbose_name='Прочитано',
# Что произошло
verb = models.CharField(
max_length=255,
verbose_name='Действие',
help_text='Например: liked, commented, approved, mentioned',
db_index=True
)
read_at = models.DateTimeField(
null=True,
# Объект действия (GenericForeignKey) - опционально
action_object_content_type = models.ForeignKey(
ContentType,
related_name='notify_action_object',
on_delete=models.CASCADE,
null=True,
blank=True,
verbose_name='Прочитано когда'
verbose_name='Тип объекта действия'
)
is_archived = models.BooleanField(
default=False,
verbose_name='В архиве'
)
archived_at = models.DateTimeField(
null=True,
action_object_object_id = models.CharField(
max_length=255,
null=True,
blank=True,
verbose_name='Заархивировано когда'
verbose_name='ID объекта действия'
)
action_object = GenericForeignKey(
'action_object_content_type',
'action_object_object_id'
)
# Доставка
sent_web = models.BooleanField(default=False, verbose_name='Отправлено на сайт')
sent_email = models.BooleanField(default=False, verbose_name='Отправлено на email')
sent_telegram = models.BooleanField(default=False, verbose_name='Отправлено в Telegram')
sent_whatsapp = models.BooleanField(default=False, verbose_name='Отправлено в WhatsApp')
sent_wechat = models.BooleanField(default=False, verbose_name='Отправлено в WeChat')
sent_at = models.DateTimeField(
null=True,
# Целевой объект (GenericForeignKey) - опционально
target_content_type = models.ForeignKey(
ContentType,
related_name='notify_target',
on_delete=models.CASCADE,
null=True,
blank=True,
verbose_name='Отправлено когда'
verbose_name='Тип целевого объекта'
)
target_object_id = models.CharField(
max_length=255,
null=True,
blank=True,
verbose_name='ID целевого объекта'
)
target = GenericForeignKey('target_content_type', 'target_object_id')
# === Контент ===
description = models.TextField(
blank=True,
null=True,
verbose_name='Описание',
help_text='Человекочитаемое описание уведомления'
)
# Группировка
group_key = models.CharField(
max_length=255,
blank=True,
# URL для перехода
action_url = models.CharField(
max_length=500,
blank=True,
null=True,
verbose_name='URL действия',
help_text='Куда перейти при клике на уведомление'
)
# Дополнительные данные (JSON)
data = models.JSONField(
default=dict,
blank=True,
verbose_name='Дополнительные данные',
help_text='Любые метаданные для рендеринга или логики'
)
# === Статусы ===
unread = models.BooleanField(
default=True,
db_index=True,
verbose_name='Ключ группировки',
help_text='Для объединения похожих уведомлений'
)
is_grouped = models.BooleanField(
default=False,
verbose_name='Сгруппировано'
)
grouped_count = models.IntegerField(
default=1,
verbose_name='Количество в группе',
help_text='Сколько уведомлений объединено'
verbose_name='Непрочитано'
)
created_at = models.DateTimeField(
auto_now_add=True,
timestamp_read = models.DateTimeField(
null=True,
blank=True,
verbose_name='Время прочтения'
)
public = models.BooleanField(
default=True,
db_index=True,
verbose_name='Публичное',
help_text='Можно показывать другим пользователям'
)
deleted = models.BooleanField(
default=False,
db_index=True,
verbose_name='Удалено',
help_text='Мягкое удаление'
)
emailed = models.BooleanField(
default=False,
db_index=True,
verbose_name='Отправлено на email'
)
# === Временные метки ===
timestamp = models.DateTimeField(
default=timezone.now,
db_index=True,
verbose_name='Создано'
)
updated_at = models.DateTimeField(auto_now=True, verbose_name='Обновлено')
# === Manager ===
objects = NotificationQuerySet.as_manager()
class Meta:
ordering = ['-created_at']
ordering = ['-timestamp']
indexes = [
models.Index(fields=['recipient', '-created_at']),
models.Index(fields=['recipient', 'is_read']),
models.Index(fields=['notification_type', '-created_at']),
models.Index(fields=['group_key', '-created_at']),
models.Index(fields=['recipient', '-timestamp']),
models.Index(fields=['recipient', 'unread', '-timestamp']),
models.Index(fields=['verb', '-timestamp']),
models.Index(fields=['-timestamp']),
]
verbose_name = 'Уведомление'
verbose_name_plural = 'Уведомления'
db_table = 'notifications_notification'
db_table = 'notifications_notification_v2' # Новая таблица для миграции
def __str__(self):
return f'{self.recipient.get_short_name()}: {self.title}'
actor_str = str(self.actor) if self.actor else 'Система'
target_str = f' → {self.target}' if self.target else ''
return f'{actor_str} {self.verb}{target_str}'
# === Методы ===
def mark_as_read(self):
"""Отметить как прочитанное"""
if not self.is_read:
self.is_read = True
self.read_at = timezone.now()
self.save(update_fields=['is_read', 'read_at'])
"""Отметить уведомление как прочитанное"""
if self.unread:
self.unread = False
self.timestamp_read = timezone.now()
self.save(update_fields=['unread', 'timestamp_read'])
def archive(self):
"""Заархивировать уведомление"""
if not self.is_archived:
self.is_archived = True
self.archived_at = timezone.now()
self.save(update_fields=['is_archived', 'archived_at'])
def mark_as_unread(self):
"""Отметить уведомление как непрочитанное"""
if not self.unread:
self.unread = True
self.timestamp_read = None
self.save(update_fields=['unread', 'timestamp_read'])
@property
def timesince(self, now=None):
"""
Возвращает человекочитаемое время с момента создания
Например: "5 минут назад", "2 часа назад"
"""
from django.utils.timesince import timesince as timesince_
return timesince_(self.timestamp, now)
@property
def slug(self):
"""
Создает slug для уведомления (для URL)
Формат: id-verb-target
"""
return f"{self.id or ''}-{self.verb}-{self.target_object_id or ''}"
class UserNotificationSettings(models.Model):
"""Персональные настройки уведомлений пользователя"""
class UserChannelPreferences(models.Model):
"""
Настройки каналов доставки для пользователя
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='notification_settings',
Упрощенная модель вместо UserNotificationSettings
Один пользователь - одна запись с настройками для всех каналов
"""
user = models.OneToOneField(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='channel_preferences',
verbose_name='Пользователь'
)
notification_type = models.ForeignKey(
NotificationType,
on_delete=models.CASCADE,
verbose_name='Тип уведомления'
# === Каналы доставки ===
web_enabled = models.BooleanField(
default=True,
verbose_name='Веб уведомления',
help_text='Показывать в интерфейсе сайта'
)
# Включено ли уведомление
is_enabled = models.BooleanField(
default=True,
verbose_name='Включено'
)
# Каналы доставки
send_web = models.BooleanField(
default=True,
verbose_name='Отправлять на сайт'
)
send_email = models.BooleanField(
email_enabled = models.BooleanField(
default=False,
verbose_name='Отправлять на email'
verbose_name='Email уведомления'
)
# Частота email рассылки
push_enabled = models.BooleanField(
default=False,
verbose_name='Push уведомления',
help_text='Web Push для браузера'
)
# === Email настройки ===
EMAIL_FREQUENCY_CHOICES = [
('instant', 'Мгновенно'),
('daily', 'Ежедневный дайджест'),
('weekly', 'Еженедельный дайджест'),
('never', 'Никогда'),
]
email_frequency = models.CharField(
max_length=20,
choices=EMAIL_FREQUENCY_CHOICES,
default='instant',
verbose_name='Частота email рассылки',
help_text='Как часто отправлять уведомления на email (только если email включен)'
verbose_name='Частота email'
)
send_telegram = models.BooleanField(
# === Тихий режим (Do Not Disturb) ===
dnd_enabled = models.BooleanField(
default=False,
verbose_name='Отправлять в Telegram'
)
send_whatsapp = models.BooleanField(
default=False,
verbose_name='Отправлять в WhatsApp'
)
send_wechat = models.BooleanField(
default=False,
verbose_name='Отправлять в WeChat'
verbose_name='Режим "Не беспокоить"',
help_text='Отключить все уведомления в определенное время'
)
# Тихий режим (не беспокоить)
quiet_hours_enabled = models.BooleanField(
default=False,
verbose_name='Тихий режим включен'
)
quiet_start_time = models.TimeField(
null=True,
dnd_start_time = models.TimeField(
null=True,
blank=True,
verbose_name='Начало тихого режима',
help_text='Например, 22:00'
)
quiet_end_time = models.TimeField(
null=True,
dnd_end_time = models.TimeField(
null=True,
blank=True,
verbose_name='Конец тихого режима',
help_text='Например, 08:00'
)
created_at = models.DateTimeField(auto_now_add=True, verbose_name='Создано')
updated_at = models.DateTimeField(auto_now=True, verbose_name='Обновлено')
# === Фильтры по типам (verb) ===
class Meta:
unique_together = ['user', 'notification_type']
verbose_name = 'Настройка уведомлений пользователя'
verbose_name_plural = 'Настройки уведомлений пользователей'
db_table = 'notifications_user_settings'
def __str__(self):
return f'{self.user.get_short_name()}: {self.notification_type.name}'
class NotificationTemplate(models.Model):
"""Шаблоны для разных каналов доставки"""
notification_type = models.ForeignKey(
NotificationType,
on_delete=models.CASCADE,
related_name='templates',
verbose_name='Тип уведомления'
)
CHANNEL_CHOICES = [
('web', 'Веб'),
('email', 'Email'),
('telegram', 'Telegram'),
('whatsapp', 'WhatsApp'),
('wechat', 'WeChat'),
]
channel = models.CharField(
max_length=20,
choices=CHANNEL_CHOICES,
verbose_name='Канал'
)
# Шаблоны (используем Django template syntax)
title_template = models.TextField(
verbose_name='Шаблон заголовка',
help_text='Django template синтаксис'
)
message_template = models.TextField(
verbose_name='Шаблон сообщения',
help_text='Django template синтаксис'
)
# Для email
html_template = models.TextField(
disabled_verbs = models.JSONField(
default=list,
blank=True,
verbose_name='HTML шаблон',
help_text='Для email'
verbose_name='Отключенные типы',
help_text='Список verb для которых не показывать уведомления. Например: ["liked", "followed"]'
)
# Для кнопок действий
action_button_template = models.CharField(
max_length=200,
blank=True,
verbose_name='Шаблон кнопки действия'
)
# === Временные метки ===
is_active = models.BooleanField(default=True, verbose_name='Активен')
created_at = models.DateTimeField(auto_now_add=True, verbose_name='Создан')
updated_at = models.DateTimeField(auto_now=True, verbose_name='Обновлен')
class Meta:
unique_together = ['notification_type', 'channel']
verbose_name = 'Шаблон уведомления'
verbose_name_plural = 'Шаблоны уведомлений'
db_table = 'notifications_template'
def __str__(self):
return f'{self.notification_type.name} - {self.get_channel_display()}'
class WebPushSubscription(models.Model):
"""
Подписка пользователя на Web Push уведомления.
Хранит данные, полученные от браузера при подписке через Push API.
"""
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='web_push_subscriptions',
verbose_name='Пользователь'
)
# Endpoint - URL для отправки push-сообщений (от браузера)
endpoint = models.URLField(
max_length=512,
verbose_name='Push Endpoint',
help_text='URL для отправки push уведомлений'
)
# Ключ авторизации от браузера
auth_key = models.CharField(
max_length=512,
verbose_name='Auth Key',
help_text='Ключ авторизации (subscription.keys.auth)'
)
# P-256 Diffie-Hellman ключ
p256dh_key = models.CharField(
max_length=512,
verbose_name='P256DH Key',
help_text='Ключ шифрования (subscription.keys.p256dh)'
)
# Информация о браузере/устройстве
device_name = models.CharField(
max_length=255,
blank=True,
default='',
verbose_name='Название устройства',
help_text='iOS Device, Windows (Chrome), macOS (Safari), и т.д.'
)
user_agent = models.TextField(
blank=True,
verbose_name='User Agent',
help_text='Браузер и ОС'
)
# Активность
is_active = models.BooleanField(
default=True,
verbose_name='Активна'
)
# Timestamps
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name='Создана'
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name='Обновлена'
)
last_used_at = models.DateTimeField(
null=True,
blank=True,
verbose_name='Последнее использование'
verbose_name='Создано'
)
# Статистика ошибок
error_count = models.IntegerField(
default=0,
verbose_name='Счётчик ошибок'
)
last_error = models.TextField(
blank=True,
verbose_name='Последняя ошибка'
updated_at = models.DateTimeField(
auto_now=True,
verbose_name='Обновлено'
)
class Meta:
verbose_name = 'Web Push подписка'
verbose_name_plural = 'Web Push подписки'
db_table = 'notifications_web_push_subscription'
# Уникальность по endpoint (один браузер = одна подписка)
unique_together = ['user', 'endpoint']
ordering = ['-created_at']
verbose_name = 'Настройки каналов пользователя'
verbose_name_plural = 'Настройки каналов пользователей'
db_table = 'notifications_user_channel_preferences'
def __str__(self):
return f'WebPush: {self.user.email} ({self.endpoint[:50]}...)'
return f'{self.user.get_short_name()} - настройки каналов'
def mark_used(self):
"""Обновить время последнего использования"""
self.last_used_at = timezone.now()
self.save(update_fields=['last_used_at'])
def is_verb_enabled(self, verb):
"""Проверить, включен ли этот тип уведомлений"""
return verb not in self.disabled_verbs
def increment_error(self, error_message: str = ''):
"""Инкремент счётчика ошибок"""
self.error_count += 1
self.last_error = error_message[:1000] if error_message else ''
self.save(update_fields=['error_count', 'last_error'])
def disable_verb(self, verb):
"""Отключить определенный тип уведомлений"""
if verb not in self.disabled_verbs:
self.disabled_verbs.append(verb)
self.save(update_fields=['disabled_verbs'])
def enable_verb(self, verb):
"""Включить определенный тип уведомлений"""
if verb in self.disabled_verbs:
self.disabled_verbs.remove(verb)
self.save(update_fields=['disabled_verbs'])
def is_in_dnd_period(self):
"""Проверить, находимся ли в режиме "Не беспокоить"""
if not self.dnd_enabled or not self.dnd_start_time or not self.dnd_end_time:
return False
# Деактивировать при 5+ ошибках
if self.error_count >= 5:
self.is_active = False
self.save(update_fields=['is_active'])
def reset_errors(self):
"""Сбросить счётчик ошибок при успешной доставке"""
if self.error_count > 0:
self.error_count = 0
self.last_error = ''
self.save(update_fields=['error_count', 'last_error'])
now = timezone.now().time()
# Простой случай: начало < конец (например, 09:00 - 18:00)
if self.dnd_start_time < self.dnd_end_time:
return self.dnd_start_time <= now <= self.dnd_end_time
# Через полночь: начало > конец (например, 22:00 - 08:00)
return now >= self.dnd_start_time or now <= self.dnd_end_time
# === Web Push устройства управляются через django-push-notifications ===
# Используйте: from push_notifications.models import WebPushDevice

411
backend/notifications/models_new.py
View File

@ -1,411 +0,0 @@
"""
Упрощенная универсальная система уведомлений
Основана на архитектуре django-notifications-hq, адаптирована для Django 5.2+
Ключевые изменения:
- Одна модель Notification вместо 6
- GenericForeignKey для универсальности
- Простой API через сигналы
- Multi-channel поддержка (Web, Email, Push)
"""
from django.db import models
from django.conf import settings
from django.utils import timezone
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.db.models.query import QuerySet
class NotificationQuerySet(QuerySet):
"""QuerySet с удобными методами фильтрации"""
def unread(self):
"""Непрочитанные уведомления"""
return self.filter(unread=True)
def read(self):
"""Прочитанные уведомления"""
return self.filter(unread=False)
def mark_all_as_read(self, recipient=None):
"""Отметить все как прочитанные"""
qs = self.unread()
if recipient:
qs = qs.filter(recipient=recipient)
return qs.update(
unread=False,
timestamp_read=timezone.now()
)
def mark_all_as_unread(self, recipient=None):
"""Отметить все как непрочитанные"""
qs = self.read()
if recipient:
qs = qs.filter(recipient=recipient)
return qs.update(
unread=True,
timestamp_read=None
)
def deleted(self):
"""Удаленные уведомления"""
return self.filter(deleted=True)
def active(self):
"""Активные (не удаленные) уведомления"""
return self.filter(deleted=False)
def public(self):
"""Публичные уведомления"""
return self.filter(public=True)
def for_user(self, user):
"""Уведомления для конкретного пользователя"""
return self.filter(recipient=user).active()
class Notification(models.Model):
"""
Универсальная модель уведомления
Структура по образу django-notifications-hq:
- actor performed verb on action_object at target
Примеры:
- John (actor) commented (verb) on photo (target)
- Mary (actor) liked (verb) your comment (action_object) on post (target)
- System (actor) approved (verb) your request (action_object)
"""
# === Базовые поля ===
recipient = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='notifications',
verbose_name='Получатель',
help_text='Кому адресовано уведомление'
)
# Кто совершил действие (GenericForeignKey)
actor_content_type = models.ForeignKey(
ContentType,
related_name='notify_actor',
on_delete=models.CASCADE,
null=True,
blank=True,
verbose_name='Тип актора'
)
actor_object_id = models.CharField(
max_length=255,
null=True,
blank=True,
verbose_name='ID актора'
)
actor = GenericForeignKey('actor_content_type', 'actor_object_id')
# Что произошло
verb = models.CharField(
max_length=255,
verbose_name='Действие',
help_text='Например: liked, commented, approved, mentioned',
db_index=True
)
# Объект действия (GenericForeignKey) - опционально
action_object_content_type = models.ForeignKey(
ContentType,
related_name='notify_action_object',
on_delete=models.CASCADE,
null=True,
blank=True,
verbose_name='Тип объекта действия'
)
action_object_object_id = models.CharField(
max_length=255,
null=True,
blank=True,
verbose_name='ID объекта действия'
)
action_object = GenericForeignKey(
'action_object_content_type',
'action_object_object_id'
)
# Целевой объект (GenericForeignKey) - опционально
target_content_type = models.ForeignKey(
ContentType,
related_name='notify_target',
on_delete=models.CASCADE,
null=True,
blank=True,
verbose_name='Тип целевого объекта'
)
target_object_id = models.CharField(
max_length=255,
null=True,
blank=True,
verbose_name='ID целевого объекта'
)
target = GenericForeignKey('target_content_type', 'target_object_id')
# === Контент ===
description = models.TextField(
blank=True,
null=True,
verbose_name='Описание',
help_text='Человекочитаемое описание уведомления'
)
# URL для перехода
action_url = models.CharField(
max_length=500,
blank=True,
null=True,
verbose_name='URL действия',
help_text='Куда перейти при клике на уведомление'
)
# Дополнительные данные (JSON)
data = models.JSONField(
default=dict,
blank=True,
verbose_name='Дополнительные данные',
help_text='Любые метаданные для рендеринга или логики'
)
# === Статусы ===
unread = models.BooleanField(
default=True,
db_index=True,
verbose_name='Непрочитано'
)
timestamp_read = models.DateTimeField(
null=True,
blank=True,
verbose_name='Время прочтения'
)
public = models.BooleanField(
default=True,
db_index=True,
verbose_name='Публичное',
help_text='Можно показывать другим пользователям'
)
deleted = models.BooleanField(
default=False,
db_index=True,
verbose_name='Удалено',
help_text='Мягкое удаление'
)
emailed = models.BooleanField(
default=False,
db_index=True,
verbose_name='Отправлено на email'
)
# === Временные метки ===
timestamp = models.DateTimeField(
default=timezone.now,
db_index=True,
verbose_name='Создано'
)
# === Manager ===
objects = NotificationQuerySet.as_manager()
class Meta:
ordering = ['-timestamp']
indexes = [
models.Index(fields=['recipient', '-timestamp']),
models.Index(fields=['recipient', 'unread', '-timestamp']),
models.Index(fields=['verb', '-timestamp']),
models.Index(fields=['-timestamp']),
]
verbose_name = 'Уведомление'
verbose_name_plural = 'Уведомления'
db_table = 'notifications_notification_v2' # Новая таблица для миграции
def __str__(self):
actor_str = str(self.actor) if self.actor else 'Система'
target_str = f' → {self.target}' if self.target else ''
return f'{actor_str} {self.verb}{target_str}'
# === Методы ===
def mark_as_read(self):
"""Отметить уведомление как прочитанное"""
if self.unread:
self.unread = False
self.timestamp_read = timezone.now()
self.save(update_fields=['unread', 'timestamp_read'])
def mark_as_unread(self):
"""Отметить уведомление как непрочитанное"""
if not self.unread:
self.unread = True
self.timestamp_read = None
self.save(update_fields=['unread', 'timestamp_read'])
@property
def timesince(self, now=None):
"""
Возвращает человекочитаемое время с момента создания
Например: "5 минут назад", "2 часа назад"
"""
from django.utils.timesince import timesince as timesince_
return timesince_(self.timestamp, now)
@property
def slug(self):
"""
Создает slug для уведомления (для URL)
Формат: id-verb-target
"""
return f"{self.id or ''}-{self.verb}-{self.target_object_id or ''}"
class UserChannelPreferences(models.Model):
"""
Настройки каналов доставки для пользователя
Упрощенная модель вместо UserNotificationSettings
Один пользователь - одна запись с настройками для всех каналов
"""
user = models.OneToOneField(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='channel_preferences',
verbose_name='Пользователь'
)
# === Каналы доставки ===
web_enabled = models.BooleanField(
default=True,
verbose_name='Веб уведомления',
help_text='Показывать в интерфейсе сайта'
)
email_enabled = models.BooleanField(
default=False,
verbose_name='Email уведомления'
)
push_enabled = models.BooleanField(
default=False,
verbose_name='Push уведомления',
help_text='Web Push для браузера'
)
# === Email настройки ===
EMAIL_FREQUENCY_CHOICES = [
('instant', 'Мгновенно'),
('daily', 'Ежедневный дайджест'),
('weekly', 'Еженедельный дайджест'),
('never', 'Никогда'),
]
email_frequency = models.CharField(
max_length=20,
choices=EMAIL_FREQUENCY_CHOICES,
default='instant',
verbose_name='Частота email'
)
# === Тихий режим (Do Not Disturb) ===
dnd_enabled = models.BooleanField(
default=False,
verbose_name='Режим "Не беспокоить"',
help_text='Отключить все уведомления в определенное время'
)
dnd_start_time = models.TimeField(
null=True,
blank=True,
verbose_name='Начало тихого режима',
help_text='Например, 22:00'
)
dnd_end_time = models.TimeField(
null=True,
blank=True,
verbose_name='Конец тихого режима',
help_text='Например, 08:00'
)
# === Фильтры по типам (verb) ===
disabled_verbs = models.JSONField(
default=list,
blank=True,
verbose_name='Отключенные типы',
help_text='Список verb для которых не показывать уведомления. Например: ["liked", "followed"]'
)
# === Временные метки ===
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name='Создано'
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name='Обновлено'
)
class Meta:
verbose_name = 'Настройки каналов пользователя'
verbose_name_plural = 'Настройки каналов пользователей'
db_table = 'notifications_user_channel_preferences'
def __str__(self):
return f'{self.user.get_short_name()} - настройки каналов'
def is_verb_enabled(self, verb):
"""Проверить, включен ли этот тип уведомлений"""
return verb not in self.disabled_verbs
def disable_verb(self, verb):
"""Отключить определенный тип уведомлений"""
if verb not in self.disabled_verbs:
self.disabled_verbs.append(verb)
self.save(update_fields=['disabled_verbs'])
def enable_verb(self, verb):
"""Включить определенный тип уведомлений"""
if verb in self.disabled_verbs:
self.disabled_verbs.remove(verb)
self.save(update_fields=['disabled_verbs'])
def is_in_dnd_period(self):
"""Проверить, находимся ли в режиме "Не беспокоить"""
if not self.dnd_enabled or not self.dnd_start_time or not self.dnd_end_time:
return False
now = timezone.now().time()
# Простой случай: начало < конец (например, 09:00 - 18:00)
if self.dnd_start_time < self.dnd_end_time:
return self.dnd_start_time <= now <= self.dnd_end_time
# Через полночь: начало > конец (например, 22:00 - 08:00)
return now >= self.dnd_start_time or now <= self.dnd_end_time
# === Web Push устройства управляются через django-push-notifications ===
# Используйте: from push_notifications.models import WebPushDevice

766
backend/notifications/services.py
View File

@ -1,146 +1,26 @@
"""
Сервисный слой для работы с уведомлениями
Слой обратной совместимости для старого API уведомлений.
DEPRECATED: Этот модуль предоставляет facade для старого кода.
Новый код должен использовать напрямую:
- from notifications.signals import notify
- notify.send(sender=user, recipient=other, verb='liked', ...)
"""
import json
import logging
from typing import Any, Dict, Optional
from typing import Optional, Dict, Any
from django.contrib.auth import get_user_model
from django.utils import timezone
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
from .models import (
Notification,
NotificationType,
UserNotificationSettings,
)
# DEPRECATED: Old senders, use new architecture in senders/
# from .email_sender import EmailNotificationSender
# from .telegram_sender import TelegramNotificationSender
# New senders available in notifications.senders module
from .senders import EmailNotificationSender
# Telegram support removed, use new notification system
TelegramNotificationSender = None # Placeholder for backward compatibility
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,
) -> Optional[Notification]:
"""
Создать уведомление асинхронно через Celery (рекомендуется)
Args:
recipient: Получатель уведомления
notification_type_code: Код типа уведомления
title: Заголовок
message: Текст сообщения
content_object: Связанный объект (опционально)
action_url: URL для действия
action_text: Текст кнопки действия
metadata: Дополнительные данные
Returns:
Созданное уведомление (если sync fallback) или None (если async)
"""
try:
# Проверяем доступность Celery
if not is_celery_available():
logger.warning(
f"[create_notification_async] Celery недоступен, fallback на синхронный режим"
)
return 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,
)
# Импортируем задачу
from notifications.tasks import send_notification_task
# Отправляем задачу в Celery
result = 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}"
)
# В EAGER режиме (тесты) задача выполняется синхронно
# Пытаемся получить созданное уведомление
try:
from django.conf import settings
if getattr(settings, 'CELERY_TASK_ALWAYS_EAGER', False):
# В тестах пытаемся найти только что созданное уведомление
from notifications.models import Notification
notification = Notification.objects.filter(
recipient=recipient,
notification_type__code=notification_type_code,
title=title
).order_by('-created_at').first()
return notification
except Exception:
pass
return None
except Exception as e:
logger.exception(
f"[create_notification_async] ❌ Ошибка при отправке в Celery: {e}"
)
# Fallback на синхронный режим
return 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,
)
"""
Facade для обратной совместимости со старым API.
Все методы теперь работают через новую систему (notify.send()).
"""
@staticmethod
def create_notification(
recipient: User,
@ -152,520 +32,168 @@ class NotificationService:
action_text: str = 'Посмотреть',
metadata: Optional[Dict[str, Any]] = None,
send_immediately: bool = True,
) -> Optional[Notification]:
):
"""
Создать и отправить уведомление
Создает уведомление (старый API → новая система).
DEPRECATED: Используйте notify.send() напрямую.
Args:
recipient: Получатель уведомления
notification_type_code: Код типа уведомления
title: Заголовок
message: Текст сообщения
content_object: Связанный объект (опционально)
notification_type_code: Код типа (mapping в verb)
title: Заголовок (не используется в новой системе)
message: Описание уведомления
content_object: Объект-источник (→ action_object)
action_url: URL для действия
action_text: Текст кнопки действия
action_text: Текст кнопки (не используется)
metadata: Дополнительные данные
send_immediately: Отправить сразу или отложить
send_immediately: Игнорируется (всегда через Celery)
Returns:
Созданное уведомление или None если тип не найден
Notification объект (новая модель)
"""
logger.debug(
f"[NotificationService.create_notification] НАЧАЛО: "
f"recipient={recipient.id} ({recipient.email}), "
f"type={notification_type_code}, "
f"send_immediately={send_immediately}, "
f"title='{title[:50]}...'"
)
from .signals import notify
try:
notification_type = NotificationType.objects.get(
code=notification_type_code,
is_active=True
)
logger.debug(
f"[NotificationService.create_notification] Тип уведомления найден: "
f"{notification_type.name} (category={notification_type.category.code})"
)
except NotificationType.DoesNotExist:
logger.error(
f"[NotificationService.create_notification] ❌ ОШИБКА: "
f"Тип уведомления '{notification_type_code}' не найден или неактивен!"
)
return None
# Проверить настройки пользователя
settings = NotificationService.get_user_settings(
recipient, notification_type
)
# Маппинг старых типов на новые verbs
type_to_verb = {
'message': 'messaged',
'message_new': 'messaged',
'document': 'shared',
'document_new': 'shared',
'request': 'created',
'request_status_changed': 'updated',
'request_comment_added': 'commented',
'feed_post': 'posted',
'feed_comment': 'commented',
'feed_like': 'liked',
'calendar_event': 'scheduled',
'procurement': 'created',
'procurement_approved': 'approved',
'procurement_rejected': 'rejected',
}
logger.debug(
f"[NotificationService.create_notification] Настройки пользователя: "
f"enabled={settings.is_enabled}, "
f"web={settings.send_web}, "
f"email={settings.send_email}, "
f"telegram={settings.send_telegram}"
)
if not settings.is_enabled:
logger.warning(
f"[NotificationService.create_notification] ⏭️ ПРОПУСК: "
f"Уведомления отключены для user={recipient.id}"
)
return None
# Создать краткое сообщение
short_message = message[:150] if len(message) > 150 else message
# Создать уведомление
notification = Notification.objects.create(
verb = type_to_verb.get(notification_type_code, 'notified')
# Создаем через новую систему
notification = notify.send(
sender=None, # System notification
recipient=recipient,
notification_type=notification_type,
verb=verb,
action_object=content_object,
description=message,
action_url=action_url,
)
# notify.send() возвращает список [(signal, [notifications])]
if notification and len(notification) > 0:
return notification[0][1][0] if notification[0][1] else None
return None
@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,
):
"""
Асинхронное создание уведомления.
DEPRECATED: Новая система всегда асинхронна через Celery.
Этот метод теперь идентичен create_notification().
"""
return NotificationService.create_notification(
recipient=recipient,
notification_type_code=notification_type_code,
title=title,
message=message,
short_message=short_message,
content_object=content_object,
action_url=action_url,
action_text=action_text,
metadata=metadata or {},
metadata=metadata,
send_immediately=True,
)
logger.debug(
f"[NotificationService.create_notification] ✅ Уведомление создано: "
f"id={notification.id}, "
f"action_url={action_url}"
)
# Отправить если нужно
if send_immediately:
logger.debug(
f"[NotificationService.create_notification] ➡️ Немедленная отправка notification_id={notification.id}"
)
NotificationService.send_notification(notification, settings)
else:
logger.debug(
f"[NotificationService.create_notification] ⏸️ Отложенная отправка notification_id={notification.id}"
)
return notification
@staticmethod
def send_notification(
notification: Notification,
settings: Optional[UserNotificationSettings] = None
):
"""Отправить уведомление по всем каналам"""
logger.debug(
f"[NotificationService.send_notification] 📤 НАЧАЛО notification_id={notification.id} "
f"recipient={notification.recipient.id} type={notification.notification_type.code}"
)
if settings is None:
settings = NotificationService.get_user_settings(
notification.recipient,
notification.notification_type
)
logger.debug(
f"[NotificationService.send_notification] Активные каналы: "
f"{'✅' if settings.send_web else '❌'} Web, "
f"{'✅' if settings.send_email else '❌'} Email, "
f"{'✅' if settings.send_telegram else '❌'} Telegram"
)
# Отправить на веб (WebSocket) - для онлайн пользователей
if settings.send_web:
logger.debug(
f"[NotificationService.send_notification] 🌐 Отправка через WebSocket..."
)
try:
# Отправляем WebSocket только для онлайн уведомлений (не push)
NotificationService.send_web_socket(notification)
notification.sent_web = True
logger.debug(
f"[NotificationService.send_notification] ✅ WebSocket: успешно отправлено"
)
except Exception as e:
logger.error(
f"[NotificationService.send_notification] ❌ WebSocket: ошибка - {e}",
exc_info=True
)
else:
logger.debug(
f"[NotificationService.send_notification] ⏭️ WebSocket: канал отключен"
)
# Отправить Web Push (для offline пользователей)
# ВАЖНО: отправляем независимо от send_web настройки,
# если есть активные push-подписки
if settings.send_web:
logger.debug(
f"[NotificationService.send_notification] 📲 Отправка Web Push (offline)..."
)
try:
push_count = NotificationService.send_web_push_notification(notification)
if push_count > 0:
logger.debug(
f"[NotificationService.send_notification] ✅ Web Push: отправлено на {push_count} устройств"
)
except Exception as e:
logger.error(
f"[NotificationService.send_notification] ❌ Web Push: ошибка - {e}",
exc_info=True
)
else:
logger.debug(
f"[NotificationService.send_notification] ⏭️ Web Push: канал отключен"
)
# Отправить на email
if settings.send_email:
logger.debug(
f"[NotificationService.send_notification] 📧 Отправка Email..."
)
try:
recipient_email = notification.recipient.email
if recipient_email:
logger.debug(
f"[NotificationService.send_notification] Email адрес: {recipient_email}"
)
success = EmailNotificationSender.send_notification_email(
notification=notification,
recipient_email=recipient_email,
)
notification.sent_email = success
if success:
logger.debug(
f"[NotificationService.send_notification] ✅ Email: успешно отправлено на {recipient_email}"
)
else:
logger.warning(
f"[NotificationService.send_notification] ⚠️ Email: отправка вернула False"
)
else:
notification.sent_email = False
logger.warning(
f"[NotificationService.send_notification] ⚠️ Email: у пользователя {notification.recipient} нет email адреса"
)
except Exception as e:
notification.sent_email = False
logger.error(
f"[NotificationService.send_notification] ❌ Email: ошибка отправки - {e}",
exc_info=True
)
else:
logger.debug(
f"[NotificationService.send_notification] ⏭️ Email: канал отключен"
)
# Отправить в Telegram
if settings.send_telegram:
logger.debug(
f"[NotificationService.send_notification] 💬 Отправка в Telegram..."
)
logger.debug(
f"[NotificationService.send_notification] Получатель: user_id={notification.recipient.id}"
)
try:
# Проверяем наличие Telegram Chat ID в профиле
telegram_chat_id = getattr(
notification.recipient, 'telegram', ''
)
if telegram_chat_id:
telegram_chat_id = telegram_chat_id.strip()
logger.debug(
f"[NotificationService.send_notification] Telegram Chat ID: '{telegram_chat_id}'"
)
if telegram_chat_id:
# Получаем site_url для формирования полных ссылок
from django.conf import settings as django_settings
site_url = getattr(
django_settings,
'SITE_URL',
'http://localhost:9000'
)
logger.debug(
f"[NotificationService.send_notification] ⚠️ Telegram support deprecated - use new notification system"
)
# DEPRECATED: Telegram support moved to new architecture
# Use notifications.signals_new.notify() with new UserChannelPreferences
notification.sent_telegram = False
else:
notification.sent_telegram = False
logger.warning(
f"[NotificationService.send_notification] ⚠️ Telegram: у пользователя нет Chat ID"
)
except Exception as e:
notification.sent_telegram = False
logger.error(
f"[NotificationService.send_notification] ❌ Telegram: ошибка - {e}",
exc_info=True
)
else:
logger.debug(
f"[NotificationService.send_notification] ⏭️ Telegram: канал отключен"
)
# Обновить время отправки
notification.sent_at = timezone.now()
notification.save(update_fields=[
'sent_web',
'sent_email',
'sent_telegram',
'sent_whatsapp',
'sent_wechat',
'sent_at'
])
logger.debug(
f"[NotificationService.send_notification] 🎯 ЗАВЕРШЕНО notification_id={notification.id} "
f"Web={'✅' if notification.sent_web else '❌'} "
f"Email={'✅' if notification.sent_email else '❌'} "
f"Telegram={'✅' if notification.sent_telegram else '❌'}"
)
@staticmethod
def send_web_socket(notification: Notification):
"""
Отправить уведомление через WebSocket (только для онлайн пользователей).
Переименовано из send_web_notification для ясности.
"""
logger.debug(
f"[NotificationService.send_web_socket] НАЧАЛО: "
f"notification_id={notification.id}, "
f"recipient={notification.recipient.id}"
)
channel_layer = get_channel_layer()
notification_data = {
'id': notification.id,
'title': notification.title,
'message': notification.short_message,
'category': notification.notification_type.category.code,
'category_name': notification.notification_type.category.name,
'icon': notification.notification_type.category.icon,
'color': notification.notification_type.category.color,
'priority': notification.notification_type.priority,
'action_url': notification.action_url,
'action_text': notification.action_text,
'created_at': notification.created_at.isoformat(),
}
group_name = f'notifications_{notification.recipient.id}'
logger.debug(
f"[NotificationService.send_web_socket] Отправка в группу: {group_name}"
)
async_to_sync(channel_layer.group_send)(
group_name,
{
'type': 'notification_new',
'notification': notification_data
}
)
logger.debug(
f"[NotificationService.send_web_socket] ✅ Уведомление отправлено в WebSocket"
)
# Обновить счетчик
unread_count = Notification.objects.filter(
recipient=notification.recipient,
is_read=False,
is_archived=False
).count()
logger.debug(
f"[NotificationService.send_web_socket] Обновление счетчика: {unread_count} непрочитанных"
)
async_to_sync(channel_layer.group_send)(
group_name,
{
'type': 'notification_count_update',
'count': unread_count
}
)
logger.debug(
f"[NotificationService.send_web_socket] ✅ Счетчик обновлен"
)
@staticmethod
def send_web_push_notification(notification: Notification) -> int:
"""
Отправить push-уведомление через Web Push API используя django-push-notifications.
Работает даже когда браузер закрыт.
Args:
notification: Объект уведомления
Returns:
Количество успешно отправленных push-уведомлений
"""
from push_notifications.models import WebPushDevice
logger.debug(
f"[NotificationService.send_web_push_notification] НАЧАЛО: "
f"notification_id={notification.id}, "
f"recipient={notification.recipient.id}"
)
# Получаем все активные устройства пользователя
devices = WebPushDevice.objects.filter(
user=notification.recipient,
active=True
)
device_count = devices.count()
if device_count == 0:
logger.debug(
f"[NotificationService.send_web_push_notification] ℹ️ Нет активных push-устройств"
)
return 0
logger.debug(
f"[NotificationService.send_web_push_notification] Найдено устройств: {device_count}"
)
# Формируем payload для push-уведомления
payload = {
'title': notification.title,
'body': notification.short_message or notification.message[:150],
'tag': f'notification-{notification.id}',
'icon': '/static/img/logo-192.png',
'badge': '/static/img/badge-72.png',
'data': {
'url': notification.action_url or '/',
'notification_id': notification.id,
'category': notification.notification_type.category.code,
},
'requireInteraction': notification.notification_type.priority in ['high', 'urgent'],
}
# Отправляем через библиотеку (она автоматически обрабатывает батчинг и ошибки)
try:
devices.send_message(json.dumps(payload), ttl=86400)
success_count = device_count
logger.debug(
f"[NotificationService.send_web_push_notification] ✅ Завершено: "
f"{success_count} устройств уведомлены"
)
return success_count
except Exception as e:
logger.error(
f"[NotificationService.send_web_push_notification] ❌ Ошибка отправки: {e}"
)
return 0
@staticmethod
def get_user_settings(
user: User,
notification_type: NotificationType
) -> UserNotificationSettings:
"""Получить или создать настройки пользователя"""
settings, created = UserNotificationSettings.objects.get_or_create(
user=user,
notification_type=notification_type,
defaults={
'is_enabled': notification_type.default_enabled,
'send_web': notification_type.default_channels.get(
'web', True
),
'send_email': notification_type.default_channels.get(
'email', False
),
'send_telegram': notification_type.default_channels.get(
'telegram', False
),
'send_whatsapp': notification_type.default_channels.get(
'whatsapp', False
),
'send_wechat': notification_type.default_channels.get(
'wechat', False
),
}
)
logger.debug(
f"[Settings] User {user.id} type={notification_type.code} "
f"created={created} enabled={settings.is_enabled} "
f"web={settings.send_web} email={settings.send_email} "
f"telegram={settings.send_telegram}"
)
return settings
@staticmethod
def mark_as_read(notification_id: int, user: User) -> bool:
"""Отметить уведомление как прочитанное"""
"""
Отмечает уведомление как прочитанное.
Args:
notification_id: ID уведомления
user: Пользователь (проверка владения)
Returns:
True если успешно, False иначе
"""
try:
from .models import Notification
notification = Notification.objects.get(
id=notification_id,
recipient=user
)
notification.mark_as_read()
# Обновить счетчик через WebSocket
channel_layer = get_channel_layer()
unread_count = Notification.objects.filter(
recipient=user,
is_read=False,
is_archived=False
).count()
async_to_sync(channel_layer.group_send)(
f'notifications_{user.id}',
{
'type': 'notification_count_update',
'count': unread_count
}
)
return True
except Notification.DoesNotExist:
logger.warning(f"Notification {notification_id} not found for user {user.id}")
return False
@staticmethod
def mark_all_as_read(user: User, category: Optional[str] = None) -> int:
"""Отметить все уведомления как прочитанные"""
queryset = Notification.objects.filter(
"""
Отмечает все уведомления как прочитанные.
Args:
user: Пользователь
category: Категория (не используется в новой системе)
Returns:
Количество отмеченных уведомлений
"""
from .models import Notification
return Notification.objects.filter(
recipient=user,
is_read=False
unread=True
).mark_all_as_read()
@staticmethod
def get_user_settings(user: User, setting_type: Optional[str] = None):
"""
Получает настройки уведомлений пользователя.
DEPRECATED: Используйте UserChannelPreferences напрямую.
Args:
user: Пользователь
setting_type: Тип настройки (игнорируется)
Returns:
UserChannelPreferences объект
"""
try:
return user.channel_preferences
except Exception:
from .models import UserChannelPreferences
return UserChannelPreferences.objects.create(user=user)
@staticmethod
def send_websocket_notification(user_id: int, notification_data: Optional[dict] = None):
"""
Отправка WebSocket уведомления.
DEPRECATED: WebSocket уведомления отправляются автоматически
через channels.py при создании Notification.
"""
logger.warning(
"NotificationService.send_websocket_notification() is deprecated. "
"WebSocket notifications are sent automatically via channels.py"
)
if category:
queryset = queryset.filter(
notification_type__category__code=category
)
count = queryset.update(
is_read=True,
read_at=timezone.now()
)
# Обновить счетчик
channel_layer = get_channel_layer()
unread_count = Notification.objects.filter(
recipient=user,
is_read=False,
is_archived=False
).count()
async_to_sync(channel_layer.group_send)(
f'notifications_{user.id}',
{
'type': 'notification_count_update',
'count': unread_count
}
)
return count
return True

123
backend/notifications/signals.py
View File

@ -1,2 +1,121 @@
# Пустой файл для регистрации сигналов
# Сигналы будут добавлены на этапе 5 (интеграция с модулями)
"""
Сигнал notify для создания уведомлений
По образу django-notifications-hq
Использование:
from notifications.signals import notify
notify.send(
sender=user,
recipient=other_user,
verb='liked',
action_object=comment,
target=post,
description='liked your comment on "Post Title"',
action_url='/posts/123/',
)
"""
from django.dispatch import Signal
# Сигнал для создания уведомлений
notify = Signal()
def create_notification_handler(sender, **kwargs):
"""
Обработчик сигнала notify
Создает уведомление в БД
Параметры:
sender: Объект-актор (кто создал уведомление)
recipient: User или список User (кому)
verb: str - действие (обязательно)
action_object: Объект действия (опционально)
target: Целевой объект (опционально)
description: str - описание (опционально)
action_url: str - URL для перехода (опционально)
data: dict - дополнительные данные (опционально)
public: bool - публичное ли (по умолчанию True)
timestamp: datetime - время создания (по умолчанию now)
Возвращает:
Notification или список Notification
"""
from .models import Notification
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
# Извлекаем параметры
recipient = kwargs.pop('recipient', None)
verb = kwargs.pop('verb', None)
action_object = kwargs.pop('action_object', None)
target = kwargs.pop('target', None)
description = kwargs.pop('description', None)
action_url = kwargs.pop('action_url', None)
data = kwargs.pop('data', {})
public = kwargs.pop('public', True)
timestamp = kwargs.pop('timestamp', None)
# Валидация
if not recipient:
raise ValueError('recipient is required')
if not verb:
raise ValueError('verb is required')
# Поддержка множественных получателей
recipients = recipient if isinstance(recipient, (list, tuple)) else [recipient]
# Создаем уведомления
notifications = []
for recip in recipients:
if not isinstance(recip, User):
raise ValueError(f'recipient must be User instance, got {type(recip)}')
# Подготовка данных для GenericForeignKey
new_notification = Notification(
recipient=recip,
verb=verb,
description=description or '',
action_url=action_url or '',
data=data,
public=public,
)
# Actor
if sender:
new_notification.actor_content_type = ContentType.objects.get_for_model(sender)
new_notification.actor_object_id = sender.pk
# Action Object
if action_object:
new_notification.action_object_content_type = ContentType.objects.get_for_model(action_object)
new_notification.action_object_object_id = action_object.pk
# Target
if target:
new_notification.target_content_type = ContentType.objects.get_for_model(target)
new_notification.target_object_id = target.pk
# Timestamp
if timestamp:
new_notification.timestamp = timestamp
new_notification.save()
notifications.append(new_notification)
# Возвращаем одно уведомление или список
if len(notifications) == 1:
return notifications[0]
return notifications
# Подключаем обработчик
notify.connect(
create_notification_handler,
dispatch_uid='notifications.create_notification'
)

121
backend/notifications/signals_new.py
View File

@ -1,121 +0,0 @@
"""
Сигнал notify для создания уведомлений
По образу django-notifications-hq
Использование:
from notifications.signals import notify
notify.send(
sender=user,
recipient=other_user,
verb='liked',
action_object=comment,
target=post,
description='liked your comment on "Post Title"',
action_url='/posts/123/',
)
"""
from django.dispatch import Signal
# Сигнал для создания уведомлений
notify = Signal()
def create_notification_handler(sender, **kwargs):
"""
Обработчик сигнала notify
Создает уведомление в БД
Параметры:
sender: Объект-актор (кто создал уведомление)
recipient: User или список User (кому)
verb: str - действие (обязательно)
action_object: Объект действия (опционально)
target: Целевой объект (опционально)
description: str - описание (опционально)
action_url: str - URL для перехода (опционально)
data: dict - дополнительные данные (опционально)
public: bool - публичное ли (по умолчанию True)
timestamp: datetime - время создания (по умолчанию now)
Возвращает:
Notification или список Notification
"""
from .models_new import Notification
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
User = get_user_model()
# Извлекаем параметры
recipient = kwargs.pop('recipient', None)
verb = kwargs.pop('verb', None)
action_object = kwargs.pop('action_object', None)
target = kwargs.pop('target', None)
description = kwargs.pop('description', None)
action_url = kwargs.pop('action_url', None)
data = kwargs.pop('data', {})
public = kwargs.pop('public', True)
timestamp = kwargs.pop('timestamp', None)
# Валидация
if not recipient:
raise ValueError('recipient is required')
if not verb:
raise ValueError('verb is required')
# Поддержка множественных получателей
recipients = recipient if isinstance(recipient, (list, tuple)) else [recipient]
# Создаем уведомления
notifications = []
for recip in recipients:
if not isinstance(recip, User):
raise ValueError(f'recipient must be User instance, got {type(recip)}')
# Подготовка данных для GenericForeignKey
new_notification = Notification(
recipient=recip,
verb=verb,
description=description or '',
action_url=action_url or '',
data=data,
public=public,
)
# Actor
if sender:
new_notification.actor_content_type = ContentType.objects.get_for_model(sender)
new_notification.actor_object_id = sender.pk
# Action Object
if action_object:
new_notification.action_object_content_type = ContentType.objects.get_for_model(action_object)
new_notification.action_object_object_id = action_object.pk
# Target
if target:
new_notification.target_content_type = ContentType.objects.get_for_model(target)
new_notification.target_object_id = target.pk
# Timestamp
if timestamp:
new_notification.timestamp = timestamp
new_notification.save()
notifications.append(new_notification)
# Возвращаем одно уведомление или список
if len(notifications) == 1:
return notifications[0]
return notifications
# Подключаем обработчик
notify.connect(
create_notification_handler,
dispatch_uid='notifications.create_notification'
)

382
backend/notifications/tasks.py
View File

@ -1,382 +0,0 @@
"""
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"""
<div style="margin-bottom: 15px; padding: 10px; border-left: 3px solid #007bff;">
<strong>{notif.title}</strong><br>
<span style="color: #666;">{notif.message}</span><br>
<small style="color: #999;">{notif.created_at.strftime('%H:%M, %d.%m.%Y')}</small>
</div>
"""
html_content = f"""
<html>
<body style="font-family: Arial, sans-serif; padding: 20px;">
<h2>У вас {len(unread_notifications)} непрочитанных уведомлений</h2>
<div style="margin-top: 20px;">
{notification_html}
</div>
<p style="margin-top: 30px;">
<a href="{settings.BASE_URL}/notifications/" style="color: #007bff;">
Посмотреть все уведомления
</a>
</p>
</body>
</html>
"""
send_email(
subject=f"У вас {len(unread_notifications)} новых уведомлений",
to_email=user.email,
html_content=html_content,
)
logger.info(f"Digest email sent to user {user_id} with {len(unread_notifications)} notifications")
return {
"status": "success",
"user_id": user_id,
"notification_count": len(unread_notifications),
}
except User.DoesNotExist:
logger.error(f"User {user_id} not found for digest email")
return {"status": "error", "reason": "user_not_found"}
except Exception as exc:
logger.exception(f"Failed to send digest email to user {user_id}: {exc}")
raise self.retry(exc=exc)
@shared_task(
name="communications.process_message_notifications",
bind=True,
max_retries=3,
default_retry_delay=30,
)
def process_message_notifications_task(self, message_id: int):
"""
Обрабатывает ВСЕ уведомления для нового сообщения в чате асинхронно.
Это позволяет signal handler сразу вернуть управление,
а всю логику обработки (поиск упоминаний, проверка настроек, отправка)
выполнить в фоновом режиме.
Args:
message_id: ID сообщения
"""
from notifications.task_base import MessageNotificationProcessor
processor = MessageNotificationProcessor(task=self)
return processor.process(message_id)
@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', # 'created', 'updated', 'cancelled'
changed_fields: list = None
):
"""
Обрабатывает уведомления для событий календаря асинхронно.
Args:
event_id: ID события
action: Тип действия ('created', 'updated', 'cancelled')
changed_fields: Список изменённых полей (для updated)
"""
from notifications.task_base import EventNotificationProcessor
processor = EventNotificationProcessor(task=self)
return processor.process(event_id)
@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')
"""
from notifications.task_base import PostNotificationProcessor
processor = PostNotificationProcessor(task=self)
return processor.process(post_id)
@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', # 'created', 'approved', 'rejected', 'comment'
comment_id: int = None
):
"""
Обрабатывает уведомления для заявлений асинхронно.
Args:
request_id: ID заявления
action: Тип действия
comment_id: ID комментария (если action='comment')
"""
from notifications.task_base import RequestNotificationProcessor
processor = RequestNotificationProcessor(task=self)
# Определяем тип уведомления на основе action
notification_type_map = {
'created': 'created',
'approved': 'status_changed',
'rejected': 'status_changed',
'cancelled': 'status_changed',
'updated': 'status_changed',
'comment': 'comment_added',
}
return processor.process(request_id, notification_type_map.get(action, 'created'))
@shared_task(
name="documents.process_document_notifications",
bind=True,
max_retries=3,
default_retry_delay=30,
)
def process_document_notifications_task(self, document_id: int):
"""
Обрабатывает уведомления о новом документе асинхронно.
Args:
document_id: ID документа
"""
from notifications.task_base import DocumentNotificationProcessor
processor = DocumentNotificationProcessor(task=self)
return processor.process(document_id)

19
backend/notifications/tasks/__init__.py Normal file
View File

@ -0,0 +1,19 @@
"""
Celery задачи для асинхронной отправки уведомлений через различные каналы
Архитектура:
- Базовый класс BaseNotificationTask (инкапсуляция общей логики)
- Класс-задача для каждого канала (EmailNotificationTask, PushNotificationTask, ...)
- Автоматическая регистрация в Celery через register_task()
- Единообразное логирование, retry, rate limiting
"""
from .email import send_email_notification, send_digest_email
from .push import send_push_notification
from .websocket import send_websocket_notification
__all__ = [
'send_email_notification',
'send_push_notification',
'send_websocket_notification',
'send_digest_email',
]

138
backend/notifications/tasks/base.py Normal file
View File

@ -0,0 +1,138 @@
"""
Базовый класс для Celery задач отправки уведомлений
"""
from abc import ABC, abstractmethod
from celery import shared_task
import logging
logger = logging.getLogger(__name__)
class BaseNotificationTask(ABC):
"""
Абстрактный базовый класс для Celery задач по отправке уведомлений.
Инкапсулирует общую логику:
- Получение notification из БД
- Обработка ошибок
- Логирование
- Retry логика
Наследники реализуют только send_notification() с конкретным sender'ом.
"""
# Настройки Celery для конкретного канала (переопределяются в наследниках)
task_name = None
max_retries = 3
retry_delay = 60 # секунд
rate_limit = None # например '10/m' для email
def __init__(self):
self.logger = logging.getLogger(self.__class__.__name__)
@classmethod
def register_task(cls):
"""
Регистрирует Celery задачу для этого класса.
Должен вызываться при импорте модуля.
"""
instance = cls()
@shared_task(
name=cls.task_name,
bind=True,
max_retries=cls.max_retries,
default_retry_delay=cls.retry_delay,
autoretry_for=(Exception,),
rate_limit=cls.rate_limit,
)
def task_wrapper(self_task, notification_id: int, **kwargs):
"""Wrapper для вызова метода класса из Celery"""
return instance.execute(self_task, notification_id, **kwargs)
return task_wrapper
def execute(self, celery_task, notification_id: int, **kwargs):
"""
Выполняет задачу отправки уведомления.
Args:
celery_task: Объект Celery task (для retry)
notification_id: ID уведомления
**kwargs: Дополнительные параметры для sender
Returns:
True если успешно, False иначе
"""
try:
# Получаем уведомление из БД
notification = self.get_notification(notification_id)
if not notification:
self.logger.error(f"❌ Notification {notification_id} not found")
return False
# Отправляем через конкретный sender
success = self.send_notification(notification, **kwargs)
if success:
self.log_success(notification)
else:
self.log_failure(notification)
return success
except Exception as exc:
self.log_error(notification_id, exc)
raise celery_task.retry(exc=exc)
def get_notification(self, notification_id: int):
"""Получает Notification из БД"""
try:
from notifications.models import Notification
return Notification.objects.get(id=notification_id)
except Exception:
return None
@abstractmethod
def send_notification(self, notification, **kwargs) -> bool:
"""
Отправляет уведомление через конкретный канал.
Должен быть реализован в наследниках.
Args:
notification: Объект Notification
**kwargs: Дополнительные параметры
Returns:
True если успешно, False иначе
"""
pass
def log_success(self, notification):
"""Логирует успешную отправку"""
self.logger.info(
f"✅ Notification sent: "
f"id={notification.id}, "
f"recipient={notification.recipient.id}, "
f"channel={self.__class__.__name__}"
)
def log_failure(self, notification):
"""Логирует неуспешную отправку"""
self.logger.warning(
f"⚠️ Notification failed: "
f"id={notification.id}, "
f"recipient={notification.recipient.id}, "
f"channel={self.__class__.__name__}"
)
def log_error(self, notification_id: int, error: Exception):
"""Логирует ошибку"""
self.logger.error(
f"❌ Task error: "
f"notification_id={notification_id}, "
f"channel={self.__class__.__name__}, "
f"error={type(error).__name__}: {error}",
exc_info=True
)

111
backend/notifications/tasks/email.py Normal file
View File

@ -0,0 +1,111 @@
"""
Celery задачи для отправки email уведомлений
"""
from .base import BaseNotificationTask
class EmailNotificationTask(BaseNotificationTask):
"""
Celery задача для асинхронной отправки email уведомлений.
Особенности:
- Rate limiting: 10 email/минуту (защита от SMTP бана)
- Retry: 3 попытки с интервалом 5 минут
- Поддержка кастомных тем и получателей
"""
task_name = "notifications.send_email"
max_retries = 3
retry_delay = 300 # 5 минут между попытками
rate_limit = '10/m' # Не более 10 email в минуту
def send_notification(self, notification, **kwargs) -> bool:
"""
Отправляет email через EmailNotificationSender.
Args:
notification: Объект Notification
**kwargs:
- custom_subject: кастомная тема письма
- recipient_email: явный email получателя
Returns:
True если успешно, False иначе
"""
from notifications.senders.email import EmailNotificationSender
sender = EmailNotificationSender()
return sender.send(notification, **kwargs)
class DigestEmailTask(BaseNotificationTask):
"""
Celery задача для отправки email дайджестов.
Особенности:
- Отправка сводки непрочитанных уведомлений
- Поддержка daily/weekly частоты
- Максимум 50 уведомлений в одном дайджесте
"""
task_name = "notifications.send_digest_email"
max_retries = 2
retry_delay = 600 # 10 минут между попытками
rate_limit = '5/h' # Не более 5 дайджестов в час
def execute(self, celery_task, user_id: int, frequency: str = 'daily', **kwargs):
"""Переопределяем execute для работы с user_id вместо notification_id"""
try:
from django.contrib.auth import get_user_model
from notifications.models import Notification
from notifications.senders.email import EmailNotificationSender
from django.utils import timezone
from datetime import timedelta
User = get_user_model()
user = User.objects.get(id=user_id)
# Определяем период для дайджеста
if frequency == 'weekly':
cutoff = timezone.now() - timedelta(days=7)
else: # daily
cutoff = timezone.now() - timedelta(days=1)
# Получаем непрочитанные уведомления за период
notifications = Notification.objects.filter(
recipient=user,
unread=True,
emailed=False, # Еще не отправлены по email
timestamp__gte=cutoff,
).order_by('-timestamp')[:50] # Максимум 50 в дайджесте
if not notifications.exists():
self.logger.info(f"📭 No notifications for digest: user={user_id}, frequency={frequency}")
return False
# Отправляем дайджест
sender = EmailNotificationSender()
success = sender.send_digest(user, list(notifications), frequency=frequency)
if success:
self.logger.info(
f"✅ Digest sent: user={user_id}, frequency={frequency}, "
f"count={notifications.count()}"
)
else:
self.logger.warning(f"⚠️ Digest failed: user={user_id}, frequency={frequency}")
return success
except Exception as exc:
self.logger.exception(f"❌ Digest task error: user={user_id}, frequency={frequency}")
raise celery_task.retry(exc=exc)
def send_notification(self, notification, **kwargs) -> bool:
"""Не используется для дайджестов"""
pass
# Регистрируем задачи в Celery
send_email_notification = EmailNotificationTask.register_task()
send_digest_email = DigestEmailTask.register_task()

40
backend/notifications/tasks/push.py Normal file
View File

@ -0,0 +1,40 @@
"""
Celery задачи для отправки Web Push уведомлений
"""
from .base import BaseNotificationTask
class PushNotificationTask(BaseNotificationTask):
"""
Celery задача для асинхронной отправки Web Push уведомлений.
Особенности:
- Rate limiting: 50 push/минуту
- Retry: 2 попытки с интервалом 1 минута
- Автоматическое удаление неактивных устройств
"""
task_name = "notifications.send_push"
max_retries = 2
retry_delay = 60 # 1 минута между попытками
rate_limit = '50/m' # Не более 50 push в минуту
def send_notification(self, notification, **kwargs) -> bool:
"""
Отправляет Web Push через PushNotificationSender.
Args:
notification: Объект Notification
**kwargs: Дополнительные параметры
Returns:
True если успешно (хотя бы одно устройство), False иначе
"""
from notifications.senders.push import PushNotificationSender
sender = PushNotificationSender()
return sender.send(notification, **kwargs)
# Регистрируем задачу в Celery
send_push_notification = PushNotificationTask.register_task()

64
backend/notifications/tasks/websocket.py Normal file
View File

@ -0,0 +1,64 @@
"""
Celery задачи для отправки WebSocket уведомлений (realtime)
"""
from .base import BaseNotificationTask
class WebSocketNotificationTask(BaseNotificationTask):
"""
Celery задача для асинхронной отправки WebSocket уведомлений.
Особенности:
- Минимальный retry (1 попытка) - realtime не критично
- Быстрая повторная попытка (5 секунд)
- Поддержка silent режима (DND)
"""
task_name = "notifications.send_websocket"
max_retries = 1 # WebSocket - не критично, 1 попытка достаточно
retry_delay = 5 # 5 секунд (быстро)
rate_limit = None # Без ограничений для realtime
def execute(self, celery_task, notification_id: int, silent: bool = False, **kwargs):
"""Переопределяем для поддержки параметра silent"""
try:
notification = self.get_notification(notification_id)
if not notification:
self.logger.error(f"❌ Notification {notification_id} not found")
return False
# Отправляем с учетом silent режима
success = self.send_notification(notification, silent=silent, **kwargs)
if success:
self.log_success(notification)
else:
self.log_failure(notification)
return success
except Exception as exc:
self.log_error(notification_id, exc)
raise celery_task.retry(exc=exc)
def send_notification(self, notification, silent: bool = False, **kwargs) -> bool:
"""
Отправляет WebSocket через WebSocketNotificationSender.
Args:
notification: Объект Notification
silent: Если True, уведомление без звука/визуальных эффектов
**kwargs: Дополнительные параметры
Returns:
True если успешно, False иначе
"""
from notifications.senders.websocket import WebSocketNotificationSender
sender = WebSocketNotificationSender()
return sender.send(notification, silent=silent, **kwargs)
# Регистрируем задачу в Celery
send_websocket_notification = WebSocketNotificationTask.register_task()

515
backend/notifications/telegram_bot.py
View File

@ -1,515 +0,0 @@
"""
Telegram bot handlers для обработки команд пользователей
"""
import logging
from typing import Optional
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, KeyboardButton, ReplyKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import (
Application,
CommandHandler,
ContextTypes,
MessageHandler,
filters,
ConversationHandler,
)
from telegram.constants import ParseMode
from telegram.error import TelegramError
from django.conf import settings
from django.utils import timezone
from .telegram_models import TelegramUser
from employees.models import Employee
logger = logging.getLogger(__name__)
# Состояния для ConversationHandler
WAITING_FOR_CONTACT = 1
class TelegramBotHandlers:
"""
Обработчики команд Telegram бота
"""
@staticmethod
async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Команда /start - приветствие и инструкции
"""
user = update.effective_user
# Проверяем, привязан ли аккаунт
try:
tg_user = TelegramUser.objects.get(
telegram_id=user.id,
is_active=True
)
# Аккаунт уже привязан
employee = tg_user.user
await update.message.reply_text(
f"👋 Привет, {employee.first_name}!\n\n"
f"Ваш аккаунт уже привязан к EUSRR.\n\n"
f"Доступные команды:\n"
f"/settings - Настройки уведомлений\n"
f"/unlink - Отвязать аккаунт\n"
f"/help - Справка",
parse_mode=ParseMode.HTML
)
# Обновляем время взаимодействия
tg_user.update_last_interaction()
except TelegramUser.DoesNotExist:
# Аккаунт не привязан - показываем инструкцию
await update.message.reply_text(
"👋 Добро пожаловать в бот EUSRR!\n\n"
"Этот бот будет отправлять вам уведомления из системы EUSRR.\n\n"
"🔗 <b>Для привязки аккаунта:</b>\n"
"1. Войдите в личный кабинет EUSRR\n"
"2. Перейдите в настройки уведомлений\n"
"3. Нажмите 'Привязать Telegram'\n"
"4. Получите код привязки\n"
"5. Отправьте команду: /link ВАШ_КОД\n\n"
"❓ Помощь: /help",
parse_mode=ParseMode.HTML
)
@staticmethod
async def link_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Команда /link CODE - запрашивает код и контакт для привязки
"""
user = update.effective_user
# Проверяем что код указан
if not context.args or len(context.args) == 0:
await update.message.reply_text(
"❌ Укажите код привязки.\n\n"
"Использование: /link ВАШ_КОД\n\n"
"Код можно получить в настройках уведомлений EUSRR.",
parse_mode=ParseMode.HTML
)
return
link_code = context.args[0].upper()
# Проверяем что аккаунт еще не привязан
if TelegramUser.objects.filter(
telegram_id=user.id,
is_active=True
).exists():
await update.message.reply_text(
"⚠️ Ваш Telegram уже привязан к аккаунту EUSRR.\n\n"
"Если хотите привязать другой аккаунт, "
"сначала используйте /unlink",
parse_mode=ParseMode.HTML
)
return
# Ищем пользователя с таким кодом
try:
tg_user = TelegramUser.objects.select_related('user').get(
link_code=link_code,
telegram_id__isnull=True
)
# Проверяем срок действия кода
if not tg_user.is_link_code_valid():
await update.message.reply_text(
"❌ Код привязки истёк.\n\n"
"Пожалуйста, получите новый код в настройках EUSRR.",
parse_mode=ParseMode.HTML
)
return
# Сохраняем код в контексте для проверки контакта
context.user_data['pending_link_code'] = link_code
context.user_data['pending_tg_user_id'] = tg_user.id
# Создаем кнопку "Поделиться контактом"
from telegram import KeyboardButton, ReplyKeyboardMarkup
keyboard = ReplyKeyboardMarkup(
[[KeyboardButton("📱 Поделиться контактом",
request_contact=True)]],
one_time_keyboard=True,
resize_keyboard=True
)
await update.message.reply_text(
"🔐 <b>Подтверждение личности</b>\n\n"
"Для безопасности нажмите кнопку ниже, "
"чтобы поделиться вашим контактом.\n\n"
"Мы проверим что он совпадает с данными, "
"указанными при регистрации в EUSRR.",
parse_mode=ParseMode.HTML,
reply_markup=keyboard
)
except TelegramUser.DoesNotExist:
await update.message.reply_text(
"❌ Неверный код привязки.\n\n"
"Проверьте код и попробуйте снова.\n"
"Если код истёк, получите новый в настройках EUSRR.",
parse_mode=ParseMode.HTML
)
@staticmethod
async def contact_received(
update: Update,
context: ContextTypes.DEFAULT_TYPE
):
"""
Обработка полученного контакта для верификации
"""
from telegram import ReplyKeyboardRemove
user = update.effective_user
contact = update.message.contact
# Проверяем что контакт от самого пользователя
if contact.user_id != user.id:
await update.message.reply_text(
"❌ Пожалуйста, поделитесь <b>своим</b> контактом, "
"а не контактом другого пользователя.",
parse_mode=ParseMode.HTML,
reply_markup=ReplyKeyboardRemove()
)
return
# Проверяем что есть pending привязка
pending_code = context.user_data.get('pending_link_code')
pending_tg_user_id = context.user_data.get('pending_tg_user_id')
if not pending_code or not pending_tg_user_id:
await update.message.reply_text(
"❌ Нет активной попытки привязки.\n\n"
"Сначала используйте команду /link КОД",
parse_mode=ParseMode.HTML,
reply_markup=ReplyKeyboardRemove()
)
return
try:
tg_user = TelegramUser.objects.select_related('user').get(
id=pending_tg_user_id,
link_code=pending_code
)
employee = tg_user.user
# Проверяем совпадение данных
telegram_field = employee.telegram.strip().lower()
# Получаем username и phone из контакта
contact_username = f"@{contact.username}".lower() \
if contact.username else None
contact_phone = contact.phone_number
# Проверка совпадения
match_found = False
match_type = ""
if telegram_field:
# Проверка по username
if contact_username and telegram_field == contact_username:
match_found = True
match_type = f"username ({contact_username})"
# Проверка по номеру телефона
elif contact_phone:
# Нормализуем номера (убираем + и пробелы)
normalized_contact = contact_phone.replace('+', '') \
.replace(' ', '').replace('-', '')
normalized_field = telegram_field.replace('+', '') \
.replace(' ', '').replace('-', '')
if normalized_contact == normalized_field:
match_found = True
match_type = f"номер телефона ({contact_phone})"
if not match_found:
await update.message.reply_text(
"❌ <b>Данные не совпадают</b>\n\n"
f"Ваш контакт:\n"
f"Username: {contact_username or 'не указан'}\n"
f"Телефон: {contact_phone or 'не указан'}\n\n"
f"В профиле EUSRR указано: {telegram_field or 'не указано'}\n\n"
"Обновите поле 'Telegram' в вашем профиле EUSRR "
"и попробуйте снова.",
parse_mode=ParseMode.HTML,
reply_markup=ReplyKeyboardRemove()
)
# Очищаем pending данные
context.user_data.clear()
return
# ✅ Данные совпали - привязываем аккаунт
tg_user.telegram_id = user.id
tg_user.telegram_username = user.username or ''
tg_user.first_name = user.first_name or ''
tg_user.last_name = user.last_name or ''
tg_user.is_active = True
tg_user.is_blocked = False
tg_user.clear_link_code()
tg_user.update_last_interaction()
tg_user.save()
await update.message.reply_text(
f"✅ <b>Аккаунт успешно привязан!</b>\n\n"
f"Подтверждено через: {match_type}\n\n"
f"👤 Привязан к: {employee.get_full_name()}\n"
f"📧 Email: {employee.email}\n\n"
f"Теперь вы будете получать уведомления EUSRR в Telegram.\n\n"
f"Доступные команды:\n"
f"/settings - Настройки уведомлений\n"
f"/unlink - Отвязать аккаунт\n"
f"/help - Справка",
parse_mode=ParseMode.HTML,
reply_markup=ReplyKeyboardRemove()
)
logger.info(
f"Telegram привязан через {match_type}: "
f"{user.id} -> {employee.email}"
)
# Очищаем pending данные
context.user_data.clear()
except TelegramUser.DoesNotExist:
await update.message.reply_text(
"❌ Код привязки истёк или недействителен.\n\n"
"Получите новый код в настройках EUSRR.",
parse_mode=ParseMode.HTML,
reply_markup=ReplyKeyboardRemove()
)
context.user_data.clear()
@staticmethod
async def unlink_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Команда /unlink - отвязка аккаунта
"""
user = update.effective_user
try:
tg_user = TelegramUser.objects.get(
telegram_id=user.id,
is_active=True
)
employee_name = tg_user.user.get_full_name()
# Отвязываем аккаунт
tg_user.is_active = False
tg_user.save()
await update.message.reply_text(
f"✅ <b>Аккаунт отвязан</b>\n\n"
f"Связь с {employee_name} удалена.\n"
f"Вы больше не будете получать уведомления.\n\n"
f"Для повторной привязки используйте /link",
parse_mode=ParseMode.HTML
)
logger.info(f"Telegram аккаунт отвязан: {user.id}")
except TelegramUser.DoesNotExist:
await update.message.reply_text(
"⚠️ Ваш аккаунт не привязан к EUSRR.\n\n"
"Используйте /start для инструкций по привязке.",
parse_mode=ParseMode.HTML
)
@staticmethod
async def settings_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Команда /settings - показать текущие настройки
"""
user = update.effective_user
try:
tg_user = TelegramUser.objects.get(
telegram_id=user.id,
is_active=True
)
employee = tg_user.user
# Получаем настройки уведомлений
from .models import UserNotificationSettings, NotificationCategory
settings_text = f"⚙️ <b>Настройки уведомлений</b>\n\n"
settings_text += f"👤 Пользователь: {employee.get_full_name()}\n"
settings_text += f"📧 Email: {employee.email}\n"
settings_text += f"🔗 Telegram: @{tg_user.telegram_username or 'не указан'}\n\n"
settings_text += "<b>Категории уведомлений:</b>\n\n"
categories = NotificationCategory.objects.all()
for category in categories:
try:
user_settings = UserNotificationSettings.objects.get(
user=employee,
category=category
)
# Эмодзи для категорий
emoji_map = {
'communications': '💬',
'documents': '📄',
'requests': '📋',
'calendar': '📅',
'department': '👥',
'profile': '👤',
'feed': '📰',
'system': '⚙️',
}
emoji = emoji_map.get(category.code, '🔔')
status = "✅" if user_settings.send_telegram else "❌"
settings_text += f"{emoji} {category.name}: {status}\n"
except UserNotificationSettings.DoesNotExist:
pass
# Создаем кнопку для перехода в настройки
site_url = getattr(settings, 'SITE_URL', 'http://localhost:9000')
keyboard = InlineKeyboardMarkup([
[InlineKeyboardButton(
"🔧 Изменить настройки",
url=f"{site_url}/notifications/settings/"
)]
])
await update.message.reply_text(
settings_text,
parse_mode=ParseMode.HTML,
reply_markup=keyboard
)
tg_user.update_last_interaction()
except TelegramUser.DoesNotExist:
await update.message.reply_text(
"⚠️ Ваш аккаунт не привязан к EUSRR.\n\n"
"Используйте /start для инструкций по привязке.",
parse_mode=ParseMode.HTML
)
@staticmethod
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Команда /help - справка
"""
await update.message.reply_text(
"📖 <b>Справка по командам бота EUSRR</b>\n\n"
"<b>Основные команды:</b>\n\n"
"/start - Приветствие и инструкции\n"
"/link КОД - Привязать аккаунт EUSRR\n"
"/unlink - Отвязать аккаунт\n"
"/settings - Показать текущие настройки\n"
"/help - Эта справка\n\n"
"<b>Как привязать аккаунт:</b>\n"
"1. Войдите в личный кабинет EUSRR\n"
"2. Откройте настройки уведомлений\n"
"3. Нажмите 'Привязать Telegram'\n"
"4. Скопируйте код привязки\n"
"5. Отправьте: /link ВАШ_КОД\n\n"
"<b>О боте:</b>\n"
"Этот бот отправляет уведомления из системы EUSRR "
"о новых сообщениях, документах, заявках и других событиях.\n\n"
"Настроить какие уведомления получать можно в личном кабинете EUSRR.",
parse_mode=ParseMode.HTML
)
@staticmethod
async def unknown_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Обработчик неизвестных команд и текстовых сообщений
"""
await update.message.reply_text(
"❓ Неизвестная команда.\n\n"
"Используйте /help для списка доступных команд.",
parse_mode=ParseMode.HTML
)
def create_telegram_bot_application() -> Optional[Application]:
"""
Создает и настраивает приложение Telegram бота
Returns:
Настроенное приложение или None если токен не настроен
"""
token = getattr(settings, 'TELEGRAM_BOT_TOKEN', None)
if not token:
logger.error("TELEGRAM_BOT_TOKEN не настроен в settings")
return None
# Создаем приложение
application = Application.builder().token(token).build()
# Регистрируем handlers
handlers = TelegramBotHandlers
application.add_handler(CommandHandler("start", handlers.start_command))
application.add_handler(CommandHandler("link", handlers.link_command))
application.add_handler(CommandHandler("unlink", handlers.unlink_command))
application.add_handler(
CommandHandler("settings", handlers.settings_command)
)
application.add_handler(CommandHandler("help", handlers.help_command))
# Обработчик контактов (для верификации при привязке)
application.add_handler(
MessageHandler(filters.CONTACT, handlers.contact_received)
)
# Обработчик неизвестных команд и текстовых сообщений
application.add_handler(
MessageHandler(
filters.COMMAND | filters.TEXT,
handlers.unknown_command
)
)
logger.info("Telegram bot application создан")
return application
async def run_telegram_bot():
"""
Запускает Telegram бота в режиме polling
"""
application = create_telegram_bot_application()
if not application:
logger.error("Не удалось создать Telegram bot application")
return
logger.info("Запуск Telegram бота...")
try:
# Инициализация
await application.initialize()
await application.start()
# Запуск polling
await application.updater.start_polling(
allowed_updates=Update.ALL_TYPES
)
logger.info("Telegram бот запущен и слушает обновления")
# Ожидание остановки
await application.updater.stop()
await application.stop()
await application.shutdown()
except Exception as e:
logger.error(f"Ошибка при запуске Telegram бота: {e}", exc_info=True)

146
backend/notifications/telegram_models.py
View File

@ -1,146 +0,0 @@
"""
Модели для интеграции с Telegram
"""
from django.db import models
from django.conf import settings
from django.utils import timezone
class TelegramUser(models.Model):
"""Связь пользователя системы с Telegram аккаунтом"""
user = models.OneToOneField(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='telegram_account',
verbose_name='Пользователь'
)
telegram_id = models.BigIntegerField(
unique=True,
null=True,
blank=True,
verbose_name='Telegram ID',
help_text='Уникальный ID пользователя в Telegram'
)
telegram_username = models.CharField(
max_length=100,
blank=True,
null=True,
verbose_name='Telegram Username',
help_text='@username пользователя в Telegram'
)
first_name = models.CharField(
max_length=100,
blank=True,
verbose_name='Имя в Telegram'
)
last_name = models.CharField(
max_length=100,
blank=True,
verbose_name='Фамилия в Telegram'
)
# Код для привязки аккаунта
link_code = models.CharField(
max_length=10,
blank=True,
null=True,
unique=True,
verbose_name='Код привязки',
help_text='Временный код для привязки аккаунта'
)
link_code_created_at = models.DateTimeField(
null=True,
blank=True,
verbose_name='Код создан'
)
# Статус
is_active = models.BooleanField(
default=True,
verbose_name='Активен'
)
is_blocked = models.BooleanField(
default=False,
verbose_name='Заблокирован',
help_text='Пользователь заблокировал бота'
)
# Даты
linked_at = models.DateTimeField(
auto_now_add=True,
verbose_name='Привязан'
)
last_interaction_at = models.DateTimeField(
null=True,
blank=True,
verbose_name='Последнее взаимодействие'
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name='Обновлено'
)
class Meta:
verbose_name = 'Telegram аккаунт'
verbose_name_plural = 'Telegram аккаунты'
db_table = 'notifications_telegram_user'
indexes = [
models.Index(fields=['telegram_id']),
models.Index(fields=['link_code']),
models.Index(fields=['user']),
]
def __str__(self):
if self.telegram_username:
username = f"@{self.telegram_username}"
elif self.telegram_id:
username = str(self.telegram_id)
else:
username = "не привязан"
return f'{self.user.get_short_name()} ({username})'
def generate_link_code(self):
"""Генерирует уникальный код для привязки аккаунта"""
import random
import string
# Генерируем 6-значный код
code = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))
# Проверяем уникальность
while TelegramUser.objects.filter(link_code=code).exists():
code = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))
self.link_code = code
self.link_code_created_at = timezone.now()
self.save(update_fields=['link_code', 'link_code_created_at'])
return code
def is_link_code_valid(self):
"""Проверяет, действителен ли код привязки (не старше 15 минут)"""
if not self.link_code or not self.link_code_created_at:
return False
age = timezone.now() - self.link_code_created_at
return age.total_seconds() < 900 # 15 минут
def clear_link_code(self):
"""Удаляет код привязки"""
self.link_code = None
self.link_code_created_at = None
self.save(update_fields=['link_code', 'link_code_created_at'])
def update_last_interaction(self):
"""Обновляет время последнего взаимодействия"""
self.last_interaction_at = timezone.now()
self.save(update_fields=['last_interaction_at'])

31
backend/notifications/test_views.py
View File

@ -1,31 +0,0 @@
"""
Тестовый view для создания уведомлений
"""
from django.http import JsonResponse
from django.contrib.auth.decorators import login_required
from notifications.services import NotificationService
@login_required
def create_test_notification(request):
"""Создать тестовое уведомление для текущего пользователя"""
notification = NotificationService.create_notification(
recipient=request.user,
notification_type_code='system_announcement',
title='🎉 Тестовое уведомление',
message='Это тестовое сообщение отправлено через API. Вы должны увидеть toast!',
action_url='/notifications/',
action_text='Открыть уведомления'
)
if notification:
return JsonResponse({
'success': True,
'notification_id': notification.id,
'message': 'Уведомление создано и отправлено'
})
else:
return JsonResponse({
'success': False,
'message': 'Не удалось создать уведомление'
}, status=400)

3
backend/notifications/tests.py
View File

@ -1,3 +0,0 @@
from django.test import TestCase
# Create your tests here.

10
backend/notifications/urls.py
View File

@ -1,10 +0,0 @@
from django.urls import path
from . import views, test_views
app_name = 'notifications'
urlpatterns = [
path('', views.notification_list, name='list'),
path('settings/', views.notification_settings, name='settings'),
path('test/create/', test_views.create_test_notification, name='test_create'),
]

24
backend/notifications/views.py
View File

@ -1,24 +0,0 @@
from django.shortcuts import render
from django.contrib.auth.decorators import login_required
@login_required
def notification_list(request):
"""Страница списка уведомлений с фильтрами и пагинацией"""
return render(request, 'notifications/notification_list.html')
@login_required
def notification_settings(request):
"""Страница настроек уведомлений"""
from django.conf import settings
context = {
'telegram_bot_username': getattr(
settings,
'TELEGRAM_BOT_USERNAME',
'eusrr_bot'
)
}
return render(request, 'notifications/notification_settings.html', context)