from rest_framework import generics, status from rest_framework.response import Response from rest_framework.authentication import TokenAuthentication from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi from rest_framework.permissions import IsAuthenticated from apps.account.serializers import NotificationSerializer, NotificationSendSerializer, AdminNotificationSerializer from apps.account.models import Notification, User from apps.account.tasks import send_notification from utils.pagination import StandardResultsSetPagination from django.db.models import Q from rest_framework.viewsets import ModelViewSet from apps.account.permissions import IsSuperAdmin import logging logger = logging.getLogger(__name__) class NotificationListView(generics.ListAPIView): queryset = Notification.objects.all() serializer_class = NotificationSerializer permission_classes = [IsAuthenticated,] authentication_classes = [TokenAuthentication] pagination_class = StandardResultsSetPagination @swagger_auto_schema( operation_description="Retrieve a list of notifications for the authenticated user or merchant account.", tags=['Notifications'], manual_parameters=[ openapi.Parameter( 'service', openapi.IN_QUERY, description="Filter notifications by service (imam-javad or doboodi)", type=openapi.TYPE_STRING, enum=['imam-javad', 'doboodi'], required=False ) ] ) def get(self, request, *args, **kwargs): """ This API allows you to retrieve a list of notifications based on the authenticated user's type. If the user is a regular user, their notifications will be fetched from the `Notification` model. If the user is a merchant, their notifications will be fetched from the `MerchantAccountNotification` model. - **Method**: GET - **URL**: /api/notifications/ - **Query Parameters**: - `service`: Optional. Filter notifications by service ('imam-javad' or 'doboodi') - **Response**: Includes details of notifications such as title, message, is read status, service, creation date, and update date. - **Headers**: `Authorization: Bearer ` for authentication. """ return super().get(request, *args, **kwargs) def get_queryset(self): user = self.request.user queryset = Notification.objects.filter(user=user) # Filter by service if provided in query params service = self.request.query_params.get('service', None) if service: queryset = queryset.filter(service=service) return queryset.order_by('-created_at') class NotificationReadAllView(generics.GenericAPIView): permission_classes = [IsAuthenticated,] authentication_classes = [TokenAuthentication] queryset = Notification.objects.all() @swagger_auto_schema( operation_description="Mark all notifications as read for the authenticated user.", tags=['Notifications'], responses={ 200: "All notifications marked as read", } ) def get(self, request, *args, **kwargs): user = request.user # Mark all of the user's notifications as read Notification.objects.filter(user=user).update(is_read=True) return Response({'status': 'all notifications marked as read'}, status=status.HTTP_200_OK) class SendNotificationView(generics.GenericAPIView): @swagger_auto_schema( operation_description="Send a notification to a user by user_id.", tags=['Notifications'], request_body=openapi.Schema( type=openapi.TYPE_OBJECT, required=['user_id', 'title', 'body'], properties={ 'user_id': openapi.Schema(type=openapi.TYPE_INTEGER, description='ID of the user to send notification to'), 'title': openapi.Schema(type=openapi.TYPE_STRING, description='Notification title'), 'body': openapi.Schema(type=openapi.TYPE_STRING, description='Notification body'), 'data': openapi.Schema(type=openapi.TYPE_OBJECT, description='Additional data payload', default={'slam': 'qatreh'}), }, ), responses={ 200: openapi.Response('Notification sent successfully.'), 400: openapi.Response('FCM token not available for this user.'), 404: openapi.Response('User not found.'), 500: openapi.Response('Internal server error.'), } ) def post(self, request, *args, **kwargs): user_id = request.data.get('user_id', 1) try: user = User.objects.get(id=user_id) except User.DoesNotExist: return Response({'error': 'User not found.'}, status=status.HTTP_404_NOT_FOUND) notification_title = request.data.get('title', 'test qatreh') notification_body = request.data.get('body', 'test qatreh body') data_payload = request.data.get('data', {'slam':'qatreh'}) fcm_token = user.fcm # Ensure that 'fcm' is a field in your User model if not fcm_token: return Response({ 'error': 'FCM token not available for this user.' }, status=status.HTTP_400_BAD_REQUEST) try: send_notification([fcm_token], notification_title, notification_body, data_payload) return Response({ 'message': 'Notification sent successfully.' }, status=status.HTTP_200_OK) except Exception as e: return Response({ 'error': str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class AdminNotificationViewSet(ModelViewSet): permission_classes = [IsSuperAdmin] authentication_classes = [TokenAuthentication] serializer_class = AdminNotificationSerializer pagination_class = StandardResultsSetPagination def get_queryset(self): queryset = Notification.objects.all().select_related('user') # Filter by service if provided service = self.request.query_params.get('service', None) if service: queryset = queryset.filter(service=service) # Search by user name or email or notification title/message search = self.request.query_params.get('search', None) if search: queryset = queryset.filter( Q(title__icontains=search) | Q(message__icontains=search) | Q(user__fullname__icontains=search) | Q(user__email__icontains=search) ) # Date filtering date_after = self.request.query_params.get('date_after', None) if date_after: queryset = queryset.filter(created_at__date__gte=date_after) date_before = self.request.query_params.get('date_before', None) if date_before: queryset = queryset.filter(created_at__date__lte=date_before) # User filtering user_param = self.request.query_params.get('user', None) if user_param: queryset = queryset.filter(user_id=user_param) # Read status filtering (is_read=true/false) is_read_param = self.request.query_params.get('is_read', None) if is_read_param is not None: if is_read_param.lower() == 'true': queryset = queryset.filter(is_read=True) elif is_read_param.lower() == 'false': queryset = queryset.filter(is_read=False) return queryset.order_by('-created_at') def create(self, request, *args, **kwargs): title = request.data.get('title') message = request.data.get('message') service = request.data.get('service', 'imam-javad') send_to_all = request.data.get('send_to_all', False) user_id = request.data.get('user_id', None) if not title or not message: return Response({'error': 'Title and message are required.'}, status=status.HTTP_400_BAD_REQUEST) if send_to_all: # Get all active users users = User.objects.filter(is_active=True) notifications_to_create = [] for u in users: notifications_to_create.append(Notification( title=title, message=message, service=service, user=u )) Notification.objects.bulk_create(notifications_to_create) # Send push notifications fcm_tokens = list(users.exclude(fcm__isnull=True).exclude(fcm='').values_list('fcm', flat=True)) if fcm_tokens: try: send_notification(fcm_tokens, title, message, {'service': service}) except Exception as e: logger.error(f"Failed to send push notifications: {e}") return Response({'message': f'Notification created for {len(notifications_to_create)} users.'}, status=status.HTTP_201_CREATED) else: if not user_id: return Response({'error': 'user_id is required when send_to_all is false.'}, status=status.HTTP_400_BAD_REQUEST) try: user = User.objects.get(id=user_id) except User.DoesNotExist: return Response({'error': 'User not found.'}, status=status.HTTP_404_NOT_FOUND) notif = Notification.objects.create( title=title, message=message, service=service, user=user ) # Send push notification if user.fcm: try: send_notification([user.fcm], title, message, {'service': service}) except Exception as e: logger.error(f"Failed to send push notification: {e}") serializer = self.get_serializer(notif) return Response(serializer.data, status=status.HTTP_201_CREATED)