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

refactor: удалены deprecated Python API файлы

Browse Source 
- Deleted notifications/api_views.py (585 lines, @deprecated)
- Deleted notifications/api_urls.py (83 lines, @deprecated)
- Deleted test_notifications_refactoring.py (manual test)

Все notifications API уже мигрированы в api/v1/notifications/
This commit is contained in:
Игорь Надеин 2026-01-13 10:19:26 +03:00
parent 2a012c297f
commit 8b6e437599
3 changed files with 0 additions and 861 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
backend/notifications/api_urls.py
View File

@ -1,83 +0,0 @@
"""
DEPRECATED: Этот файл устарел и будет удалён в следующей версии.
Используйте вместо него: api.v1.notifications.urls
URL routing перенесён в api/v1/notifications/urls.py
Новые пути: /api/v1/notifications/* вместо /api/v1/notifications/*
Миграция: 9 января 2026
"""
from django.urls import path
from . import api_views
app_name = 'notifications_api'
urlpatterns = [
path('notifications/', api_views.get_notifications, name='list'),
path('notifications/count/', api_views.get_unread_count, name='count'),
path(
'notifications/<int:notification_id>/read/',
api_views.mark_as_read,
name='mark_read'
),
path(
'notifications/read-all/',
api_views.mark_all_as_read,
name='mark_all_read'
),
path(
'notifications/<int:notification_id>/',
api_views.delete_notification,
name='delete'
),
path('notifications/categories/', api_views.get_categories, name='categories'),
path('notifications/settings/', api_views.get_user_settings, name='settings_get'),
path(
'notifications/settings/update/',
api_views.update_user_settings,
name='settings_update'
),
path(
'notifications/settings/category/update/',
api_views.update_category_settings,
name='category_settings_update'
),
# Telegram integration
path(
'notifications/telegram/status/',
api_views.get_telegram_link_status,
name='telegram_status'
),
path(
'notifications/telegram/generate-code/',
api_views.generate_telegram_link_code,
name='telegram_generate_code'
),
path(
'notifications/telegram/unlink/',
api_views.unlink_telegram,
name='telegram_unlink'
),
# Web Push integration
path(
'notifications/push/vapid-key/',
api_views.get_vapid_public_key,
name='push_vapid_key'
),
path(
'notifications/push/subscribe/',
api_views.subscribe_push,
name='push_subscribe'
),
path(
'notifications/push/unsubscribe/',
api_views.unsubscribe_push,
name='push_unsubscribe'
),
path(
'notifications/push/subscriptions/',
api_views.get_push_subscriptions,
name='push_subscriptions'
),
]

584
backend/notifications/api_views.py
View File

