# -*- coding: utf-8 -*- """ Notifications Views This module contains class-based views for the notifications app, including web views for displaying notifications and API views for RESTful access to notification data. Views: - NotificationListView: Display user notifications - NotificationDetailView: Display notification details - NotificationPreferencesView: Manage user preferences - NotificationStatsView: Display notification statistics - NotificationAPIView: RESTful API for notifications - NotificationPreferencesAPIView: API for preferences - NotificationStatsAPIView: API for statistics Author: AdTlas Development Team Version: 1.0.0 Last Updated: 2024 """ import json from datetime import datetime, timedelta from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib.auth.decorators import login_required from django.contrib import messages from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger from django.db.models import Q, Count, Avg, Max, Min from django.http import JsonResponse, HttpResponse, Http404 from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse_lazy, reverse from django.utils import timezone from django.utils.decorators import method_decorator from django.utils.translation import gettext_lazy as _ from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods, require_POST from django.views.generic import ( ListView, DetailView, CreateView, UpdateView, DeleteView, TemplateView, View, ) from django.views.generic.edit import FormView from rest_framework import status, permissions from rest_framework.decorators import api_view, permission_classes from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.generics import ( ListCreateAPIView, RetrieveUpdateDestroyAPIView, ListAPIView, RetrieveAPIView, ) from rest_framework.pagination import PageNumberPagination from rest_framework.filters import SearchFilter, OrderingFilter from django_filters.rest_framework import DjangoFilterBackend from apps.core.mixins import ( AdminRequiredMixin, AjaxResponseMixin, JSONResponseMixin, CacheMixin, ) from apps.core.utils import get_client_ip, log_user_activity from .models import ( NotificationType, Notification, NotificationPreference, NotificationTemplate, NotificationQueue, NotificationHistory, ) from .serializers import ( NotificationSerializer, NotificationTypeSerializer, NotificationPreferenceSerializer, NotificationStatsSerializer, NotificationCreateSerializer, NotificationUpdateSerializer, ) from .forms import ( NotificationFilterForm, NotificationPreferenceForm, NotificationCreateForm, BulkNotificationForm, ) from .services import ( NotificationService, EmailService, TemplateService, PreferenceService, ) class NotificationPagination(PageNumberPagination): """ Custom pagination for notifications API. """ page_size = 20 page_size_query_param = 'page_size' max_page_size = 100 # Web Views class NotificationListView(LoginRequiredMixin, ListView): """ Display a list of notifications for the current user. This view shows all notifications for the authenticated user with filtering and pagination capabilities. """ model = Notification template_name = 'notifications/notification_list.html' context_object_name = 'notifications' paginate_by = 20 def get_queryset(self): """ Get notifications for the current user with optional filtering. Returns: QuerySet: Filtered notifications """ queryset = Notification.objects.filter( recipient=self.request.user ).select_related( 'notification_type' ).order_by('-created_at') # Apply filters from form form = NotificationFilterForm(self.request.GET) if form.is_valid(): # Filter by type if form.cleaned_data.get('notification_type'): queryset = queryset.filter( notification_type=form.cleaned_data['notification_type'] ) # Filter by read status read_status = form.cleaned_data.get('read_status') if read_status == 'read': queryset = queryset.filter(is_read=True) elif read_status == 'unread': queryset = queryset.filter(is_read=False) # Filter by archived status if form.cleaned_data.get('show_archived'): queryset = queryset.filter(is_archived=True) else: queryset = queryset.filter(is_archived=False) # Filter by date range date_from = form.cleaned_data.get('date_from') date_to = form.cleaned_data.get('date_to') if date_from: queryset = queryset.filter(created_at__gte=date_from) if date_to: queryset = queryset.filter(created_at__lte=date_to) # Search in title and message search = form.cleaned_data.get('search') if search: queryset = queryset.filter( Q(title__icontains=search) | Q(message__icontains=search) ) return queryset def get_context_data(self, **kwargs): """ Add additional context data. Args: **kwargs: Additional keyword arguments Returns: dict: Context data """ context = super().get_context_data(**kwargs) # Add filter form context['filter_form'] = NotificationFilterForm(self.request.GET) # Add statistics user_notifications = Notification.objects.filter(recipient=self.request.user) context['stats'] = { 'total': user_notifications.count(), 'unread': user_notifications.filter(is_read=False).count(), 'archived': user_notifications.filter(is_archived=True).count(), 'today': user_notifications.filter( created_at__date=timezone.now().date() ).count(), } # Add notification types for filtering context['notification_types'] = NotificationType.objects.active() return context class NotificationDetailView(LoginRequiredMixin, DetailView): """ Display detailed view of a notification. This view shows the full details of a notification and automatically marks it as read when viewed. """ model = Notification template_name = 'notifications/notification_detail.html' context_object_name = 'notification' def get_queryset(self): """ Get notifications for the current user only. Returns: QuerySet: User's notifications """ return Notification.objects.filter( recipient=self.request.user ).select_related('notification_type') def get_object(self, queryset=None): """ Get the notification object and mark it as read. Args: queryset: Optional queryset Returns: Notification: The notification object """ obj = super().get_object(queryset) # Mark as read if not already read if not obj.is_read: obj.mark_as_read() # Log activity log_user_activity( user=self.request.user, action='notification_read', description=f'Read notification: {obj.title}', ip_address=get_client_ip(self.request), user_agent=self.request.META.get('HTTP_USER_AGENT', ''), extra_data={'notification_id': obj.id} ) return obj def get_context_data(self, **kwargs): """ Add additional context data. Args: **kwargs: Additional keyword arguments Returns: dict: Context data """ context = super().get_context_data(**kwargs) # Add related notifications context['related_notifications'] = Notification.objects.filter( recipient=self.request.user, notification_type=self.object.notification_type ).exclude( id=self.object.id ).order_by('-created_at')[:5] return context class NotificationPreferencesView(LoginRequiredMixin, TemplateView): """ Display and manage user notification preferences. This view allows users to configure their notification preferences for different notification types. """ template_name = 'notifications/preferences.html' def get_context_data(self, **kwargs): """ Add notification preferences to context. Args: **kwargs: Additional keyword arguments Returns: dict: Context data """ context = super().get_context_data(**kwargs) # Get or create preferences for all notification types notification_types = NotificationType.objects.active() preferences = [] for notification_type in notification_types: preference, created = NotificationPreference.objects.get_or_create( user=self.request.user, notification_type=notification_type, defaults={ 'is_enabled': notification_type.default_enabled, 'email_enabled': notification_type.supports_email, 'in_app_enabled': notification_type.supports_in_app, } ) preferences.append({ 'type': notification_type, 'preference': preference, 'form': NotificationPreferenceForm(instance=preference) }) context['preferences'] = preferences return context def post(self, request, *args, **kwargs): """ Handle preference updates. Args: request: HTTP request *args: Additional arguments **kwargs: Additional keyword arguments Returns: HttpResponse: Response """ updated_count = 0 for key, value in request.POST.items(): if key.startswith('preference_'): try: preference_id = int(key.split('_')[1]) preference = NotificationPreference.objects.get( id=preference_id, user=request.user ) form = NotificationPreferenceForm(request.POST, instance=preference) if form.is_valid(): form.save() updated_count += 1 except (ValueError, NotificationPreference.DoesNotExist): continue if updated_count > 0: messages.success( request, _(f'Updated {updated_count} notification preferences.') ) # Log activity log_user_activity( user=request.user, action='preferences_updated', description=f'Updated {updated_count} notification preferences', ip_address=get_client_ip(request), user_agent=request.META.get('HTTP_USER_AGENT', '') ) else: messages.warning(request, _('No preferences were updated.')) return redirect('notifications:preferences') class NotificationStatsView(LoginRequiredMixin, TemplateView): """ Display notification statistics for the current user. This view shows various statistics about the user's notifications including counts, trends, and summaries. """ template_name = 'notifications/stats.html' def get_context_data(self, **kwargs): """ Add notification statistics to context. Args: **kwargs: Additional keyword arguments Returns: dict: Context data """ context = super().get_context_data(**kwargs) user_notifications = Notification.objects.filter(recipient=self.request.user) # Basic statistics context['stats'] = { 'total': user_notifications.count(), 'unread': user_notifications.filter(is_read=False).count(), 'read': user_notifications.filter(is_read=True).count(), 'archived': user_notifications.filter(is_archived=True).count(), 'today': user_notifications.filter( created_at__date=timezone.now().date() ).count(), 'this_week': user_notifications.filter( created_at__gte=timezone.now() - timedelta(days=7) ).count(), 'this_month': user_notifications.filter( created_at__gte=timezone.now() - timedelta(days=30) ).count(), } # Statistics by type context['type_stats'] = user_notifications.values( 'notification_type__name', 'notification_type__color' ).annotate( count=Count('id'), unread_count=Count('id', filter=Q(is_read=False)) ).order_by('-count') # Daily statistics for the last 30 days thirty_days_ago = timezone.now() - timedelta(days=30) daily_stats = [] for i in range(30): date = thirty_days_ago + timedelta(days=i) count = user_notifications.filter( created_at__date=date.date() ).count() daily_stats.append({ 'date': date.strftime('%Y-%m-%d'), 'count': count }) context['daily_stats'] = daily_stats # Recent activity context['recent_notifications'] = user_notifications.order_by( '-created_at' )[:10] return context class NotificationCreateView(AdminRequiredMixin, CreateView): """ Create a new notification (Admin only). This view allows administrators to create and send notifications to users. """ model = Notification form_class = NotificationCreateForm template_name = 'notifications/notification_create.html' success_url = reverse_lazy('notifications:admin_list') def form_valid(self, form): """ Handle valid form submission. Args: form: Valid form instance Returns: HttpResponse: Response """ notification = form.save() # Send the notification try: NotificationService.send_notification(notification) messages.success( self.request, _('Notification created and sent successfully.') ) except Exception as e: messages.warning( self.request, _(f'Notification created but sending failed: {e}') ) # Log activity log_user_activity( user=self.request.user, action='notification_created', description=f'Created notification: {notification.title}', ip_address=get_client_ip(self.request), user_agent=self.request.META.get('HTTP_USER_AGENT', ''), extra_data={'notification_id': notification.id} ) return super().form_valid(form) class BulkNotificationView(AdminRequiredMixin, FormView): """ Send bulk notifications (Admin only). This view allows administrators to send notifications to multiple users at once. """ form_class = BulkNotificationForm template_name = 'notifications/bulk_notification.html' success_url = reverse_lazy('notifications:admin_list') def form_valid(self, form): """ Handle valid form submission. Args: form: Valid form instance Returns: HttpResponse: Response """ recipients = form.cleaned_data['recipients'] notification_type = form.cleaned_data['notification_type'] title = form.cleaned_data['title'] message = form.cleaned_data['message'] action_url = form.cleaned_data.get('action_url') # Create and send notifications created_count = 0 sent_count = 0 for recipient in recipients: try: notification = Notification.objects.create( recipient=recipient, notification_type=notification_type, title=title, message=message, action_url=action_url ) created_count += 1 # Send the notification try: NotificationService.send_notification(notification) sent_count += 1 except Exception: pass # Continue with other notifications except Exception: continue # Skip this recipient messages.success( self.request, _(f'Created {created_count} notifications, sent {sent_count} successfully.') ) # Log activity log_user_activity( user=self.request.user, action='bulk_notification_sent', description=f'Sent bulk notification to {created_count} users', ip_address=get_client_ip(self.request), user_agent=self.request.META.get('HTTP_USER_AGENT', ''), extra_data={ 'recipients_count': created_count, 'sent_count': sent_count, 'title': title } ) return super().form_valid(form) # AJAX Views @method_decorator(login_required, name='dispatch') class NotificationMarkReadView(AjaxResponseMixin, View): """ Mark a notification as read via AJAX. This view handles AJAX requests to mark notifications as read or unread. """ def post(self, request, *args, **kwargs): """ Handle POST request to mark notification as read. Args: request: HTTP request *args: Additional arguments **kwargs: Additional keyword arguments Returns: JsonResponse: JSON response """ try: notification_id = request.POST.get('notification_id') action = request.POST.get('action', 'read') # 'read' or 'unread' notification = get_object_or_404( Notification, id=notification_id, recipient=request.user ) if action == 'read': success = notification.mark_as_read() message = _('Notification marked as read.') else: success = notification.mark_as_unread() message = _('Notification marked as unread.') if success: return JsonResponse({ 'success': True, 'message': str(message), 'is_read': notification.is_read, 'read_at': notification.read_at.isoformat() if notification.read_at else None }) else: return JsonResponse({ 'success': False, 'message': _('No changes were made.') }) except Exception as e: return JsonResponse({ 'success': False, 'message': str(e) }, status=400) @method_decorator(login_required, name='dispatch') class NotificationArchiveView(AjaxResponseMixin, View): """ Archive/unarchive a notification via AJAX. This view handles AJAX requests to archive or unarchive notifications. """ def post(self, request, *args, **kwargs): """ Handle POST request to archive notification. Args: request: HTTP request *args: Additional arguments **kwargs: Additional keyword arguments Returns: JsonResponse: JSON response """ try: notification_id = request.POST.get('notification_id') action = request.POST.get('action', 'archive') # 'archive' or 'unarchive' notification = get_object_or_404( Notification, id=notification_id, recipient=request.user ) if action == 'archive': success = notification.archive() message = _('Notification archived.') else: success = notification.unarchive() message = _('Notification unarchived.') if success: return JsonResponse({ 'success': True, 'message': str(message), 'is_archived': notification.is_archived }) else: return JsonResponse({ 'success': False, 'message': _('No changes were made.') }) except Exception as e: return JsonResponse({ 'success': False, 'message': str(e) }, status=400) @method_decorator(login_required, name='dispatch') class NotificationBulkActionView(AjaxResponseMixin, View): """ Handle bulk actions on notifications via AJAX. This view handles bulk operations like marking multiple notifications as read or archiving them. """ def post(self, request, *args, **kwargs): """ Handle POST request for bulk actions. Args: request: HTTP request *args: Additional arguments **kwargs: Additional keyword arguments Returns: JsonResponse: JSON response """ try: notification_ids = request.POST.getlist('notification_ids[]') action = request.POST.get('action') if not notification_ids or not action: return JsonResponse({ 'success': False, 'message': _('Missing notification IDs or action.') }, status=400) notifications = Notification.objects.filter( id__in=notification_ids, recipient=request.user ) updated_count = 0 if action == 'mark_read': for notification in notifications: if notification.mark_as_read(): updated_count += 1 message = _(f'Marked {updated_count} notifications as read.') elif action == 'mark_unread': for notification in notifications: if notification.mark_as_unread(): updated_count += 1 message = _(f'Marked {updated_count} notifications as unread.') elif action == 'archive': for notification in notifications: if notification.archive(): updated_count += 1 message = _(f'Archived {updated_count} notifications.') elif action == 'unarchive': for notification in notifications: if notification.unarchive(): updated_count += 1 message = _(f'Unarchived {updated_count} notifications.') elif action == 'delete': deleted_count = notifications.delete()[0] message = _(f'Deleted {deleted_count} notifications.') updated_count = deleted_count else: return JsonResponse({ 'success': False, 'message': _('Invalid action.') }, status=400) return JsonResponse({ 'success': True, 'message': str(message), 'updated_count': updated_count }) except Exception as e: return JsonResponse({ 'success': False, 'message': str(e) }, status=400) # API Views class NotificationAPIView(ListCreateAPIView): """ API view for listing and creating notifications. This view provides RESTful access to notifications with filtering, searching, and pagination. """ serializer_class = NotificationSerializer permission_classes = [permissions.IsAuthenticated] pagination_class = NotificationPagination filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter] filterset_fields = ['notification_type', 'is_read', 'is_archived'] search_fields = ['title', 'message'] ordering_fields = ['created_at', 'read_at'] ordering = ['-created_at'] def get_queryset(self): """ Get notifications for the current user. Returns: QuerySet: User's notifications """ return Notification.objects.filter( recipient=self.request.user ).select_related('notification_type') def get_serializer_class(self): """ Get the appropriate serializer class. Returns: Serializer: Serializer class """ if self.request.method == 'POST': return NotificationCreateSerializer return NotificationSerializer def perform_create(self, serializer): """ Create a new notification. Args: serializer: Validated serializer """ notification = serializer.save() # Send the notification try: NotificationService.send_notification(notification) except Exception: pass # Log error but don't fail the request class NotificationDetailAPIView(RetrieveUpdateDestroyAPIView): """ API view for retrieving, updating, and deleting notifications. This view provides detailed access to individual notifications with automatic read marking on retrieval. """ serializer_class = NotificationSerializer permission_classes = [permissions.IsAuthenticated] def get_queryset(self): """ Get notifications for the current user. Returns: QuerySet: User's notifications """ return Notification.objects.filter( recipient=self.request.user ).select_related('notification_type') def get_serializer_class(self): """ Get the appropriate serializer class. Returns: Serializer: Serializer class """ if self.request.method in ['PUT', 'PATCH']: return NotificationUpdateSerializer return NotificationSerializer def retrieve(self, request, *args, **kwargs): """ Retrieve a notification and mark it as read. Args: request: HTTP request *args: Additional arguments **kwargs: Additional keyword arguments Returns: Response: API response """ instance = self.get_object() # Mark as read if not already read if not instance.is_read: instance.mark_as_read() serializer = self.get_serializer(instance) return Response(serializer.data) class NotificationStatsAPIView(APIView): """ API view for notification statistics. This view provides statistical data about the user's notifications in JSON format. """ permission_classes = [permissions.IsAuthenticated] def get(self, request, *args, **kwargs): """ Get notification statistics. Args: request: HTTP request *args: Additional arguments **kwargs: Additional keyword arguments Returns: Response: Statistics data """ user_notifications = Notification.objects.filter(recipient=request.user) # Calculate statistics stats = { 'total': user_notifications.count(), 'unread': user_notifications.filter(is_read=False).count(), 'read': user_notifications.filter(is_read=True).count(), 'archived': user_notifications.filter(is_archived=True).count(), 'today': user_notifications.filter( created_at__date=timezone.now().date() ).count(), 'this_week': user_notifications.filter( created_at__gte=timezone.now() - timedelta(days=7) ).count(), 'this_month': user_notifications.filter( created_at__gte=timezone.now() - timedelta(days=30) ).count(), } # Statistics by type type_stats = list(user_notifications.values( 'notification_type__name', 'notification_type__color', 'notification_type__icon' ).annotate( count=Count('id'), unread_count=Count('id', filter=Q(is_read=False)) ).order_by('-count')) serializer = NotificationStatsSerializer({ 'stats': stats, 'type_stats': type_stats }) return Response(serializer.data) class NotificationPreferencesAPIView(ListAPIView): """ API view for notification preferences. This view provides access to user notification preferences with the ability to update them. """ serializer_class = NotificationPreferenceSerializer permission_classes = [permissions.IsAuthenticated] def get_queryset(self): """ Get preferences for the current user. Returns: QuerySet: User's preferences """ return NotificationPreference.objects.filter( user=self.request.user ).select_related('notification_type') def post(self, request, *args, **kwargs): """ Update notification preferences. Args: request: HTTP request *args: Additional arguments **kwargs: Additional keyword arguments Returns: Response: Updated preferences """ preferences_data = request.data.get('preferences', []) updated_preferences = [] for pref_data in preferences_data: try: preference = NotificationPreference.objects.get( id=pref_data.get('id'), user=request.user ) serializer = NotificationPreferenceSerializer( preference, data=pref_data, partial=True ) if serializer.is_valid(): serializer.save() updated_preferences.append(serializer.data) except NotificationPreference.DoesNotExist: continue return Response({ 'message': f'Updated {len(updated_preferences)} preferences.', 'preferences': updated_preferences }) # Utility API Views @api_view(['POST']) @permission_classes([permissions.IsAuthenticated]) def mark_notification_read_api(request, notification_id): """ Mark a notification as read via API. Args: request: HTTP request notification_id: Notification ID Returns: Response: API response """ try: notification = get_object_or_404( Notification, id=notification_id, recipient=request.user ) success = notification.mark_as_read() return Response({ 'success': success, 'message': 'Notification marked as read.' if success else 'No changes made.', 'is_read': notification.is_read, 'read_at': notification.read_at }) except Exception as e: return Response({ 'success': False, 'message': str(e) }, status=status.HTTP_400_BAD_REQUEST) @api_view(['POST']) @permission_classes([permissions.IsAuthenticated]) def mark_all_read_api(request): """ Mark all notifications as read via API. Args: request: HTTP request Returns: Response: API response """ try: updated_count = 0 notifications = Notification.objects.filter( recipient=request.user, is_read=False ) for notification in notifications: if notification.mark_as_read(): updated_count += 1 return Response({ 'success': True, 'message': f'Marked {updated_count} notifications as read.', 'updated_count': updated_count }) except Exception as e: return Response({ 'success': False, 'message': str(e) }, status=status.HTTP_400_BAD_REQUEST) @api_view(['GET']) @permission_classes([permissions.IsAuthenticated]) def notification_count_api(request): """ Get notification counts via API. Args: request: HTTP request Returns: Response: Notification counts """ user_notifications = Notification.objects.filter(recipient=request.user) counts = { 'total': user_notifications.count(), 'unread': user_notifications.filter(is_read=False).count(), 'archived': user_notifications.filter(is_archived=True).count(), } return Response(counts) @api_view(['POST']) @permission_classes([permissions.IsAdminUser]) def send_notification_api(request): """ Send a notification via API (Admin only). Args: request: HTTP request Returns: Response: API response """ serializer = NotificationCreateSerializer(data=request.data) if serializer.is_valid(): notification = serializer.save() # Send the notification try: NotificationService.send_notification(notification) return Response({ 'success': True, 'message': 'Notification sent successfully.', 'notification': NotificationSerializer(notification).data }, status=status.HTTP_201_CREATED) except Exception as e: return Response({ 'success': False, 'message': f'Notification created but sending failed: {e}', 'notification': NotificationSerializer(notification).data }, status=status.HTTP_201_CREATED) return Response({ 'success': False, 'message': 'Invalid data.', 'errors': serializer.errors }, status=status.HTTP_400_BAD_REQUEST)