@ -1,584 +0,0 @@
"""
DEPRECATED: Этот файл устарел и будет удалён в следующей версии.
Используйте вместо него: api.v1.notifications.views
Все API endpoints перенесены в api/v1/notifications/ для единообразия
с остальными модулями проекта (employees, documents, requests и т.д.)
Миграция: 9 января 2026
"""
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework import status
from django.shortcuts import get_object_or_404
from .models import Notification, NotificationCategory, NotificationType
from .services import NotificationService
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_notifications(request):
"""Получить список уведомлений с пагинацией и фильтрами"""
page = int(request.GET.get('page', 1))
page_size = int(request.GET.get('page_size', 20))
category = request.GET.get('category')
is_read = request.GET.get('is_read')
search = request.GET.get('search', '').strip()
unread_only = request.GET.get('unread_only', 'false').lower() == 'true'
queryset = Notification.objects.filter(
recipient=request.user,
is_archived=False
).select_related(
'notification_type',
'notification_type__category'
).order_by('-created_at')
# Фильтр по категории
if category:
queryset = queryset.filter(notification_type__category__code=category)
# Фильтр по статусу прочитанности
if is_read and is_read in ('true', 'false'):
queryset = queryset.filter(is_read=(is_read == 'true'))
elif unread_only:
queryset = queryset.filter(is_read=False)
# Поиск по заголовку и тексту
if search:
from django.db.models import Q
queryset = queryset.filter(
Q(title__icontains=search) |
Q(message__icontains=search) |
Q(short_message__icontains=search)
)
total = queryset.count()
start = (page - 1) * page_size
end = start + page_size
notifications = queryset[start:end]
data = {
'total': total,
'page': page,
'page_size': page_size,
'notifications': [
{
'id': n.id,
'title': n.title,
'message': n.message, # Полное сообщение для списка
'category': n.notification_type.category.code,
'category_name': n.notification_type.category.name,
'icon': n.notification_type.category.icon,
'color': n.notification_type.category.color,
'priority': n.notification_type.priority,
'is_read': n.is_read,
'action_url': n.action_url,
'action_text': n.action_text,
'created_at': n.created_at.isoformat(),
}
for n in notifications
]
}
return Response(data)
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_unread_count(request):
"""Получить количество непрочитанных уведомлений"""
count = Notification.objects.filter(
recipient=request.user,
is_read=False,
is_archived=False
).count()
return Response({'count': count})
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def mark_as_read(request, notification_id):
"""Отметить уведомление как прочитанное"""
success = NotificationService.mark_as_read(notification_id, request.user)
if success:
return Response({'status': 'success'})
else:
return Response(
{'status': 'error', 'message': 'Notification not found'},
status=status.HTTP_404_NOT_FOUND
)
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def mark_all_as_read(request):
"""Отметить все уведомления как прочитанные"""
category = request.data.get('category')
count = NotificationService.mark_all_as_read(request.user, category)
return Response({'status': 'success', 'count': count})
@api_view(['DELETE'])
@permission_classes([IsAuthenticated])
def delete_notification(request, notification_id):
"""Удалить уведомление (архивировать)"""
try:
notification = Notification.objects.get(
id=notification_id,
recipient=request.user
)
notification.archive()
return Response({'status': 'success'})
except Notification.DoesNotExist:
return Response(
{'status': 'error', 'message': 'Notification not found'},
status=status.HTTP_404_NOT_FOUND
)
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_categories(request):
"""Получить список категорий уведомлений"""
categories = NotificationCategory.objects.filter(is_active=True)
data = [
{
'code': cat.code,
'name': cat.name,
'icon': cat.icon,
'color': cat.color,
}
for cat in categories
]
return Response(data)
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_user_settings(request):
"""Получить настройки уведомлений пользователя по категориям"""
categories = NotificationCategory.objects.filter(is_active=True)
settings_by_category = {}
for category in categories:
# Получить типы уведомлений этой категории
types = NotificationType.objects.filter(
category=category,
is_active=True
)
if not types.exists():
continue
# Получить настройки первого типа категории как представитель
# (предполагается, что настройки одинаковы для всей категории)
first_type = types.first()
settings = NotificationService.get_user_settings(
request.user, first_type
)
settings_by_category[category.code] = {
'is_enabled': settings.is_enabled,
'web_enabled': settings.send_web,
'email_enabled': settings.send_email,
'email_frequency': settings.email_frequency,
'telegram_enabled': settings.send_telegram,
}
return Response({
'settings': settings_by_category
})
@api_view(['PUT'])
@permission_classes([IsAuthenticated])
def update_user_settings(request):
"""Обновить настройки уведомлений пользователя"""
notification_type_code = request.data.get('notification_type_code')
try:
notification_type = NotificationType.objects.get(
code=notification_type_code
)
except NotificationType.DoesNotExist:
return Response(
{'status': 'error', 'message': 'Notification type not found'},
status=status.HTTP_404_NOT_FOUND
)
settings = NotificationService.get_user_settings(
request.user, notification_type
)
# Обновить настройки
if 'is_enabled' in request.data and not notification_type.is_required:
settings.is_enabled = request.data['is_enabled']
if 'send_web' in request.data:
settings.send_web = request.data['send_web']
if 'send_email' in request.data:
settings.send_email = request.data['send_email']
if 'send_telegram' in request.data:
settings.send_telegram = request.data['send_telegram']
if 'send_whatsapp' in request.data:
settings.send_whatsapp = request.data['send_whatsapp']
if 'send_wechat' in request.data:
settings.send_wechat = request.data['send_wechat']
if 'quiet_hours_enabled' in request.data:
settings.quiet_hours_enabled = request.data['quiet_hours_enabled']
if 'quiet_start_time' in request.data:
settings.quiet_start_time = request.data['quiet_start_time']
if 'quiet_end_time' in request.data:
settings.quiet_end_time = request.data['quiet_end_time']
settings.save()
return Response({'status': 'success'})
@api_view(['PUT'])
@permission_classes([IsAuthenticated])
def update_category_settings(request):
"""
Обновить настройки для всех уведомлений категории
"""
category_code = request.data.get('category')
if not category_code:
return Response(
{'status': 'error', 'message': 'Category is required'},
status=status.HTTP_400_BAD_REQUEST
)
try:
category = NotificationCategory.objects.get(code=category_code)
except NotificationCategory.DoesNotExist:
return Response(
{'status': 'error', 'message': 'Category not found'},
status=status.HTTP_404_NOT_FOUND
)
# Получить все типы уведомлений категории
notification_types = NotificationType.objects.filter(
category=category,
is_active=True
)
# Обновить настройки для каждого типа
updated_count = 0
for notification_type in notification_types:
settings = NotificationService.get_user_settings(
request.user, notification_type
)
# Обновить поля
if 'is_enabled' in request.data and not notification_type.is_required:
settings.is_enabled = request.data['is_enabled']
if 'web_enabled' in request.data:
settings.send_web = request.data['web_enabled']
if 'email_enabled' in request.data:
settings.send_email = request.data['email_enabled']
if 'email_frequency' in request.data:
settings.email_frequency = request.data['email_frequency']
if 'telegram_enabled' in request.data:
settings.send_telegram = request.data['telegram_enabled']
settings.save()
updated_count += 1
return Response({
'status': 'success',
'updated_count': updated_count
})
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_telegram_link_status(request):
"""Получить статус привязки Telegram аккаунта"""
from .telegram_models import TelegramUser
try:
tg_user = TelegramUser.objects.get(
user=request.user,
is_active=True,
telegram_id__isnull=False
)
return Response({
'is_linked': True,
'telegram_username': tg_user.telegram_username,
'first_name': tg_user.first_name,
'last_name': tg_user.last_name,
'linked_at': tg_user.linked_at,
'is_blocked': tg_user.is_blocked,
})
except TelegramUser.DoesNotExist:
# Проверяем, может быть есть незавершенная привязка с кодом
pending = TelegramUser.objects.filter(
user=request.user,
link_code__isnull=False
).first()
response_data = {'is_linked': False}
if pending and pending.is_link_code_valid():
response_data['link_code'] = pending.link_code
return Response(response_data)
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def generate_telegram_link_code(request):
"""Сгенерировать код для привязки Telegram аккаунта"""
from .telegram_models import TelegramUser
# Проверяем что аккаунт еще не привязан
existing = TelegramUser.objects.filter(
user=request.user,
is_active=True,
telegram_id__isnull=False
).first()
if existing:
return Response({
'status': 'error',
'message': 'Telegram аккаунт уже привязан'
}, status=status.HTTP_400_BAD_REQUEST)
# Получаем или создаем запись БЕЗ telegram_id (он будет установлен при привязке)
tg_user = TelegramUser.objects.filter(
user=request.user
).first()
if not tg_user:
tg_user = TelegramUser.objects.create(
user=request.user,
is_active=False
)
# Генерируем новый код (даже если уже был старый)
link_code = tg_user.generate_link_code()
from django.conf import settings
bot_username = getattr(settings, 'TELEGRAM_BOT_USERNAME', 'eusrr_bot')
return Response({
'status': 'success',
'link_code': link_code,
'bot_username': bot_username,
'bot_link': f'https://t.me/{bot_username}',
'expires_in_seconds': 900, # 15 минут
})
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def unlink_telegram(request):
"""Отвязать Telegram аккаунт"""
from .telegram_models import TelegramUser
try:
tg_user = TelegramUser.objects.get(
user=request.user,
is_active=True
)
tg_user.is_active = False
tg_user.save()
return Response({
'status': 'success',
'message': 'Telegram аккаунт отвязан'
})
except TelegramUser.DoesNotExist:
return Response({
'status': 'error',
'message': 'Telegram аккаунт не привязан'
}, status=status.HTTP_404_NOT_FOUND)
# ============================================
# Web Push API (для браузерных уведомлений)
# ============================================
from django.conf import settings
from .models import WebPushSubscription
import logging
logger = logging.getLogger(__name__)
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_vapid_public_key(request):
"""
Получить публичный VAPID ключ для подписки на push-уведомления.
Этот ключ нужен браузеру для создания подписки.
"""
return Response({
'vapid_public_key': settings.VAPID_PUBLIC_KEY
})
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def subscribe_push(request):
"""
Подписаться на Web Push уведомления.
Ожидаемые данные:
{
"endpoint": "https://fcm.googleapis.com/...",
"keys": {
"p256dh": "...",
"auth": "..."
},
"user_agent": "...", // опционально
"device_name": "..." // опционально
}
"""
try:
endpoint = request.data.get('endpoint')
keys = request.data.get('keys', {})
p256dh_key = keys.get('p256dh')
auth_key = keys.get('auth')
user_agent = request.data.get('user_agent', '')
device_name = request.data.get('device_name', '')
if not endpoint or not p256dh_key or not auth_key:
return Response({
'status': 'error',
'message': 'Отсутствуют обязательные поля: endpoint, keys.p256dh, keys.auth'
}, status=status.HTTP_400_BAD_REQUEST)
# Обновляем или создаем подписку
subscription, created = WebPushSubscription.objects.update_or_create(
user=request.user,
endpoint=endpoint,
defaults={
'p256dh_key': p256dh_key,
'auth_key': auth_key,
'user_agent': user_agent,
'device_name': device_name,
'is_active': True,
'error_count': 0,
'last_error': '',
}
)
# Логируем только создание новой подписки, не обновление
if created:
logger.info(f"[WebPush] New subscription for user {request.user.id}")
return Response({
'status': 'success',
'message': 'Подписка на push-уведомления активирована',
'created': created
})
except Exception as e:
logger.exception(f"[WebPush] Error subscribing user {request.user.id}: {e}")
return Response({
'status': 'error',
'message': 'Ошибка при создании подписки'
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@api_view(['DELETE'])
@permission_classes([IsAuthenticated])
def unsubscribe_push(request):
"""
Отписаться от Web Push уведомлений.
Ожидаемые данные:
{
"endpoint": "https://fcm.googleapis.com/..."
}
Если endpoint не указан - отписывает от всех подписок пользователя.
"""
try:
endpoint = request.data.get('endpoint')
if endpoint:
# Удаляем конкретную подписку
deleted, _ = WebPushSubscription.objects.filter(
user=request.user,
endpoint=endpoint
).delete()
if deleted:
logger.info(f"[WebPush] Deleted subscription for user {request.user.id} (endpoint: {endpoint[:50]}...)")
return Response({
'status': 'success',
'message': 'Подписка удалена'
})
else:
return Response({
'status': 'error',
'message': 'Подписка не найдена'
}, status=status.HTTP_404_NOT_FOUND)
else:
# Удаляем все подписки пользователя
deleted, _ = WebPushSubscription.objects.filter(user=request.user).delete()
logger.info(f"[WebPush] Deleted all ({deleted}) subscriptions for user {request.user.id}")
return Response({
'status': 'success',
'message': f'Удалено подписок: {deleted}'
})
except Exception as e:
logger.exception(f"[WebPush] Error unsubscribing user {request.user.id}: {e}")
return Response({
'status': 'error',
'message': 'Ошибка при удалении подписки'
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_push_subscriptions(request):
"""
Получить список активных push-подписок пользователя.
"""
subscriptions = WebPushSubscription.objects.filter(
user=request.user,
is_active=True
).order_by('-updated_at')
return Response({
'subscriptions': [
{
'id': sub.id,
'device_name': sub.device_name or 'Неизвестное устройство',
'user_agent': sub.user_agent[:100] if sub.user_agent else '',
'created_at': sub.created_at.isoformat(),
'updated_at': sub.updated_at.isoformat(),
}
for sub in subscriptions
]
})

194
backend/test_notifications_refactoring.py
View File

@ -1,194 +0,0 @@
"""
Тесты для проверки результатов рефакторинга модуля notifications.
"""
import os
import sys
import django
# Setup Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
django.setup()
print('Тестирование рефакторинга notifications...')
print()
passed = 0
failed = 0
# Тест 1: Импорт процессоров
try:
from notifications.task_base import (
BaseNotificationProcessor,
MessageNotificationProcessor,
EventNotificationProcessor,
PostNotificationProcessor,
RequestNotificationProcessor,
DocumentNotificationProcessor,
truncate_text
)
print('[PASS] Task processors imported')
passed += 1
except Exception as e:
print(f'[FAIL] Task processors import failed: {e}')
failed += 1
# Тест 2: Импорт задач
try:
from notifications.tasks import (
process_message_notifications_task,
process_event_notifications_task,
process_post_notifications_task,
process_request_notifications_task,
process_document_notifications_task
)
print('[PASS] Celery tasks imported')
passed += 1
except Exception as e:
print(f'[FAIL] Celery tasks import failed: {e}')
failed += 1
# Тест 3: Импорт новых API views
try:
from api.v1.notifications.views import (
get_notifications,
mark_as_read,
get_user_settings,
subscribe_push,
get_telegram_link_status
)
print('[PASS] API v1 views imported (16 endpoints)')
passed += 1
except Exception as e:
print(f'[FAIL] API v1 views import failed: {e}')
failed += 1
# Тест 4: Проверка truncate_text
try:
from notifications.task_base import truncate_text
result = truncate_text('Test message that is very long', 10)
assert result == 'Test messa...', f'Expected "Test messa...", got "{result}"'
result2 = truncate_text('Short', 100)
assert result2 == 'Short', f'Expected "Short", got "{result2}"'
result3 = truncate_text('', 10)
assert result3 == '', f'Expected empty string, got "{result3}"'
print('[PASS] truncate_text works correctly')
passed += 1
except Exception as e:
print(f'[FAIL] truncate_text failed: {e}')
failed += 1
# Тест 5: Проверка, что задачи зарегистрированы в Celery
try:
from notifications.tasks import process_message_notifications_task
assert hasattr(process_message_notifications_task, 'delay'), 'Task should have delay method'
assert hasattr(process_message_notifications_task, 'apply_async'), 'Task should have apply_async method'
assert process_message_notifications_task.name == 'communications.process_message_notifications'
print('[PASS] Celery tasks are properly registered')
passed += 1
except Exception as e:
print(f'[FAIL] Celery task registration failed: {e}')
failed += 1
# Тест 6: Проверка URL роутинга
try:
from django.urls import reverse, resolve
# Проверяем новые URL
url = reverse('api:v1:notifications_api_v1:list')
assert url == '/api/v1/notifications/', f'Expected /api/v1/notifications/, got {url}'
url2 = reverse('api:v1:notifications_api_v1:count')
assert url2 == '/api/v1/notifications/count/', f'Expected /api/v1/notifications/count/, got {url2}'
print('[PASS] URL routing configured correctly')
passed += 1
except Exception as e:
print(f'[FAIL] URL routing failed: {e}')
failed += 1
# Тест 7: Проверка сигналов
try:
import communications.notification_signals
import calendar_app.notification_signals
import feed.notification_signals
import requests_app.notification_signals
import documents.notification_signals
print('[PASS] Notification signals imported')
passed += 1
except Exception as e:
print(f'[FAIL] Notification signals import failed: {e}')
failed += 1
# Тест 8: Проверка старых deprecated файлов
try:
from notifications import api_views, api_urls
import warnings
# Проверяем, что файлы содержат deprecation warning
with open('notifications/api_views.py', 'r', encoding='utf-8') as f:
content = f.read()
assert 'DEPRECATED' in content, 'api_views.py should have DEPRECATED notice'
with open('notifications/api_urls.py', 'r', encoding='utf-8') as f:
content = f.read()
assert 'DEPRECATED' in content, 'api_urls.py should have DEPRECATED notice'
print('[PASS] Deprecated files marked correctly')
passed += 1
except Exception as e:
print(f'[FAIL] Deprecated files check failed: {e}')
failed += 1
# Тест 9: Проверка размера файлов
try:
import os
tasks_size = os.path.getsize('notifications/tasks.py')
task_base_size = os.path.getsize('notifications/task_base.py')
# tasks.py должен быть значительно меньше task_base.py после рефакторинга
# tasks.py ~ 380 строк (20-25KB), task_base.py ~ 600 строк (25-35KB)
assert tasks_size < 30000, f'tasks.py too large: {tasks_size} bytes (expected < 30KB)'
assert task_base_size > 20000, f'task_base.py too small: {task_base_size} bytes (expected > 20KB)'
print(f'[PASS] File sizes correct (tasks.py: {tasks_size//1024}KB, task_base.py: {task_base_size//1024}KB)')
passed += 1
except Exception as e:
print(f'[FAIL] File size check failed: {e}')
failed += 1
# Тест 10: Проверка, что процессоры наследуются от базового класса
try:
from notifications.task_base import (
BaseNotificationProcessor,
MessageNotificationProcessor,
EventNotificationProcessor
)
assert issubclass(MessageNotificationProcessor, BaseNotificationProcessor)
assert issubclass(EventNotificationProcessor, BaseNotificationProcessor)
# Проверяем наличие обязательных методов
assert hasattr(MessageNotificationProcessor, 'process')
assert hasattr(BaseNotificationProcessor, 'send_notifications')
assert hasattr(BaseNotificationProcessor, 'log_result')
print('[PASS] Processor inheritance structure correct')
passed += 1
except Exception as e:
print(f'[FAIL] Processor inheritance check failed: {e}')
failed += 1
print()
print('=' * 50)
print(f'Результаты: {passed} PASSED, {failed} FAILED')
print('=' * 50)
if failed == 0:
print('Все тесты пройдены успешно!')
sys.exit(0)
else:
print(f'Обнаружены ошибки в {failed} тестах')
sys.exit(1)