from rest_framework.response import Response from rest_framework.permissions import AllowAny,IsAdminUser,IsAuthenticated from rest_framework_simplejwt.tokens import RefreshToken from django.contrib.auth.models import Group from rest_framework.pagination import PageNumberPagination from .serializers import * from rest_framework import viewsets, status, permissions from rest_framework.views import APIView from django.contrib.auth import authenticate from django.contrib.auth.hashers import check_password from django.contrib.auth.models import Permission from django.contrib.contenttypes.models import ContentType from django.db.models import Q from .models import * from .permissions import IsAdminOrReadOnly, IsManagerOrAdmin, IsExecutiveOrAdmin # Custom permission class import numpy as np # import cv2 from django.http import JsonResponse import requests from django.core.files.uploadedfile import InMemoryUploadedFile from celery import Celery from django.core.management import call_command import csv from django.http import HttpResponse from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib import colors from reportlab.platypus.tables import Table, TableStyle from io import BytesIO # from django.utils import timezone from openpyxl import Workbook from django.db.models import Count from datetime import timedelta, date, time # from django.utils.timezone import now from collections import defaultdict from django.core.exceptions import ObjectDoesNotExist from myprofile.filterfunction import validate_dates from rest_framework import filters from decimal import Decimal # admin_group, _ = Group.objects.get_or_create(name='Admin') # run in python shell # manager_group, _ = Group.objects.get_or_create(name='Manager') # executive_group, _ = Group.objects.get_or_create(name='Executive') class StandardResultsSetPagination(PageNumberPagination): page_size = 10 page_size_query_param = 'page_size' lookup_url_kwarg = 'pk' class GroupCreateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def create(self, request, *args, **kwargs): try: user = request.user name = request.data.get('name') permission_ids = request.data.get('permission_id',[]) if not user.has_perm('auth.add_group'): return Response({'status': False, 'message': 'You are not authorized.!'}) if name and permission_ids: grp = Group.objects.filter(name = name).first() if grp: return Response({'status':False,'message': 'Group already exists!'}) permissions_to_assign = Permission.objects.filter(id__in=permission_ids) group = Group(name = name) group.save() group.permissions.add(*permissions_to_assign) group.save() return Response({ 'status': True, 'message': 'Group created successfully.'}) else: return Response({'status':False,'message': 'name and permissions are requred!'}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) # def create(self, request): # group_name = request.data.get('group_name') # if not group_name: # return Response({"status": False,'error': 'Group name is required.'}, status = 400) # if Group.objects.filter(name= group_name).exists(): # return Response({"status": False,'error': f'Group "{group_name}" already exists.'}, status = 400) # created = Group.objects.create(name = group_name) # if created: # return Response({"status": True,'message': f'Group "{group_name}" created successfully.'}, status = 200) # else: # return Response({"status": False,'error': f'Failed to create group "{group_name}".'}) def destroy(self, request, *args, **kwargs): try: group_id = self.kwargs.get('group_id') user = request.user if not user.has_perm('auth.delete_group'): return Response({'status': False, 'message': 'You are not authorized.!'}) if not group_id: return Response({'status': False, 'message': 'Group id is required'}) group = Group.objects.filter(id=group_id).first() group.delete() return Response({"status": True,'message':'Group Deleted successfully.'}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def list(self, request): try: group = Group.objects.all() return Response({"status": True,'data':GroupUserSerializer(group,many=True).data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class GroupListAPIView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): try: user = request.user name_filter = request.GET.get('name') groups = Group.objects.all().exclude(name = 'super_admin') if not user.has_perm('auth.view_group'): return Response({'status': False, 'message': 'You are not authorized.!'}) if name_filter: groups = groups.filter(name__icontains=name_filter) groups_list = [] for group in groups: groups_list.append({ 'group_id': group.id, 'group_name': str(group) }) return Response({ 'status':True, 'message':'Group list!', 'data':groups_list}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class GroupPermissionAssignAPIView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def create(self, request, *args, **kwargs): try: # Check if the user is a superadmin if not request.user.groups.filter(name='super_admin').exists(): return Response({'status': False,'message': 'Only SuperAdmins can assign/unassign permissions.'}) # Get the data from the request group_id = request.data.get('group_id') name = request.data.get('name') permission_ids = request.data.get('permission_id',[]) action = request.data.get('action') if not permission_ids or not action: return Response({'status': False,'message': 'Missing or null data'}) # Retrieve the group object try: group = Group.objects.filter(id=group_id).first() except Group.DoesNotExist: return Response({'status': False, 'message': 'Invalid group ID'}) if name: group.name = name group.save() # Assign or unassign the permission based on the action permissions_to_unassign = Permission.objects.all() group.permissions.remove(*permissions_to_unassign) for user in group.user_set.all(): user.user_permissions.remove(*permissions_to_unassign) if action == 'assign': permissions_to_assign = Permission.objects.filter(id__in=permission_ids) group.permissions.add(*permissions_to_assign) for user in group.user_set.all(): user.user_permissions.add(*permissions_to_assign) return Response({ 'status': True, 'message': 'Permissions assigned successfully!' }) else: return Response({'status':False,'message': 'Invalid action!'}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class PermissionListView(viewsets.ViewSet): permission_classes = [IsAuthenticated] # filterset_class = PermissionFilter def list(self, request, *args, **kwargs): try: if not request.user.has_perm('auth.view_permission'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) group_id = request.GET.get('group_id', None) if group_id: group_id = int(group_id) user = request.user group_permissions_ids = [] if group_id is not None: try: group = Group.objects.filter(id=group_id).first() group_permissions_ids = group.permissions.values_list('id', flat=True) except Group.DoesNotExist: return Response({'status': False, 'message': 'Group not found!'}) # Get all available permissions for the model all_permissions = Permission.objects.all() permission_dict = {} for permission in all_permissions: model_name = permission.content_type.model if model_name not in permission_dict: permission_dict[model_name] = { 'name': model_name, 'add': None, 'change': None, 'delete': None, 'view': None, 'isAdd': False, 'isChange': False, 'isDelete': False, 'isView': False, } # Check which permission is available and update the dictionary # Additionally, check if the permission is in the specified group if permission.codename == f'add_{model_name}': permission_dict[model_name]['isAdd'] = permission.id in group_permissions_ids if group_permissions_ids else False permission_dict[model_name]['add'] = permission.id elif permission.codename == f'change_{model_name}': permission_dict[model_name]['isChange'] = permission.id in group_permissions_ids if group_permissions_ids else False permission_dict[model_name]['change'] = permission.id elif permission.codename == f'delete_{model_name}': permission_dict[model_name]['isDelete'] = permission.id in group_permissions_ids if group_permissions_ids else False permission_dict[model_name]['delete'] = permission.id elif permission.codename == f'view_{model_name}': permission_dict[model_name]['isView'] = permission.id in group_permissions_ids if group_permissions_ids else False permission_dict[model_name]['view'] = permission.id # Convert dictionary to a list permission_data = list(permission_dict.values()) return Response({ 'status': True, 'message': 'Permission List!', 'data': permission_data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) # permission_classes = [permissions.IsAuthenticated] # def list(self, request): # user = request.user # if not user.has_perm('auth.view_permission'): # return Response({'status': False, 'message': 'Permission denied!'}) # permission_data = {} # user_permissions = Permission.objects.filter(user=user) # for permission in user_permissions: # content_type = permission.content_type.model # if content_type not in permission_data: # permission_data[content_type] = {} # action_name = permission.codename.split("_")[0] # permission_data[content_type][action_name] = permission.id # formatted_data = [ # {content_type: permissions} # for content_type, permissions in permission_data.items() # ] # return Response({'status': True,'message': 'Permission List!','data': formatted_data},status = 200) class UserPermissionAssignView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def create(self, request): # Check if the user is a Admin if not request.user.groups.filter(name='super_admin').exists(): return Response({'status': False,'message': 'Only Admin can assign/unassign permissions.'},status = 400) user_id = request.data.get('user_id') permission_ids = request.data.get('permission_id') action = request.data.get('action') if not user_id or not permission_ids or not action: return Response({'status': False,'message': 'Missing or null data'},status = 400) if action not in ['assign', 'unassign']: return Response({'status': False,'message': 'Invalid action. Action must be "assign" or "unassign".'},status = 400) # Convert the permission IDs to a list of integers permission_id = [int(id) for id in permission_ids.split(',')] # Retrieve the user and permission objects try: user = CustomUser.objects.get(id = user_id) permission = Permission.objects.filter(id__in = permission_id) except CustomUser.DoesNotExist: return Response({'status':False ,'message': 'User Not Found'},status = 400) except Permission.DoesNotExist: return Response({'status':False ,'message': 'PermissionId not found'},status = 400) # Assign or unassign the permission based on the action if action == 'assign': user.user_permissions.add(*permission) return Response({'status':True,'message': 'Permission assigned successfully!',},status = 200) elif action == 'unassign': user.user_permissions.remove(*permission) return Response({'status':True,'message': 'Permission Unassigned successfully!'},status = 200) else: return Response({'status':False,'error': 'Invalid action'},status = 400) class UserRegisterView(viewsets.ModelViewSet): permission_classes = [AllowAny] queryset = CustomUser.objects.all().exclude(is_superuser=True) serializer_class = CustomUserSerializer pagination_class = StandardResultsSetPagination def send_email_touser(self, user_email, user_password): subject = 'User Registered credentials' message = f'You registration is successfully completed.\n\n Username: {user_email}\nPassword: {user_password}' from_email= settings.EMAIL_HOST_USER email = EmailMessage(subject, message, from_email, to=[user_email]) email.send() def create(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.add_customuser'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) errors = {} first_name = request.data.get('first_name') last_name = request.data.get('last_name') profile_image = request.data.get('profile_image') email = request.data.get('email').lower() password = request.data.get('password') group = request.data.get('group') org = Organization.objects.filter(organization_user=user).first() if not email: return Response({"status": False,'message': 'email is required',"data":[]}) existing_email = CustomUser.objects.filter(email=email).first() if existing_email: return Response({"status": False, 'message': 'Email is already exists.',"data":{}}) if not password: return Response({"status": False,'message': 'Password is required',"data":[]}) if not first_name: return Response({"status": False,'message': 'First name is required',"data":[]}) if not last_name: return Response({"status": False,'message': 'Last name is required',"data":[]}) if not org: return Response({"status": False,'message': 'Organization not found.',"data":[]}) if errors: return Response({"status": False,'error': errors}, status = 400) else: if group: serializer = CustomUserSerializer(data=request.data,context={'request': request}) if serializer.is_valid(): serializer.save() group_details = Group.objects.filter(id = group).first() permission = group_details.permissions.all() group_details.user_set.add(serializer.instance) for per in permission: serializer.instance.user_permissions.add(per) serializer.instance.save() serializer = CustomUserSerializer(serializer.instance,context={'request': request}) group_serializer = GroupSerializer(group_details) org.organization_user.add(serializer.data['id']) return Response({"status": True,'message': f'User {group_details.name} created successfully.',"data":serializer.data,"group":group_serializer.data}) else: return Response({"status": False,'error': serializer.errors}, status = 400) else: return Response({"status": False,'error': 'Please select group'}, status = 400) # if group == 'Admin': # serializer = CustomUserSerializer(data=request.data,context={'request': request}) # if serializer.is_valid(): # serializer.save() # permission = Permission.objects.filter(Q(codename__startswith='add_') | Q(codename__startswith='change_') | Q(codename__startswith='view_')| Q(codename__startswith='delete_')) # admin_group = Group.objects.get(name = 'Admin') # admin_group.user_set.add(serializer.instance) # for per in permission: # serializer.instance.user_permissions.add(per) # serializer.instance.save() # serializer = CustomUserSerializer(serializer.instance,context={'request': request}) # group_serializer = GroupSerializer(admin_group) # self.send_email_touser(email,password) # org.organization_user.add(serializer.data['id']) # return Response({"status": True,'message': 'User Admin created successfully.',"data":serializer.data,"group":group_serializer.data}) # else: # return Response({"status": False,'error': serializer.errors}, status = 400) # if group == 'Manager': # serializer = CustomUserSerializer(data=request.data,context={'request': request}) # if serializer.is_valid(): # serializer.save() # permission = Permission.objects.filter(Q(codename__startswith='add_') | Q(codename__startswith='change_') | Q(codename__startswith='view_')) # manager_group = Group.objects.get(name = 'Manager') # manager_group.user_set.add(serializer.instance) # for per in permission: # serializer.instance.user_permissions.add(per) # serializer.instance.save() # serializer = CustomUserSerializer(serializer.instance,context={'request': request}) # group_serializer = GroupSerializer(manager_group) # self.send_email_touser(email, password) # org.organization_user.add(serializer.data['id']) # return Response({"status": True,'message': 'User Manager created successfully.',"data":serializer.data}, status = 200) # else: # return Response({"status": False,'error': serializer.errors}, status = 400) # if group == 'Executive': # serializer = CustomUserSerializer(data=request.data,context={'request': request}) # if serializer.is_valid(): # serializer.save() # permission = Permission.objects.filter(Q(codename__startswith='view_')) # executive_group = Group.objects.get(name = 'Executive') # executive_group.user_set.add(serializer.instance) # for per in permission: # serializer.instance.user_permissions.add(per) # serializer.instance.save() # serializer = CustomUserSerializer(serializer.instance,context={'request': request}) # group_serializer = GroupSerializer(executive_group) # self.send_email_touser(email, password) # org.organization_user.add(serializer.data['id']) # return Response({"status": True,'message': 'User Executive created successfully.',"data":serializer.data}, status = 200) # else: # return Response({"status": False,'error': serializer.errors}, status = 400) # else: # return Response({"status": False,'error': 'Please select group'}, status = 400) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def list(self,request,*args,**kwargs): try: user = self.request.user if not user.has_perm('myprofile.view_customuser'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) search_query = request.query_params.get('search', '') start_date = request.query_params.get('start_date', '') end_date = request.query_params.get('end_date', '') group_name = request.query_params.get('group_name', '') # users = CustomUser.objects.all().exclude(is_superuser=True).order_by('-created_at') org = Organization.objects.filter(organization_user=user).first() org_user = org.organization_user.all().exclude(is_superuser=True).exclude(groups__name = "super_admin") # if request.user.groups.filter(name='Admin').exists(): # user = user.filter(Q(groups__name = 'Manager') | Q(groups__name = 'Executive')) # elif request.user.groups.filter(name='Manager').exists(): # user = user.filter(Q(groups__name = 'Executive')) # elif request.user.groups.filter(name='Executive').exists(): # user = user.filter(Q(groups__name = 'Executive')) # else: # return Response({'status': False,'message': 'You can not access user list'}, status = 400) # user = users[0] if search_query: search_query = search_query.strip().lower() org_user = org_user.filter(Q(first_name__icontains = search_query) | Q(last_name__icontains = search_query) | Q(email__icontains = search_query)) if start_date or end_date: start_date,end_date, error_message = validate_dates(start_date, end_date) if error_message: return Response({'status': False,'message': error_message,"data":[]}) org_user = org_user.filter(created_at__range=[start_date, end_date]) # else: # if request.user.groups.filter(name= 'Admin').exists(): # user = user.filter(Q(groups__name = 'Executive') | Q(groups__name = 'Manager')) # elif request.user.groups.filter(name= 'Manager').exists(): # user = user.filter(Q(groups__name = 'Executive')) # elif request.user.groups.filter(name= 'Executive').exists(): # user = user.filter(Q(groups__name = 'Executive')) if group_name: if not Group.objects.filter(name=group_name).exists(): return Response({'status': False,'message': 'Invalid group name',"data":[]}) # else: # user = user.filter(groups__name = group_name) if org_user.exists(): page = self.paginate_queryset(org_user) if page is not None: serializer = CustomUserSerializer(page,context={'request': request}, many=True) serializer = self.get_paginated_response(serializer.data) else: serializer = CustomUserSerializer(user,context={'request': request}, many=True) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"User List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({"status": True,'message':'User Not Found',"data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class UserListView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = CustomUser.objects.all() serializer_class = CustomUserSerializer ########## request user can see only his/her profile ############## def list(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.view_customuser'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) org = Organization.objects.filter(organization_user=user)[0] user_obj = org.organization_user.all().exclude(is_superuser=True).exclude(groups__name = "super_admin") if user_obj: serializer = CustomUserSerializer(user_obj,context={'request': request}, many=True) return Response({"status": True, "message":"User List Successfully", "data": serializer.data}) else: return Response({"status": True,'message':'User Not Found',"data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class UserupdateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CustomUserSerializer def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_customuser'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) user = request.user # Check if user is authenticated if not user.is_authenticated: return Response({'status': False, 'message': 'You cannot update user profile'}, status=401) # Get data from request.data first_name = request.data.get('first_name') last_name = request.data.get('last_name') profile_image = request.data.get('profile_image') # Check if required fields are provided if not first_name or not last_name: return Response({'status': False, 'message': 'First name, last name are required'}, status=400) user.first_name = first_name user.last_name = last_name # Only update profile_image if it's provided and not None if profile_image is not None and not isinstance(profile_image, str): user.profile_image = profile_image user.save() # Serialize the user and group separately user_serializer = CustomUserSerializer(user, context={'request': request}) return Response({'status': True,'message': 'User profile updated successfully.','user_data': user_serializer.data,}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'error': str(e)}, status=500) def list(self, request): try: user = request.user if not user.has_perm('myprofile.view_customuser'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) serializer = CustomUserSerializer(user) return Response({'status': True,'data':serializer.data},status = 200) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class UserDeleteView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CustomUserSerializer queryset = CustomUser.objects.all() lookup_field = 'user_id' def update(self, request, *args, **kwargs): try: user_id = self.kwargs.get("user_id") first_name = request.data.get('first_name', None) last_name = request.data.get('last_name',None) group_id = request.data.get('group_name',None) user = request.user if not user.has_perm('myprofile.change_customuser'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) # Check if the user exists if not CustomUser.objects.filter(id=user_id).exists(): return Response({'status': False, 'message': 'User not found',"data":[]}) user = CustomUser.objects.get(id=user_id) user.first_name = first_name user.last_name = last_name user.groups.clear() if group_id: grp = Group.objects.filter(id=group_id).first() user.groups.add(grp) permission = grp.permissions.all() for per in permission: user.user_permissions.add(per) user.save() user.save() # if group_name == 'Manager': # user.groups.add(Group.objects.get(name='Manager')) # permission = Permission.objects.filter(Q(codename__startswith='add_') | Q(codename__startswith='change_') | Q(codename__startswith='view_')) # for per in permission: # user.user_permissions.add(per) # user.save() # if group_name == 'Executive': # user.groups.add(Group.objects.get(name='Executive')) # permission = Permission.objects.filter(Q(codename__startswith='view_')) # for per in permission: # user.user_permissions.add(per) # user.save() # user.save() serializer = CustomUserSerializer(user,context={'request': request}) return Response({"status": True,'message': 'User updated successfully.',"data":serializer.data}, status = 200) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def destroy(self, request, *args, **kwargs): user = self.request.user try: if not user.has_perm('myprofile.delete_customuser'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) user_to_delete = self.kwargs.get("user_id") # Check if the user to delete is not the requester if user_to_delete == user: return Response({'status': False, 'message': 'You cannot delete your own account', 'data': []}) # Check if the user to delete exists if not CustomUser.objects.filter(id=user_to_delete).exists(): return Response({'status': False, 'message': 'User not found', 'data': []}) # Delete the user user = CustomUser.objects.get(id=user_to_delete) user.delete() return Response({'status': True, 'message': 'User deleted successfully'}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) def list(self, request, *args, **kwargs): try: user_id = self.kwargs.get("user_id") user = CustomUser.objects.get(id=user_id) serializer = self.serializer_class(user) return Response({"status": True, "data": serializer.data},status=200) except ObjectDoesNotExist: return Response({"status": False, "message": "User not found."},status=400) except Exception as e: return Response({"status": False, "message": "Something went wrong", "data": str(e)}) class LogoutView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def create(self, request, *args, **kwargs): try: refresh_token = request.data.get('refresh_token') if not refresh_token: return Response({"status": False,'message': 'Refresh token is required'},status = 400) try: token = RefreshToken(refresh_token) token.blacklist() return Response({"status": True,'message': 'Logout successfully'},status = 200) except Exception as e: return Response({"status": False,'message': 'Invalid refresh token'},status = 400) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class UserLoginView(viewsets.ModelViewSet): # permission_classes = [IsAuthenticated] def create(self, request): try: email = request.data.get('email', None).lower() password = request.data.get('password', None) if not email: return Response({"status": False,'message': 'email is required','data':[]}) if not password: return Response({"status": False,'message': 'Password is required','data':[]}) user = CustomUser.objects.filter(email = email).first() if not user: return Response({"status": False,"message": "Invalid email and password!", "data": []}) if not user.is_active: return Response({"status": False,"message": "Your account is disabled. Please contact support.", "data": []}) if user: orgs = Organization.objects.filter(organization_user = user).first() grps = user.groups.all() group_list = [] for grp in grps: group_list.append({'id':grp.id,'name':grp.name}) user_permissions = Permission.objects.filter(user=user) permission_data = {} for permission in user_permissions: content_type_model = permission.content_type.model if content_type_model not in permission_data: permission_data[content_type_model] = {} action_name = permission.codename.split("_")[0] permission_data[content_type_model][action_name] = permission.id formatted_data = [] for content_type, permissions in permission_data.items(): entry = {"name": content_type} entry.update(permissions) formatted_data.append(entry) email = user.email user_auth = authenticate(email = email, password = password) response_data = {'access':'','user_data':{},'user_permissions':[]} if user_auth: if user_auth.is_active == False: return Response({"status": False, "message": "Your account is disable, please contact to support.", "data": []}) else: refresh = RefreshToken.for_user(user_auth) serializer = CustomUserSerializer(user_auth) data = serializer.data response_data['access'] = str(refresh.access_token) response_data['user_data']['id'] = dict(data)['id'] response_data['user_data']['groups'] = group_list response_data['user_data']['profile_image'] = dict(data)['profile_image'] response_data['user_data']['last_login'] = dict(data)['last_login'] response_data['user_data']['is_superuser'] = dict(data)['is_superuser'] response_data['user_data']['email'] = dict(data)['email'] response_data['user_data']['first_name'] = dict(data)['first_name'] response_data['user_data']['last_name'] = dict(data)['last_name'] response_data['user_data']['is_active'] = dict(data)['is_active'] response_data['user_data']['is_staff'] = dict(data)['is_staff'] response_data['user_data']['phone_no'] = dict(data)['phone_no'] response_data['user_data']['created_at'] = dict(data)['created_at'] response_data['user_data']['org_type'] = orgs.org_type response_data['user_data']['org_id'] = orgs.id response_data['user_permissions'] = formatted_data # data['token'] = str(refresh.access_token) # data['org_type'] = orgs.org_type # data['refresh_token'] = str(refresh) return Response({"status": True,"message": "You are logged in!","data": response_data}) else: return Response({"status": False,"message": "Invalid username and password!","data":[]}) except Exception as e: return Response({"status": False,'message': "Something went wrong!",'error': str(e)}) class UserProfileView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CustomUserSerializer queryset = CustomUser.objects.all() lookup_field = 'user_id' def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user: return Response({'status': False,'message': 'You can not access location list','data':[]}) user_profile = CustomUser.objects.filter(email=user.email).first() if not user_profile: return Response({'status': False, 'message': 'User profile not found', 'data': []}) serializer = CustomUserSerializer(user_profile, context={'request': request}) return Response({"status": True, "message":"User profile get successfully", "data": serializer.data}) except Exception as e: return Response({"status": False,'message': "Something went wrong!",'error': str(e)}) def update(self, request, *args, **kwargs): try: user = self.request.user first_name = request.data.get('first_name') last_name = request.data.get('last_name') phone_no = request.data.get('phone_no') email = request.data.get('email') profile_image = request.data.get('profile_image','') if not user: return Response({'status': False,'message': 'You can not access location list','data':[]}) user_profile = CustomUser.objects.filter(email=user.email).first() if not user_profile: return Response({'status': False, 'message': 'User profile not found', 'data': []}) if profile_image: user_profile.profile_image = profile_image if first_name: user_profile.first_name = first_name if last_name: user_profile.last_name = last_name if phone_no: user_profile.phone_no = phone_no if email: user_profile.email = email user_profile.save() return Response({"status": True, "message":"Profile updated successfully."}) except Exception as e: return Response({"status": False,'message': "Something went wrong!",'error': str(e)}) class ChangePasswordAPIView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated,] def update(self, request, *args, **kwargs): try: # if not user.has_perm('myprofile.delete_customuser'): # return Response({'status': False, 'message': "You don't have permission to perform this action"}) user = request.user if not user: return Response({'status': False,'message': 'You can not Update User Password','data':[]}) old_password = request.data.get('old_password') new_password = request.data.get('new_password') if not old_password: return Response({'status': False,'message': 'Old Password is required','data':[]}) if not new_password: return Response({'status': False,'message': 'New Password is required','data':[]}) if old_password == new_password: return Response({'status': False,'message': 'Old and New Password is same please try again.','data':[]}) if not check_password(old_password, user.password): return Response({'status': False,'message': 'Invalid old password.','data':[]}) else: user.set_password(new_password) user.save() return Response({'status': True,'message': 'Your password changed successfully.'}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class OrganizationCreateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = OrganizationSerializer queryset = Organization.objects.all() lookup_field = 'org_id' pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user org_id = request.query_params.get('org_id') start_date = request.query_params.get('start_date','') end_date = request.query_params.get('end_date', '') search_query = request.query_params.get('search', '') if not user.has_perm('myprofile.view_organization'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) organization_obj = None # if user.groups.filter(name='Admin').exists(): organization_obj = queryset.filter(organization_user = user).order_by('-created_at') if not organization_obj: return Response({'status': False,'message': "No data found!","data":[]}) if search_query: search_query = search_query.strip().lower() organization_obj = organization_obj.filter(Q(org_name__icontains = search_query) | Q(org_address__icontains = search_query) | Q(org_phone__icontains = search_query) | Q(org_email__icontains = search_query) | Q(organization_user__first_name__icontains = search_query) | Q(organization_user__last_name__icontains = search_query)).order_by('-created_at') # if start_date or end_date: # start_date,end_date, error_message = validate_dates(start_date, end_date) # if error_message: # return Response({'status': False,'message': error_message,"data":[]}) # if user.groups.filter(name='Admin').exists(): # organization_obj = organization_obj.filter(created_at__range=[start_date, end_date]).order_by('-created_at') # else: # organization_obj = organization_obj.filter(created_at__range=[start_date, end_date], user = user).order_by('-created_at') if org_id: organization_obj = organization_obj.filter(id = org_id) if organization_obj.exists(): page = self.paginate_queryset(organization_obj) if page is not None: serializer = OrganizationSerializer(page,many=True) serializer = self.get_paginated_response(serializer.data) else: serializer = OrganizationSerializer(organization_obj,many=True) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Organization List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({"status": True,'message':'Organization Not Found',"data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def create(self, request): try: user = self.request.user user_ids = request.data.get('user_ids',[]) org_name = request.data.get('org_name') org_address = request.data.get('org_address') org_city = request.data.get('org_city') org_state = request.data.get('org_state') org_country = request.data.get('org_country') org_zipcode = request.data.get('org_zipcode') org_phone = request.data.get('org_phone') org_email = request.data.get('org_email') if not org_name or not org_address or not org_city or not org_state or not org_country or not org_zipcode or not org_phone or not org_email: return Response({'status': False,'message': 'All fields are required','data':[]}) if not user_ids: return Response({'status': False, 'message': 'User IDs are required.', 'data':[]}) if not CustomUser.objects.filter(id__in=user_ids).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) if org_name and Organization.objects.filter(org_name = org_name).exists(): return Response({'status': False,'message': f'{org_name} organization already exists.','data':[]}) if user.groups.filter(name='super_admin').exists(): created = Organization.objects.create( org_name = org_name, org_address = org_address, org_city = org_city, org_state = org_state, org_country = org_country, org_zipcode = org_zipcode, org_phone = org_phone, org_email = org_email) if not user.is_superuser==True: created.created_by = user users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) created.organization_user.set(users) created.organization_user.add(user) serializer = OrganizationSerializer(created) user_serializer = CustomUserSerializer(users,many=True,context={'request': request}) if created: return Response({'status': True,'message': f'Your Organization {org_name} created successfully.',"data":serializer.data}) else: return Response({'status': False,'message': f'Failed to create {org_name} organization.','data':[]}) else: return Response({'status': False,'message': 'Invalid user id You can not create Organization','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class OrganizationUpdateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = OrganizationSerializer queryset = Organization.objects.all() lookup_field = 'org_id' pagination_class = StandardResultsSetPagination def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_organization'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) org_id = self.kwargs.get('org_id') org_name = request.data.get('org_name') org_address = request.data.get('org_address') org_city = request.data.get('org_city') org_state = request.data.get('org_state') org_country = request.data.get('org_country') org_zipcode = request.data.get('org_zipcode') org_phone = request.data.get('org_phone') org_email = request.data.get('org_email') # notification_enable = request.data.get('email_notification_enable') user_ids = request.data.get('user_ids', []) #only Admin and Manager can update organization if user.groups.filter(name='super_admin').exists(): # if org_name and Organization.objects.filter(org_name = org_name).exists(): # return Response({'status': False,'message': f'{org_name} organization already exists.','data':[]}) update_obj = Organization.objects.get(id=org_id) update_obj.org_name = org_name update_obj.org_address = org_address update_obj.org_city = org_city update_obj.org_state = org_state update_obj.org_country = org_country update_obj.org_zipcode = org_zipcode update_obj.org_phone = org_phone update_obj.org_email = org_email if not user.is_superuser==True: update_obj.updated_by = user # update_obj.email_notification_is_enabled = notification_enable update_obj.save() # Remove users from linked location linked_location = Location.objects.filter(organization__id = org_id) for location in linked_location: # Only remove users from location if the user is not in the new list of users location_users_to_remove = location.location_user.exclude(id__in = user_ids).exclude(is_superuser = True).exclude(groups__name = 'super_admin') location.location_user.remove(*location_users_to_remove) # Remove users from linked area linked_area = Area.objects.filter(organization__id = org_id) for area in linked_area: # Only remove users from area if the user is not in the new list of users area_users_to_remove = area.area_user.exclude(id__in = user_ids).exclude(is_superuser = True).exclude(groups__name = 'super_admin') area.area_user.remove(*area_users_to_remove) # remove users from linked camera linked_camera = Camera.objects.filter(organization__id = org_id) for camera in linked_camera: # Only remove users from camera if the user is not in the new list of users camera_users_to_remove = camera.camera_user.exclude(id__in = user_ids).exclude(is_superuser = True).exclude(groups__name = 'super_admin') camera.camera_user.remove(*camera_users_to_remove) users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) update_obj.organization_user.set(users) update_obj.organization_user.add(user) serializer = OrganizationSerializer(update_obj) if update_obj: return Response({'status': True,'message': f'Your Organization {org_name} updated successfully.',"data":serializer.data},status = 200) else: return Response({'status': False,'message': f'Failed to update {org_name} organization.','data':[]}) else: return Response({'status': False,'message': 'You can not update Organization','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def destroy(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.delete_organization'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) org_id = self.kwargs.get('org_id') # if user.groups.filter(name='Admin').exists() or user.groups.filter(name='Manager').exists(): if org_id: org = Organization.objects.filter(id=org_id).delete() if org: return Response({'status': True,'message': f'Organization deleted successfully.'}) else: return Response({'status': False,'message': f'Failed to delete organization.','data':[]}) else: # return Response({'status': False,'message': 'You can not delete Organization','data':[]}) return Response({'status': False,'message': 'Organisation id is required.','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class OrganizationGetView(viewsets.ModelViewSet): # without pagination permission_classes = [IsAuthenticated] serializer_class = OrganizationSerializer queryset = Organization.objects.all() filter_backends = [filters.SearchFilter] lookup_field = 'org_id' # search_fields = ['org_name'] def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user # if user.groups.filter(name='Admin').exists(): queryset = Organization.objects.all().order_by('-created_at') # else: # queryset = Organization.objects.filter(organization_user = user).first() serializer = OrganizationSerializer(queryset,context={'request': request},many=True) return Response({"status": True, "message":"Organization List Successfully", "data": serializer.data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class EmailNotification(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = OrganizationSerializer queryset = Organization.objects.all() filter_backends = [filters.SearchFilter] lookup_field = 'org_id' # search_fields = ['org_name'] def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_organization'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) # if user.groups.filter(name='super_admin').exists(): # queryset = Organization.objects.all().order_by('-created_at') # else: queryset = Organization.objects.filter(organization_user = user).first() serializer = OrganizationSerializer(queryset,context={'request': request}) return Response({"status": True, "message":"Organization List Successfully", "data": serializer.data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_organization'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) org_id = self.kwargs.get('org_id') org_name = request.data.get('org_name') org_address = request.data.get('org_address') org_city = request.data.get('org_city') org_state = request.data.get('org_state') org_country = request.data.get('org_country') org_zipcode = request.data.get('org_zipcode') org_phone = request.data.get('org_phone') org_email = request.data.get('org_email') email_notification_is_enabled = request.data.get('email_notification_is_enabled') if user.groups.filter(name='super_admin').exists(): if org_id: email_org_obj = Organization.objects.filter(id=org_id).first() if email_org_obj: if org_name: email_org_obj.org_name = org_name if org_address: email_org_obj.org_address = org_address if org_city: email_org_obj.org_city = org_city if org_state: email_org_obj.org_state = org_state if org_country: email_org_obj.org_country = org_country if org_zipcode: email_org_obj.org_zipcode = org_zipcode if org_phone: email_org_obj.org_phone = org_phone if org_email: email_org_obj.org_email = org_email if email_notification_is_enabled: email_org_obj.email_notification_is_enabled = email_notification_is_enabled else: email_org_obj.email_notification_is_enabled = False email_org_obj.save() return Response({"status": True, "message":"Email notification updated successfully"}) else: return Response({'status': False,'message': 'Organization Not Found!!'}) else: return Response({'status': False,'message': 'Organization id is required!!'}) else: return Response({'status': False,'message': 'You can not enable or disable Email notification','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class LocationCreateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Location.objects.all() serializer_class = LocationSerializer pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_location'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) location_id = request.query_params.get('location_id') organization_id = request.query_params.get('organization_id') # start_date = request.query_params.get('start_date','') # end_date = request.query_params.get('end_date', '') search_query = request.query_params.get('search', '') location_obj = None # if start_date or end_date: # start_date,end_date, error_message = validate_dates(start_date, end_date) # if error_message: # return Response({'status': False,'message': error_message,"data":[]}) # if user.groups.filter(name='Admin').exists(): # location_obj = queryset.filter(created_at__range=[start_date, end_date]).order_by('-created_at') # else: # location_obj = queryset.filter(created_at__range=[start_date, end_date], location_user = user).order_by('-created_at') # else: # if user.groups.filter(name='Admin').exists(): # location_obj = queryset.all().order_by('-created_at') # else: # location_obj = queryset.filter(location_user = user).order_by('-created_at') # if user.groups.filter(name='Admin').exists(): location_obj = queryset.filter(organization__organization_user = user).order_by('-created_at') # elif user.groups.filter(name='Manager').exists(): # location_obj = queryset.filter(location_user = user).order_by('-created_at') if not location_obj: return Response({'status': False,'message': "No data found!","data":[]}) if search_query: search_query = search_query.strip().lower() location_obj = location_obj.filter(Q(loc_name__icontains = search_query) | Q(loc_address__icontains = search_query) | Q(organization__org_name__icontains = search_query) | Q(created_at__icontains = search_query)).order_by('-created_at') if location_id: location_obj = location_obj.filter(id = location_id) if organization_id: location_obj = location_obj.filter(organization__id = int(organization_id)) if location_obj.exists(): page = self.paginate_queryset(location_obj) if page is not None: serializer = LocationSerializer(page,many=True) serializer = self.get_paginated_response(serializer.data) else: serializer = LocationSerializer(location_obj,many=True) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Location List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({"status": True,'message':'Location Not Found',"data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def create(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.add_location'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) org_id = request.data.get('org_id') loc_name = request.data.get('loc_name') loc_address = request.data.get('loc_address') longitude = request.data.get('longitude') latitude = request.data.get('latitude') user_ids = request.data.get('user_ids', []) # Longitude and Latitude not in float then return error if longitude and not isinstance(longitude, float): return Response({'status': False,'message': 'Longitude should be in float','data':[]}) if latitude and not isinstance(latitude, float): return Response({'status': False,'message': 'Latitude should be in float','data':[]}) if not org_id or not loc_name: return Response({'status': False,'message': 'All fields are required','data':[]}) # if not Organization.objects.filter(id = org_id,organization_user = user).exists(): # return Response({'status': False,'message': 'Invalid organization or user id','data':[]}) # if location name already exists in specified organization then return error # if loc_name and Location.objects.filter(loc_name = loc_name,organization_id = org_id).exists(): # return Response({'status': False,'message': 'location already exists in this organization.','data':[]}) if loc_name and Location.objects.filter(loc_name = loc_name,organization__id = org_id).exists(): return Response({'status': False,'message': 'location already exists in this organization.','data':[]}) if not CustomUser.objects.filter(id__in=user_ids).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # Check if the organization exists organization = Organization.objects.filter(id=org_id).first() if not organization: return Response({'status': False, 'message': 'Organization not found.', 'data': []}) for user_id in user_ids: if not organization.organization_user.filter(id=user_id).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) longitude = Decimal(longitude) latitude = Decimal(latitude) location = Location.objects.create( loc_name = loc_name, loc_address = loc_address, longitude = longitude, latitude = latitude, organization = organization ) if not user.is_superuser==True: location.created_by = user users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) location.location_user.set(users) location.location_user.add(user) serializer = LocationSerializer(location) # user_serializer = CustomUserSerializer(users,many=True,context={'request': request}) return Response({ 'status': True, 'message': f'Your Location {loc_name} created successfully.', "data":serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class LocationUpdateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Location.objects.all() serializer_class = LocationSerializer lookup_field = 'loc_id' pagination_class = StandardResultsSetPagination filter_backends = [filters.SearchFilter] search_fields = ['loc_name'] def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_location'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) loc_id = self.kwargs.get('loc_id') org_id = request.data.get('org_id') loc_name = request.data.get('loc_name') loc_address = request.data.get('loc_address') longitude = request.data.get('longitude') latitude = request.data.get('latitude') user_ids = request.data.get('user_ids', []) if not Location.objects.filter(id = loc_id).exists(): return Response({'status': False,'message': 'location id not Found','data':[]}) if not org_id or not loc_name: return Response({'status': False,'message': 'All fields are required','data':[]}) # if not Organization.objects.filter(id = org_id,organization_user = user).exists(): # return Response({'status': False,'message': 'Invalid organization or user id','data':[]}) # if location name already exists in specified organization then return error # if loc_name and Location.objects.filter(loc_name = loc_name,organization_id = org_id).exists(): # return Response({'status': False,'message': 'location already exists in this organization.','data':[]}) if not CustomUser.objects.filter(id__in=user_ids).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # Check if the organization exists organization = Organization.objects.filter(id=org_id).first() # if not organization: # return Response({'status': False, 'message': 'Organization not found.', 'data': []}) if not organization: return Response({'status': False, 'message': 'Organization not found', 'data':[]}) for user_id in user_ids: if not organization.organization_user.filter(id=user_id).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) location = Location.objects.get(id=loc_id) location.loc_name = loc_name location.loc_address = loc_address location.longitude = longitude location.latitude = latitude location.organization = organization if not user.is_superuser==True: location.updated_by = user location.save() # Remove users from linked area linked_area = Area.objects.filter(location__id = loc_id) for area in linked_area: # Only remove users from area if the user is not in the new list of users area_users_to_remove = area.area_user.exclude(id__in = user_ids).exclude(is_superuser = True).exclude(groups__name = 'super_admin') area.area_user.remove(*area_users_to_remove) # remove users from linked camera linked_camera = Camera.objects.filter(location__id = loc_id) for camera in linked_camera: # Only remove users from camera if the user is not in the new list of users camera_users_to_remove = camera.camera_user.exclude(id__in = user_ids).exclude(is_superuser = True).exclude(groups__name = 'super_admin') camera.camera_user.remove(*camera_users_to_remove) users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) location.location_user.set(users) location.location_user.add(user) serializer = LocationSerializer(location) user_serializer = CustomUserSerializer(users,many=True,context={'request': request}) return Response({ 'status': True, 'message': f'Your Location {loc_name} updated successfully.', "data":serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def destroy(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.delete_location'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) loc_id = self.kwargs.get('loc_id') # if not loc_id: # return Response({'status': False,'message': 'Location id is required','data':[]}) # if not Location.objects.filter(id = loc_id).exists(): # return Response({'status': False,'message': 'Location id not Found','data':[]}) # if not Organization.objects.filter(organization_user = user).exists(): # return Response({'status': False,'message': 'You have no Authority to delete Location','data':[]}) # if user.groups.filter(name='Admin').exists(): if loc_id: loc = Location.objects.filter(id=loc_id).delete() if loc: return Response({'status': True,'message': 'Location deleted successfully.,','data':[]}) else: return Response({'status': False,'message': 'You have no Authority to delete Location','data':[]}) else: return Response({'status': False,'message': 'Location id is required.'}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class LocationGetView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Location.objects.all() serializer_class = LocationSerializer filter_backends = [filters.SearchFilter] search_fields = ['loc_name'] def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user organization_id = request.query_params.get('organization_id') if not user: return Response({'status': False,'message': 'You can not access location list','data':[]}) if not organization_id: return Response({'status': False,'message': 'Organization id is required','data':[]}) if not Organization.objects.filter(id = organization_id).exists(): return Response({'status': False,'message': 'Invalid organization id','data':[]}) # if user.groups.filter(name='Admin').exists(): queryset = Location.objects.filter(organization__id = organization_id).order_by('-created_at') # else: # queryset = Location.objects.filter(organization__id = organization_id, ).order_by('-created_at') # filter project by organization id location = queryset.filter(organization__id = organization_id) if not location: return Response({'status': False,'message': 'Location not found','data':[]}) serializer = LocationSerializer(location,many=True) return Response({'status': True,'message': 'Location List Successfully','data':serializer.data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class AreaCreateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Area.objects.all() serializer_class = AreaSerializer lookup_field = 'area_id' pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_area'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) area_id = request.query_params.get('area_id') location_id = request.query_params.get('location_id') organization_id = request.query_params.get('organization_id') # start_date = request.query_params.get('start_date','') # end_date = request.query_params.get('end_date', '') search_query = request.query_params.get('search', '') # if start_date or end_date: # start_date,end_date, error_message = validate_dates(start_date, end_date) # if error_message: # return Response({'status': False,'message': error_message,"data":[]}) # if user.groups.filter(name='Admin').exists(): # area_obj = Area.objects.filter(created_at__range=[start_date, end_date]).order_by('-created_at') # else: # area_obj = Area.objects.filter(created_at__range=[start_date, end_date], area_user = user).order_by('-created_at') # else: # if user.groups.filter(name='Admin').exists(): # area_obj = Area.objects.all().order_by('-created_at') # else: # area_obj = Area.objects.filter(area_user = user).order_by('-created_at') area_obj = None # if user.groups.filter(name='Admin').exists(): if queryset: area_obj = queryset.filter(organization__organization_user__email = user).order_by('-created_at') # elif user.groups.filter(name='Manager').exists(): # area_obj = queryset.filter(location__location_user = user).order_by('-created_at') if not area_obj: return Response({'status': False,'message': "No data found!","data":[]}) if search_query: search_query = search_query.strip().lower() area_obj = area_obj.filter(Q(area_name__icontains = search_query) | Q(area_address__icontains = search_query) | Q(organization__org_name__icontains = search_query) | Q(created_at__icontains = search_query) | Q(location__loc_name__icontains = search_query) | Q(location__loc_address__icontains = search_query) ).order_by('-created_at') if organization_id: area_obj = area_obj.filter(organization__id = organization_id) if location_id: area_obj = area_obj.filter(location__id = location_id) if area_id: area_obj = area_obj.filter(id = area_id) if area_obj.exists(): page = self.paginate_queryset(area_obj) if page is not None: serializer = AreaSerializer(page,many=True) serializer = self.get_paginated_response(serializer.data) else: serializer = AreaSerializer(area_obj,many=True) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Area List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({"status": True,'message':'Area Not Found',"data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def create(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.add_area'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) org_id = request.data.get('org_id') loc_id = request.data.get('loc_id') area_name = request.data.get('area_name') area_address = request.data.get('area_address') user_ids = request.data.get('user_ids', []) if not org_id or not loc_id or not area_name: return Response({'status': False,'message': 'All fields are required','data':[]}) # if not Organization.objects.filter(id = org_id,organization_user = user).exists(): # return Response({'status': False,'message': 'Invalid organization or user id','data':[]}) # if not Location.objects.filter(id = loc_id,organization_id = org_id).exists(): # return Response({'status': False,'message': 'Invalid location id','data':[]}) # if area name already exists in specified location and organization then return error if area_name and Area.objects.filter(area_name = area_name,location__id = loc_id,organization__id = org_id).exists(): return Response({'status': False,'message': 'area already exists in this location.','data':[]}) if not CustomUser.objects.filter(id__in=user_ids).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # Check if the organization exists organization = Organization.objects.filter(id=org_id).first() if not organization: return Response({'status': False, 'message': 'Organization not found.', 'data': []}) # Check if the location exists location = Location.objects.filter(id=loc_id).first() if not location: return Response({'status': False, 'message': 'Location not found.', 'data': []}) for user_id in user_ids: if not organization.organization_user.filter(id=user_id).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) area = Area.objects.create( organization = organization, location = location, area_name = area_name, area_address = area_address, ) if not user.is_superuser==True: area.created_by = user users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) area.area_user.set(users) area.area_user.add(user) serializer = AreaSerializer(area) user_serializer = CustomUserSerializer(users,many=True,context={'request': request}) return Response({ 'status': True, 'message': f'Your Area {area_name} created successfully.', "data":serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class AreaUpdateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Area.objects.all() serializer_class = AreaSerializer lookup_field = 'area_id' pagination_class = StandardResultsSetPagination def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_area'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) area_id = self.kwargs.get('area_id') org_id = request.data.get('org_id') loc_id = request.data.get('loc_id') area_name = request.data.get('area_name') area_address = request.data.get('area_address') user_ids = request.data.get('user_ids', []) if not Area.objects.filter(id = area_id).exists(): return Response({'status': False,'message': 'Area id not Found','data':[]}) if not org_id or not loc_id: return Response({'status': False,'message': 'All fields are required','data':[]}) # if not Organization.objects.filter(id = org_id,organization_user = user).exists(): # return Response({'status': False,'message': 'Invalid organization or user id','data':[]}) # if not Location.objects.filter(id = loc_id,organization_id = org_id).exists(): # return Response({'status': False,'message': 'Invalid location id','data':[]}) # if area name already exists in specified location and organization then return error # if area_name and Area.objects.filter(area_name = area_name,location_id = loc_id,organization_id = org_id).exists(): # return Response({'status': False,'message': 'Area Name already exists in this location.','data':[]}) if not CustomUser.objects.filter(id__in=user_ids).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # Check if the organization exists organization = Organization.objects.filter(id=org_id).first() if not organization: return Response({'status': False, 'message': 'Organization not found.', 'data': []}) # Check if the location exists location = Location.objects.filter(id=loc_id).first() if not location: return Response({'status': False, 'message': 'Location not found.', 'data': []}) for user_id in user_ids: if not organization.organization_user.filter(id=user_id).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) area = Area.objects.get(id=area_id) area.organization = organization area.location = location area.area_name = area_name area.area_address = area_address if not user.is_superuser==True: area.updated_by = user area.save() # Remove users from linked camera linked_camera = Camera.objects.filter(area__id = area_id) for camera in linked_camera: # Only remove users from camera if the user is not in the new list of users camera_users_to_remove = camera.camera_user.exclude(id__in = user_ids).exclude(is_superuser = True).exclude(groups__name = 'super_admin') camera.camera_user.remove(*camera_users_to_remove) users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) area.area_user.set(users) area.area_user.add(user) serializer = AreaSerializer(area) user_serializer = CustomUserSerializer(users,many=True,context={'request': request}) return Response({ 'status': True, 'message': f'Your Area {area_name} updated successfully.', "data":serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def destroy(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.delete_area'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) area_id = self.kwargs.get('area_id') # if not area_id: # return Response({'status': False,'message': 'Area id is required','data':[]}) # if not Area.objects.filter(id = area_id).exists(): # return Response({'status': False,'message': 'Area id not Found','data':[]}) # if not Organization.objects.filter(organization_user = user).exists(): # return Response({'status': False,'message': 'You have no Authority to delete Area','data':[]}) # if user.groups.filter(name='Admin').exists(): if area_id: area = Area.objects.filter(id=area_id).delete() if area: return Response({'status': True,'message': 'Area deleted successfully.,','data':[]}) else: return Response({'status': False,'message': 'You have no Authority to delete Area','data':[]}) else: return Response({'status': False,'message': 'Area id is required.'}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class AreaGetView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Area.objects.all() serializer_class = AreaSerializer pagination_class = StandardResultsSetPagination filter_backends = [filters.SearchFilter] search_fields = ['area_name'] def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user organization_id = request.query_params.get('organization_id') location_id = request.query_params.get('location_id') if not user: return Response({'status': False,'message': 'You can not access area list','data':[]}) # if not organization_id: # return Response({'status': False,'message': 'Organization id is required','data':[]}) # if not location_id: # return Response({'status': False,'message': 'Location id is required','data':[]}) # Filter area by Organisation id if organization_id: queryset = queryset.filter(organization__id = organization_id).order_by('-created_at') # Filter area by location id if location_id: queryset = queryset.filter(location__id = location_id).order_by('-created_at') if queryset: serializer = AreaSerializer(queryset,many=True) return Response({'status': True,'message': 'Area List Successfully','data':serializer.data}) else: return Response({'status': False,'message': 'No Area Found.','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class CameraCreateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Camera.objects.all() serializer_class = CameraSerializer lookup_field = 'camera_id' pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_camera'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) camera_name = request.query_params.get('camera_name','') area_id = request.query_params.get('area_id','') location_id = request.query_params.get('location_id','') organization_id = request.query_params.get('organization_id','') # start_date = request.query_params.get('start_date','') # end_date = request.query_params.get('end_date', '') search_query = request.query_params.get('search', '') # if start_date or end_date: # start_date,end_date, error_message = validate_dates(start_date, end_date) # if error_message: # return Response({'status': False,'message': error_message,"data":[]}) # if user.groups.filter(name='Admin').exists(): # camera_obj = Camera.objects.filter(created_at__range=[start_date, end_date]).order_by('-created_at') # else: # camera_obj = Camera.objects.filter(created_at__range=[start_date, end_date], camera_user = user).order_by('-created_at') # else: # if user.groups.filter(name='Admin').exists(): # camera_obj = Camera.objects.all().order_by('-created_at') # else: # camera_obj = Camera.objects.filter(camera_user = user).order_by('-created_at') camera_obj = None # if user.groups.filter(name='Admin').exists(): if queryset: camera_obj = queryset.filter(organization__organization_user = user).order_by('-created_at') # elif user.groups.filter(name='Manager').exists(): # camera_obj = queryset.filter(location__location_user = user).order_by('-created_at') if not camera_obj: return Response({'status': False,'message': "No data found!","data":[]}) if search_query: search_query = search_query.strip().lower() camera_obj = camera_obj.filter(Q(camera_name__icontains = search_query) | Q(created_at__icontains = search_query) | Q(rtsp_url__icontains = search_query) | Q(location__loc_name__icontains = search_query) | Q(area__area_name__icontains = search_query) ).order_by('-created_at') if organization_id: camera_obj = camera_obj.filter(organization__id = organization_id) if location_id: camera_obj = camera_obj.filter(location__id = location_id) if area_id: camera_obj = camera_obj.filter(area__id = area_id) if camera_name: camera_obj = camera_obj.filter(camera_name = camera_name) if camera_obj.exists(): page = self.paginate_queryset(camera_obj) if page is not None: serializer = CameraSerializer(page,many=True) serializer = self.get_paginated_response(serializer.data) else: serializer = CameraSerializer(camera_obj,many=True) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Camera List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({"status": True,'message':'Camera Not Found',"data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def create(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.add_camera'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) organizaion_id = request.data.get('organization_id') location_id = request.data.get('location_id') area_id = request.data.get('area_id') camera_name = request.data.get('camera_name') fps = int(request.data.get('fps',30)) ip_address = request.data.get('ip_address') rtsp_url = request.data.get('rtsp_url') frame_numbers = request.data.get('frame_numbers') port = request.data.get('port',) camera_events_ids = request.data.get('camera_events_ids', []) user_ids = request.data.get('user_ids', []) if not organizaion_id or not location_id or not area_id or not camera_name or not rtsp_url: return Response({'status': False,'message': 'All fields are required','data':[]}) if Camera.objects.filter(rtsp_url = rtsp_url).exists(): return Response({'status': False,'message': 'Duplicate RTSP not allowed.','data':[]}) # if not Organization.objects.filter(id = organizaion_id).exists(): # return Response({'status': False,'message': 'Invalid organization','data':[]}) # if not Location.objects.filter(id = location_id,organization_id = organizaion_id).exists(): # return Response({'status': False,'message': 'Invalid location id','data':[]}) # if not Area.objects.filter(id = area_id,location_id = location_id,organization_id = organizaion_id).exists(): # return Response({'status': False,'message': 'Invalid area id','data':[]}) # if camera name already exists in specified area, location and organization then return error if camera_name and Camera.objects.filter(camera_name = camera_name,area__id = area_id,location__id = location_id,organization__id = organizaion_id).exists(): return Response({'status': False,'message': 'Camera already exists in this area.','data':[]}) if not CustomUser.objects.filter(id__in=user_ids).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # Check if the organization exists organization = Organization.objects.filter(id=organizaion_id).first() if not organization: return Response({'status': False, 'message': 'Organization not found.', 'data': []}) # Check if the location exists location = Location.objects.filter(id=location_id).first() if not location: return Response({'status': False, 'message': 'Location not found.', 'data': []}) # Check if the area exists area = Area.objects.filter(id=area_id).first() if not area: return Response({'status': False, 'message': 'Area not found.', 'data': []}) for user_id in user_ids: if not organization.organization_user.filter(id=user_id).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) camera = Camera.objects.create( organization = organization, location = location, area = area, camera_name = camera_name, fps = fps, ip_address = ip_address, rtsp_url = rtsp_url, frame_numbers = frame_numbers, port = port ) if not user.is_superuser==True: camera.created_by = user users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) camera.camera_user.set(users) camera.camera_user.add(user) if camera_events_ids: for event_id in camera_events_ids: if not CameraEvent.objects.filter(id = event_id).exists(): return Response({'status': False, 'message': 'Invalid Event IDs.', 'data':[]}) link_camera_with_events = LinkCameraEventToCamera.objects.create( camera = camera ) camera_events_objs = CameraEvent.objects.filter(id__in = camera_events_ids) link_camera_with_events.camera_events.set(camera_events_objs) serializer = CameraSerializer(camera) user_serializer = CustomUserSerializer(users,many=True,context={'request': request}) return Response({ 'status': True, 'message': f'Your Camera {camera_name} created successfully.', "data":serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class CameraUpdataeView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Camera.objects.all() serializer_class = CameraSerializer lookup_field = 'camera_id' pagination_class = StandardResultsSetPagination def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_camera'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) camera_id = self.kwargs.get('camera_id') organizaion_id = request.data.get('organization_id') location_id = request.data.get('location_id') area_id = request.data.get('area_id') camera_name = request.data.get('camera_name') fps = int(request.data.get('fps',30)) ip_address = request.data.get('ip_address') rtsp_url = request.data.get('rtsp_url') frame_numbers = request.data.get('frame_numbers') port = request.data.get('port') user_ids = request.data.get('user_ids', []) if not Camera.objects.filter(id = camera_id).exists(): return Response({'status': False,'message': 'Camera id not Found','data':[]}) if not organizaion_id or not location_id or not area_id or not camera_name or not rtsp_url: return Response({'status': False,'message': 'All fields are required','data':[]}) # if not Organization.objects.filter(id = organizaion_id,organization_user = user).exists(): # return Response({'status': False,'message': 'Invalid organization or user id','data':[]}) # if not Location.objects.filter(id = location_id,organization_id = organizaion_id).exists(): # return Response({'status': False,'message': 'Invalid location id','data':[]}) # if not Area.objects.filter(id = area_id,location_id = location_id,organization_id = organizaion_id).exists(): # return Response({'status': False,'message': 'Invalid area id','data':[]}) # if camera name already exists in specified area, location and organization then return error # if camera_name and Camera.objects.filter(camera_name = camera_name,area_id = area_id,location_id = location_id,organization_id = organizaion_id).exists(): # return Response({'status': False,'message': 'Camera already exists in this area.','data':[]}) if not CustomUser.objects.filter(id__in=user_ids).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # Check if the organization exists organization = Organization.objects.filter(id=organizaion_id).first() if not organization: return Response({'status': False, 'message': 'Organization not found.', 'data': []}) # Check if the location exists location = Location.objects.filter(id=location_id).first() if not location: return Response({'status': False, 'message': 'Location not found.', 'data': []}) # Check if the area exists area = Area.objects.filter(id=area_id).first() if not area: return Response({'status': False, 'message': 'Area not found.', 'data': []}) for user_id in user_ids: if not organization.organization_user.filter(id=user_id).exists(): return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) camera = Camera.objects.get(id=camera_id) camera.organization = organization camera.location = location camera.area = area camera.camera_name = camera_name camera.fps = fps camera.ip_address = ip_address camera.rtsp_url = rtsp_url camera.frame_numbers = frame_numbers camera.port = port if not user.is_superuser==True: camera.updated_by = user camera.save() users = CustomUser.objects.filter(Q(id__in = user_ids)| Q(groups__name='super_admin')| Q(is_superuser=True)) camera.camera_user.set(users) camera.camera_user.add(user) serializer = CameraSerializer(camera) user_serializer = CustomUserSerializer(users,many=True,context={'request': request}) return Response({ 'status': True, 'message': f'Your Camera {camera_name} updated successfully.', "data":serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def destroy(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.delete_camera'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) camera_id = self.kwargs.get('camera_id') # if not camera_id: # return Response({'status': False,'message': 'Camera id is required','data':[]}) # if not Camera.objects.filter(id = camera_id).exists(): # return Response({'status': False,'message': 'Camera id not Found','data':[]}) # if not Organization.objects.filter(organization_user = user).exists(): # return Response({'status': False,'message': 'You have no Authority to delete Camera','data':[]}) # if user.groups.filter(name='Admin').exists(): if camera_id: camera = Camera.objects.filter(id=camera_id).delete() if camera: return Response({'status': True,'message': 'Camera deleted successfully.,','data':[]}) else: return Response({'status': False,'message': 'You have no Authority to delete Camera','data':[]}) else: return Response({'status': False,'message': 'Camera id is required.','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class CameraGetView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Camera.objects.all() serializer_class = CameraSerializer pagination_class = StandardResultsSetPagination filter_backends = [filters.SearchFilter] search_fields = ['camera_name'] def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user organization_id = request.query_params.get('organization_id') location_id = request.query_params.get('location_id') area_id = request.query_params.get('area_id') if not user: return Response({'status': False,'message': 'You can not access camera list','data':[]}) if not organization_id: return Response({'status': False,'message': 'Organization id is required','data':[]}) if not Organization.objects.filter(id = organization_id).exists(): return Response({'status': False,'message': 'Invalid organization id','data':[]}) if not location_id: return Response({'status': False,'message': 'Location id is required','data':[]}) if not Location.objects.filter(id = location_id,organization__id = organization_id).exists(): return Response({'status': False,'message': 'Invalid location id','data':[]}) if not area_id: return Response({'status': False,'message': 'Area id is required','data':[]}) if not Area.objects.filter(id = area_id,location__id = location_id,organization__id = organization_id).exists(): return Response({'status': False,'message': 'Invalid area id','data':[]}) # if user.groups.filter(name='Admin').exists(): queryset = queryset.filter(organization__id = organization_id).order_by('-created_at') # else: # queryset = queryset.filter(organization__id = organization_id, camera_user = user).order_by('-created_at') # filter project by organization id camera = queryset.filter(organization__id = organization_id) if location_id: camera = camera.filter(location__id = location_id) if area_id: camera = camera.filter(area__id = area_id) if not camera: return Response({'status': False,'message': 'Camera not found','data':[]}) serializer = CameraSerializer(camera,many=True) return Response({ 'status': True, 'message': 'Camera List Successfully', 'data':serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class CameraListWithoutPaginationView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Camera.objects.all() serializer_class = CameraListWithoutPaginationSerializer def list(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.view_camera'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) # if user.groups.filter(name='Admin').exists(): queryset = Camera.objects.all().order_by('-created_at') # else: # queryset = Camera.objects.filter(camera_user = user).order_by('-created_at') serializer = self.serializer_class(queryset, many=True) return Response({ 'status': True, 'message': 'Camera List Successfully', 'data':serializer.data }) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class CameraEnableDisableView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Camera.objects.all() serializer_class = CameraSerializer lookup_field = 'camera_id' pagination_class = StandardResultsSetPagination def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_camera'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) camera_id = self.kwargs.get('camera_id') try: camera_id = Camera.objects.get(id = camera_id) except: return Response({'status': False,'message': 'Camera id not Found','data':[]}) if camera_id.is_enabled: camera_id.is_enabled = False camera_id.save() return Response({'status': True,'message': 'Camera Disabled Successfully','data':[]}) else: camera_id.is_enabled = True camera_id.save() return Response({'status': True,'message': 'Camera Enabled Successfully','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class EventGetListView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = CameraEvent.objects.all() serializer_class = CameraEventSerializer pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user location_id = request.query_params.get('location_id') area_id = request.query_params.get('area_id') camera_id = self.kwargs.get('camera_id') # if user.groups.filter(name='Admin').exists(): camera_event_obj = queryset.filter(organization__organization_user = user) # elif user.groups.filter(name='Manager').exists(): # else: # camera_event_obj = queryset.filter(organization__organization_user__email = user, location__location_user__email = user) if not camera_event_obj: return Response({'status': False, 'message': 'No data found!', 'data': []}) if location_id: camera_event_obj = camera_event_obj.filter(location__id = location_id) if area_id: camera_event_obj = camera_event_obj.filter(area__id = area_id) if camera_id: camera_event_obj = camera_event_obj.filter(camera__id = camera_id) event_list = camera_event_obj.filter(is_enabled=True).order_by('-created_at') serializer = CameraEventSerializer(event_list, many=True) # event_list = list(set(event for event in event_list if event is not None)) return Response({'status': True, 'message': 'Camera Event List', 'data': serializer.data}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) from django.db import transaction class CameraEventCreateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CameraEventSerializer queryset = CameraEvent.objects.all() lookup_field = 'camera_event_id' pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_cameraevent'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) location_id = request.query_params.get('location_id') area_id = request.query_params.get('area_id') camera_id = request.query_params.get('camera_id') camera_event_id = request.query_params.get('camera_event_id') search_query = request.query_params.get('search', '') camera_event_obj = None # if user.groups.filter(name='Admin').exists(): camera_event_obj = queryset.filter(organization__organization_user__email = user).order_by('-created_at') # elif user.groups.filter(name='Manager').exists(): # camera_event_obj = queryset.filter(location__location_user = user).order_by('-created_at') if not camera_event_obj: return Response({'status': False,'message': "No data found!","data":[]}) if search_query: search_query = search_query.strip().lower() camera_event_obj = camera_event_obj.filter(Q(camera_event__icontains = search_query) | Q(created_at__icontains = search_query) | Q(camera__camera_name__icontains = search_query) | Q(location__loc_name__icontains = search_query) | Q(area__area_name__icontains = search_query) ).order_by('-created_at') if location_id: camera_event_obj = camera_event_obj.filter(camera__area__location__id = location_id) if area_id: camera_event_obj = camera_event_obj.filter(camera__area__id = area_id) if camera_id: camera_event_obj = camera_event_obj.filter(camera__id = camera_id) if camera_event_id: camera_event_obj = camera_event_obj.filter(id = camera_event_id) if camera_event_obj.exists(): page = self.paginate_queryset(camera_event_obj) if page is not None: serializer = CameraEventSerializer(page,many=True, context={'request': request}) serializer = self.get_paginated_response(serializer.data) else: serializer = CameraEventSerializer(camera_event_obj,many=True, context={'request': request}) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Camera Event List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({"status": True,'message':'Camera Event Not Found',"data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) # def create(self, request, *args, **kwargs): # try: # user = self.request.user # organization_id = request.data.get('organization_id') # camera_id = request.data.get('camera_id') # location_id = request.data.get('location_id') # area_id = request.data.get('area_id') # event_name = request.data.get('event_name') # if not organization_id or not camera_id or not location_id or not area_id or not event_name: # return Response({'status': False,'message': 'All fields are required','data':[]}) # if not Organization.objects.filter(id = organization_id).exists(): # return Response({'status': False,'message': 'Invalid organization','data':[]}) # if not Camera.objects.filter(id = camera_id,organization_id = organization_id).exists(): # return Response({'status': False,'message': 'Invalid camera id','data':[]}) # if not Location.objects.filter(id = location_id,organization_id = organization_id).exists(): # return Response({'status': False,'message': 'Invalid location id','data':[]}) # if not Area.objects.filter(id = area_id,location_id = location_id,organization_id = organization_id).exists(): # return Response({'status': False,'message': 'Invalid area id','data':[]}) # if CameraEvent.objects.filter(camera_event = event_name,organization_id = organization_id,camera_id = camera_id,location_id = location_id,area_id = area_id).exists(): # return Response({'status': False,'message': 'Event Name already exists in this area.','data':[]}) # created = CameraEvent.objects.create( # organization_id = organization_id, # camera_id = camera_id, # location_id = location_id, # area_id = area_id, # camera_event = event_name # ) # serializer = CameraEventSerializer(created) # return Response({'status': True,'message': 'Camera Event Created Successfully','data':serializer.data}) # except Exception as e: # return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def create(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.add_cameraevent'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) organization_id = request.data.get('organization_id') camera_id = request.data.get('camera_id') location_id = request.data.get('location_id') area_id = request.data.get('area_id') event_name = request.data.get('event_name') # Access the list directly # if not organization_id or not camera_id or not location_id or not area_id or not event_names: # return Response({'status': False, 'message': 'All fields are required', 'data': []}) if not Organization.objects.filter(id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid organization', 'data': []}) if not Camera.objects.filter(id=camera_id, organization_id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid camera id', 'data': []}) if not Location.objects.filter(id=location_id, organization_id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid location id', 'data': []}) if not Area.objects.filter(id=area_id, location_id=location_id, organization_id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid area id', 'data': []}) # Check if camera event already exists for the given camera ID if CameraEvent.objects.filter(camera_id=camera_id).exists(): return Response({'status': False, 'message': 'Camera Event already exists for this camera', 'data': []}) with transaction.atomic(): # Create a single camera_event entry with multiple event names camera_event = CameraEvent.objects.create( organization_id=organization_id, camera_id=camera_id, location_id=location_id, area_id=area_id, camera_event= event_name # Join event_names into a single string ) if not user.is_superuser==True: camera_event.created_by = user serializer = CameraEventSerializer(camera_event) return Response({'status': True, 'message': 'Camera Events Created Successfully', 'data': serializer.data}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) class CameraEventGetView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CameraEventSerializer queryset = CameraEvent.objects.all() lookup_field = 'camera_event_id' pagination_class = StandardResultsSetPagination def update(self, request, *args, **kwargs): user = request.user camera_event_id = self.kwargs['camera_event_id'] if not user.has_perm('myprofile.change_cameraevent'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) try: camera_event = CameraEvent.objects.get(id=camera_event_id) except CameraEvent.DoesNotExist: return Response({'status': False, 'message': 'Camera Event not found.', 'data': []}) if camera_event.is_enabled: camera_event.is_enabled = False camera_event.save() return Response({'status': True, 'message': f'Camera Disabled.'}) else: camera_event.is_enabled = True camera_event.save() return Response({'status': True, 'message': f'Camera Enabled.'}) def destroy(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.delete_cameraevent'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) camera_event_id = self.kwargs['camera_event_id'] # if not user.groups.filter(name='Admin').exists(): # return Response({'status': False, 'message': 'You have no Authority to delete Camera Event', 'data': []}) try: camera_event = CameraEvent.objects.get(id=camera_event_id) except CameraEvent.DoesNotExist: return Response({'status': False, 'message': 'Camera Event not found.', 'data': []}) camera_event.delete() return Response({'status': True, 'message': f'Camera Event deleted successfully.'}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) from django.shortcuts import get_object_or_404 class CameraEventnameUpdateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CameraEventSerializer queryset = CameraEvent.objects.all() lookup_field = 'camera_event_id' def update(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.change_cameraevent'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) camera_event_id = self.kwargs['camera_event_id'] organization_id = request.data.get('organization_id') camera_id = request.data.get('camera_id') location_id = request.data.get('location_id') area_id = request.data.get('area_id') event_name = request.data.get('event_name') if not user: return Response({'status': False, 'message': 'You do not have permission to update Camera Event', 'data': []}) if not organization_id or not camera_id or not location_id or not area_id or not event_names: return Response({'status': False, 'message': 'All fields are required', 'data': []}) # if not user.groups.filter(name='Admin').exists(): # return Response({'status': False, 'message': 'You have no Authority to update Camera Event', 'data': []}) if not Organization.objects.filter(id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid organization', 'data': []}) if not Camera.objects.filter(id=camera_id, organization_id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid camera id', 'data': []}) if not Location.objects.filter(id=location_id, organization_id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid location id', 'data': []}) if not Area.objects.filter(id=area_id, location_id=location_id, organization_id=organization_id).exists(): return Response({'status': False, 'message': 'Invalid area id', 'data': []}) camera_event_instance = CameraEvent.objects.get(id=camera_event_id) if not camera_event_instance: return Response({'status': False, 'message': 'Camera Event not found.', 'data': []}) camera_event_instance.organization_id = organization_id camera_event_instance.camera_id = camera_id camera_event_instance.location_id = location_id camera_event_instance.area_id = area_id camera_event_instance.camera_event = event_name # Join event_names into a single string if not user.is_superuser==True: camera_event_instance.updated_by = user camera_event_instance.save() serializer = CameraEventSerializer(camera_event_instance) return Response({'status': True, 'message': 'Camera Event updated successfully.', 'data': serializer.data}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) class WhatsappMessageView(viewsets.ModelViewSet): def create(self, request, *args, **kwargs): access_token = 'EAAXuFyLeNOkBO68lPeKw8JZAQnZBcCZA7bFO6sJEu0vrHYHgG2g2XBwSd236kCBoAHc4RaUphJnjSRPjesod7MSlJBcZAOAuv2LK2FWwNoflB9Pt28fBHGaNSIZBZCpDZBfWUW0s2gBRYBzOP6sO2fkb0VOg0rb8h1PZCLoF9nTHd2I54Wja2jm7SkB8qiVsk2M6jFIZCtwtFX93vcZAvzH3YZD' recipient_number = '917227993344' url = 'https://graph.facebook.com/v17.0/109829165504454/messages' message_data = { "messaging_product": "whatsapp", "to": recipient_number, "type": "text", "text": { "body": "hiii" } } headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } response = requests.post(url, json = message_data, headers=headers) if response.status_code == 200: print('Message sent successfully.=====================') print(message_data,"*******************") else: print(f'Failed to send message. Status code: {response.status_code}',"-=-=-=-=-=-==-=--=-=-=-=") print(response.text,"###################") return Response({'status': True,'message': 'Message sent successfully.'},status = 200) app = Celery('myprofile', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0') class SendWhatsAppAlert(viewsets.ModelViewSet): def create(self, request): alert_conditions = Q(is_enabled=True, camera_id=1, camera_incharge_id=2, is_event_detected = True) event_conditions = (Q(camera_events_id = 8) | Q(camera_events_id = 9) | Q(camera_events_id = 10) | Q(camera_events_id = 11) & Q(camera_events__is_enabled = True)) alert_conditions &= event_conditions matching_alerts = Alert.objects.filter(alert_conditions) message = "Triggered_alerts:\n" for alert in matching_alerts: print("=====================================") message += f"Alert: {alert.alert_message}\n" message += f"Camera: {alert.camera}\n" message += f"Event: {alert.camera_events}\n" message += f"Camera Incharge: {alert.camera_incharge}\n" print(message,"=====================================") if message: recipient_number = "917227993344" # Replace with the recipient's WhatsApp number # Send the WhatsApp alert response = self.send_whatsapp_message(recipient_number, message) if response.status_code == 200: return Response({"message": "WhatsApp alert sent successfully."}) else: return Response({"message": "Failed to send WhatsApp alert."}, status = response.status_code) else: return Response({"message": "No alerts triggered."}) def send_whatsapp_message(self, recipient_number, message): access_token = 'EAAXuFyLeNOkBO68lPeKw8JZAQnZBcCZA7bFO6sJEu0vrHYHgG2g2XBwSd236kCBoAHc4RaUphJnjSRPjesod7MSlJBcZAOAuv2LK2FWwNoflB9Pt28fBHGaNSIZBZCpDZBfWUW0s2gBRYBzOP6sO2fkb0VOg0rb8h1PZCLoF9nTHd2I54Wja2jm7SkB8qiVsk2M6jFIZCtwtFX93vcZAvzH3YZD' recipient_number = '917227993344' url = 'https://graph.facebook.com/v17.0/109829165504454/messages' message_data = { "messaging_product": "whatsapp", "to": recipient_number, "type": "text", "text": { "body": message, } } headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } response = requests.post(url, json=message_data, headers=headers) if response.status_code == 200: print('Message sent successfully.') else: print(f'Failed to send message. Status code: {response.status_code}') return response class RoiCreateView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = RoiSerializer pagination_class = StandardResultsSetPagination queryset = Roi.objects.all() filter_backends = [filters.SearchFilter] search_fields = ['roi_name'] def create(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.add_roi'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) # rois = request.data.get('rois', []) camera_id = request.data.get('camera_id') camera_events_id = request.data.get('camera_event_ids',[]) roi_name = request.data.get('roi_name') coordinates = request.data.get('coordinates',[]) frame_number = request.data.get('frame_number') color = request.data.get('color') start_time = request.data.get('start_time') end_time = request.data.get('end_time') priority = request.data.get('priority') specific_analytics = request.data.get('specific_analytics') if not camera_id or not camera_events_id: return Response({'status': False, 'message': 'Required fields are missing.'}) camera_obj = Camera.objects.filter(id = camera_id).first() if not camera_obj: return Response({'status': False, 'message': 'Camera Not Found.'}) for event_id in camera_events_id: if not CameraEvent.objects.filter(id = event_id).exists(): return Response({'status': False, 'message': 'Invalid Camera Event ids.','data':[]}) camera_events_obj = CameraEvent.objects.filter(id__in = camera_events_id) if not camera_events_obj: return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # if camera_events_obj.filter(camera_event__icontains = 'Intrusion').exists(): # if not start_time or not end_time: # return Response({'status': False, 'message': 'Start_time and End_time are required for Intrusion Detection.'}) # try: # start_time = datetime.strptime(start_time, "%H:%M:%S").time() # end_time = datetime.strptime(end_time, "%H:%M:%S").time() # except Exception as e: # return Response({'status': False, 'message': 'Start-time or End-time is not correctly formatted.', 'data': str(e)}) roi_obj = Roi.objects.create( camera = camera_obj, roi_name = roi_name, frame_number = frame_number, color = color, coordinates = {'coordinates':coordinates}, priority = priority, specific_analytics = specific_analytics, ) if not user.is_superuser==True: roi_obj.created_by = user # if start_time and end_time: # roi_obj.start_time = start_time # roi_obj.end_time = end_time roi_obj.camera_events.set(camera_events_obj) roi_obj.save() serializer = RoiSerializer(roi_obj) return Response({ 'status': True, 'message': f"Your ROI '{roi_name}' created successfully.", "data":serializer.data }) # created_rois = [] # for roi_data in rois: # camera_id = roi_data.get('camera_id') # roi_name = roi_data.get('roi_name') # x1 = roi_data.get('x1') # y1 = roi_data.get('y1') # width = roi_data.get('width') # height = roi_data.get('height') # color = roi_data.get('color') # camera_events_data = roi_data.get('camera_event', []) # try: # camera_instance = Camera.objects.get(id=camera_id) # except Camera.DoesNotExist: # return Response({"message": "Invalid camera ID."}, status=400) # # Check if the specified camera events exist # # Fetch CameraEvent instances based on event names # existing_camera_events = CameraEvent.objects.filter( # Q(camera_event__in=camera_events_data) & Q(camera=camera_instance) # ) # if len(existing_camera_events) != len(camera_events_data): # return Response({"message": "Invalid camera event name."}) # # Create Roi instance # roi_instance = Roi.objects.create( # camera=camera_instance, # frame_number=frame_number, # roi_name=roi_name, # x1=x1, # y1=y1, # width=width, # height=height, # color=color # ) # # Associate existing CameraEvent instances with the current Roi # roi_instance.camera_events.add(*existing_camera_events) # created_rois.append(roi_instance) # serializer = self.serializer_class(created_rois, many=True) # return Response({'status': True, "message": "ROIs and CameraEvents created successfully.", "data": serializer.data}, status=200) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) def list(self, request, *args, **kwargs): queryset = self.get_queryset() try: user = self.request.user if not user.has_perm('myprofile.view_roi'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) camera_id = request.query_params.get('camera_id') camera_event_id = request.query_params.get('camera_event_id') search_query = request.query_params.get('search', '') if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) # if user.groups.filter(name='Admin').exists(): queryset = queryset.all().order_by('-id') # else: # queryset = queryset.filter(camera__area__location__organization__user=user).order_by('-id') if camera_id: queryset = queryset.filter(camera__id = camera_id) if camera_event_id: queryset = queryset.filter(camera_events__id = camera_event_id) if search_query: search_query = search_query.strip().lower() queryset = queryset.filter(Q(roi_name__icontains = search_query) | Q(camera__camera_name__icontains = search_query) | Q(camera_events__camera_event__icontains = search_query) ).order_by('-created_at').distinct() if queryset.exists(): page = self.paginate_queryset(queryset) if page is not None: serializer = RoiSerializer(page, many=True, context={'request': request}) serializer = self.get_paginated_response(serializer.data) else: serializer = RoiSerializer(queryset, many=True, context={'request': request}) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message": "Roi List Successfully", 'total_page': (count + limit - 1) // limit, 'count': count, 'data': serializer.data['results']}) else: return Response({"status": False, 'message': 'Roi Not Found', "data": []}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) class RoigetView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Roi.objects.all() serializer_class = RoiSerializer lookup_field = 'roi_id' pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): try: user = self.request.user roi_id = self.kwargs.get('roi_id') roi_obj = Roi.objects.filter(id = roi_id).first() if roi_obj: serializer = RoiSerializer(roi_obj,context={'request': request}) return Response({'status': True,'data':serializer.data},status = 200) else: return Response({'status': False,'message': 'Roi not found',"data" : []}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) def update(self, request, *args, **kwargs): try: user = self.request.user roi_id = self.kwargs.get('roi_id') roi_name = request.data.get('roi_name') frame_number = request.data.get('frame_number') # x1 = request.data.get('x1') # y1 = request.data.get('y1') # width = request.data.get('width') # height = request.data.get('height') coordinates = request.data.get('coordinates') color = request.data.get('color') start_time = request.data.get('start_time') end_time = request.data.get('end_time') priority = request.data.get('priority') specific_analytics = request.data.get('specific_analytics') camera_id = request.data.get('camera_id') camera_event_ids = request.data.get('camera_event_ids', []) # List of camera event IDs to be updated if not user.has_perm('myprofile.change_roi'): return Response({'status': False, 'message': 'You have no access to update Roi', 'data': []}) if not roi_id or not roi_name or not camera_id or not camera_event_ids: return Response({'status': False, 'message': 'All fields are required', 'data': []}) # Fetch the Roi instance roi_instance = Roi.objects.filter(id=roi_id).first() camera = Camera.objects.filter(id = camera_id).first() for event_id in camera_event_ids: if not CameraEvent.objects.filter(id = event_id).exists(): return Response({'status': False, 'message': 'Invalid Camera Event ids.','data':[]}) camera_events_obj = CameraEvent.objects.filter(id__in = camera_event_ids) if not camera_events_obj: return Response({'status': False, 'message': 'Invalid user IDs.', 'data':[]}) # if camera_events_obj.filter(camera_event__icontains = 'Intrusion').exists(): # if not start_time or not end_time: # return Response({'status': False, 'message': 'Start_time and End_time are required for Intrusion Detection.'}) # try: # start_time = datetime.strptime(start_time, "%H:%M:%S").time() # end_time = datetime.strptime(end_time, "%H:%M:%S").time() # except Exception as e: # return Response({'status': False, 'message': 'Start-time or End-time is not correctly formatted.', 'data': str(e)}) if roi_instance: # Update Roi fields roi_instance.roi_name = roi_name roi_instance.frame_number = frame_number roi_instance.coordinates = {'coordinates':coordinates} roi_instance.color = color roi_instance.priority = priority roi_instance.specific_analytics = specific_analytics roi_instance.camera = camera # Assign the camera ID directly to the foreign key field if not user.is_superuser==True: roi_instance.updated_by = user # if start_time and end_time: # roi_instance.start_time = start_time # roi_instance.end_time = end_time roi_instance.save() # Update associated CameraEvents roi_instance.camera_events.set(camera_events_obj) # Set the new CameraEvent IDs directly serializer = RoiSerializer(roi_instance) return Response({'status': True, 'message': f'Roi {roi_name} updated successfully.', "data": serializer.data}, status=200) else: return Response({'status': False, 'message': 'Invalid Roi ID or unauthorized access.', 'data': []}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'data': str(e)}) def destroy(self, request, *args, **kwargs): try: user = self.request.user roi_id = self.kwargs.get('roi_id') roi_obj = Roi.objects.filter(id = roi_id).delete() if roi_obj: return Response({'status': True,'message': f'Roi deleted successfully.'}) else: return Response({'status': False,'message': f'Failed to delete roi.','data':[]},status=400) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)},status=400) class ReportGenerationView(viewsets.ModelViewSet): def list(self, request, *args, **kwargs): organizations = Organization.objects.all() locations = Location.objects.all() areas = Area.objects.all() cameras = Camera.objects.all() location_data = LocationSerializer(locations, many=True).data area_data = AreaSerializer(areas, many=True).data camera_data = CameraSerializer(cameras, many=True).data organization_data = OrganizationSerializer(organizations, many=True).data report_data = { 'organizations': organization_data, 'locations': location_data, 'areas': area_data, 'cameras': camera_data, } return Response(report_data) class EventDetectionAPIView(APIView): def post(self, request): camera_id = request.data.get('camera_id') # Get camera_id from request data if camera_id is not None: call_command('detect_events', '--camera_id={}'.format(camera_id)) if request.data.get('is_async'): return Response({"message": f"Event detection started for camera {camera_id}."}) else: return Response({"message": f"Event detection completed for camera {camera_id}."}) else: return Response({"error": "Camera ID is required in the request data."}) # from django.http import StreamingHttpResponse # import redis # def video_stream(request, camera_id): # redis_client = redis.Redis(host='localhost', port=6379, db=0) # redis_key = f'frames:{camera_id}' # def frame_generator(): # while True: # frame = redis_client.get(redis_key) # if frame: # yield (b'--frame\r\n' # b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n') # return StreamingHttpResponse(frame_generator(), content_type='multipart/x-mixed-replace; boundary=frame') class Alert_event_camera_location_CountView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): try: user = self.request.user if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) organization_id = request.query_params.get('organization_id', None) if organization_id: alert_count = Alert.objects.filter(organization__id=organization_id, is_enabled=True).count() event_count = CameraEvent.objects.filter(organization__id = organization_id, is_enabled=True).count() camera_count = Camera.objects.filter(organization__id = organization_id).count() location_count = Location.objects.filter(organization__id = organization_id).count() else: alert_count = Alert.objects.filter(organization__organization_user = user, is_enabled=True).count() event_count = CameraEvent.objects.filter(organization__organization_user = user, is_enabled=True).count() camera_count = Camera.objects.filter(organization__organization_user = user).count() location_count = Location.objects.filter(organization__organization_user = user).count() data = { "alert_count": alert_count, "event_count": event_count, "camera_count": camera_count, "location_count": location_count } return Response({'status': True, 'data': data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class UndetectedEventCountView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): all_count = Alert.objects.count() helmet_count = Alert.objects.filter(Q(alert_message__icontains='Helmet: no')).count() glasses_count = Alert.objects.filter(Q(alert_message__icontains='Glasses: no')).count() gloves_count = Alert.objects.filter(Q(alert_message__icontains='Gloves: no')).count() shoes_count = Alert.objects.filter(Q(alert_message__icontains='Shoes: no')).count() jacket_count = Alert.objects.filter(Q(alert_message__icontains='Jacket: no')).count() data = { "all_count": all_count, "helmet_count": helmet_count, "glasses_count": glasses_count, "gloves_count": gloves_count, "shoes_count": shoes_count, "jacket_count": jacket_count } return Response({'status': True, 'data': data}) class UndetectedEventDateWiseCountView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Alert.objects.all() def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user start_date = self.request.query_params.get('start_date', None) end_date = self.request.query_params.get('end_date', None) if not user: return Response({'status': False, 'message': 'You can not access the data', 'data': []}) queryset = queryset.filter(organization__organization_user = user, is_enabled=True) if start_date and end_date: start_date = datetime.strptime(start_date, '%Y-%m-%d').date() end_date = datetime.strptime(end_date, '%Y-%m-%d').date() else: current_date = datetime.now().date() start_date = current_date.replace(day=1) next_month = start_date.replace(day=28) + timedelta(days=4) end_date = (next_month - timedelta(days=next_month.day)) # start_date = datetime.now().date() - timedelta(days=30) # end_date = datetime.now().date() date_wise_counts = [] if end_date - timedelta(days=7) == start_date: current_date = start_date while current_date <= end_date: total_count = Alert.objects.filter(created_at__date=current_date).count() date_wise_counts.append({ "date": current_date.strftime('%a'), "total": total_count }) current_date += timedelta(days=1) elif end_date - timedelta(days=31) > start_date: current_date = start_date.replace(day=1) while current_date <= end_date: next_month = current_date.replace(day=28) + timedelta(days=4) last_day_of_month = next_month - timedelta(days=next_month.day) total_count = queryset.filter( created_at__year=current_date.year, created_at__month=current_date.month ).count() date_wise_counts.append({ "date": current_date.strftime('%b'), "total": total_count }) current_date = last_day_of_month + timedelta(days=1) else: current_date = start_date while current_date <= end_date: total_count = queryset.filter(created_at__date=current_date).count() date_wise_counts.append({ "date": current_date.strftime('%Y-%m-%d'), "total": total_count }) current_date += timedelta(days=1) return Response({'status': True, 'data': {"date_wise_counts": date_wise_counts}}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) # class UndetectedEventPercentageView(viewsets.ModelViewSet): # permission_classes = [IsAuthenticated] # def list(self, request, *args, **kwargs): # # Total number of alerts # total_alerts = Alert.objects.count() # if total_alerts == 0: # data = { # "helmet_percentage": "0.0", # "glasses_percentage": "0.0", # "gloves_percentage": "0.0", # "shoes_percentage": "0.0", # "jacket_percentage": "0.0" # } # else: # # Count alerts where Helmet is not present # helmet_alerts = Alert.objects.filter(Q(alert_message__icontains='Helmet: no')).count() # helmet_percentage = round((helmet_alerts / total_alerts) * 100) # # Count alerts where Glasses are not present # glasses_alerts = Alert.objects.filter(Q(alert_message__icontains='Glasses: no')).count() # glasses_percentage = round((glasses_alerts / total_alerts) * 100) # # Count alerts where Gloves are not present # gloves_alerts = Alert.objects.filter(Q(alert_message__icontains='Gloves: no')).count() # gloves_percentage = round((gloves_alerts / total_alerts) * 100) # # Count alerts where Shoes are not present # shoes_alerts = Alert.objects.filter(Q(alert_message__icontains='Shoes: no')).count() # shoes_percentage = round((shoes_alerts / total_alerts) * 100) # jacket_alerts = Alert.objects.filter(Q(alert_message__icontains='Jacket: no')).count() # jacket_percentage = round((jacket_alerts / total_alerts) * 100) # data = { # "helmet_percentage": f"{helmet_percentage}", # "glasses_percentage": f"{glasses_percentage}", # "gloves_percentage": f"{gloves_percentage}", # "shoes_percentage": f"{shoes_percentage}", # "jacket_percentage": f"{jacket_percentage}" # } # return Response({'status': True, 'data': data}) class UndetectedEventPercentageView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Alert.objects.all() # event_queryset = CameraEvent.objects.all() def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user start_date = self.request.query_params.get('start_date', None) end_date = self.request.query_params.get('end_date', None) if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) event_queryset = CameraEvent.objects.filter(organization__organization_user = user, is_enabled=True) queryset = queryset.filter(organization__organization_user = user, is_enabled=True) if start_date and end_date: start_date = datetime.strptime(start_date, '%Y-%m-%d').date() end_date = datetime.strptime(end_date, '%Y-%m-%d').date() else: start_date = datetime.now().date() - timedelta(days=30) end_date = datetime.now().date() if start_date and end_date: queryset = queryset.filter(created_at__date__gte=start_date, created_at__date__lte=end_date) organization_id = request.query_params.get('organization_id', None) if organization_id: event_queryset = event_queryset.filter(organization__id = organization_id, is_enabled = True) data = {} for event in event_queryset: data[event.camera_event] = round((queryset.filter(camera_events__camera_event = event.camera_event).count()/queryset.count() if queryset else 0)*100) return Response({'status': True, 'data': data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class UndetectedEventDateWisePercentageView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): # Get start and end date from query parameters, default to 30 days ago and today start_date_param = self.request.query_params.get('start_date', None) end_date_param = self.request.query_params.get('end_date', None) # Parse start and end dates from query parameters try: start_date = datetime.strptime(start_date_param, '%Y-%m-%d').date() if start_date_param else timezone.now().date() - timedelta(days=30) end_date = datetime.strptime(end_date_param, '%Y-%m-%d').date() if end_date_param else timezone.now().date() except ValueError: # Handle invalid date format return Response({"error": "Invalid date format. Use 'YYYY-MM-DD'."}, status=400) # Date-wise and month-wise counts and percentages for the specified date range date_wise_counts = [] month_wise_counts = defaultdict(lambda: defaultdict(int)) current_date = start_date while current_date <= end_date: counts = Alert.objects.filter(created_at__date=current_date).aggregate( total=Count('id'), helmet=Count('id', filter=Q(alert_message__icontains='Helmet: no')), glasses=Count('id', filter=Q(alert_message__icontains='Glasses: no')), gloves=Count('id', filter=Q(alert_message__icontains='Gloves: no')), shoes=Count('id', filter=Q(alert_message__icontains='Shoes: no')), jacket=Count('id', filter=Q(alert_message__icontains='Jacket: no')) ) if counts['total'] > 0: helmet_percentage = round((counts['helmet'] / counts['total']) * 100) glasses_percentage = round((counts['glasses'] / counts['total']) * 100) gloves_percentage = round((counts['gloves'] / counts['total']) * 100) shoes_percentage = round((counts['shoes'] / counts['total']) * 100) jacket_percentage = round((counts['jacket'] / counts['total']) * 100) else: helmet_percentage = glasses_percentage = gloves_percentage = shoes_percentage = jacket_percentage = 'N/A' date_wise_counts.append({ "date": current_date.strftime('%Y-%m-%d'), "counts": { "helmet": f"{helmet_percentage} %", "glasses": f"{glasses_percentage} %", "gloves": f"{gloves_percentage} %", "shoes": f"{shoes_percentage} %", "jacket": f"{jacket_percentage} %" } }) # Adding the counts to the monthly data month = current_date.strftime('%Y-%m') for key, value in counts.items(): month_wise_counts[month][key] += value current_date += timedelta(days=1) # Move to the next day # Calculate month-wise percentages and format for response formatted_month_wise_counts = [] for month, counts in month_wise_counts.items(): total = counts['total'] if total > 0: helmet_percentage = round((counts['helmet'] / total) * 100) glasses_percentage = round((counts['glasses'] / total) * 100) gloves_percentage = round((counts['gloves'] / total) * 100) shoes_percentage = round((counts['shoes'] / total) * 100) jacket_percentage = round((counts['jacket'] / total) * 100) else: helmet_percentage = glasses_percentage = gloves_percentage = shoes_percentage = jacket_percentage = 'N/A' formatted_month_wise_counts.append({ "month": month, "counts": { "helmet": f"{helmet_percentage} %", "glasses": f"{glasses_percentage} %", "gloves": f"{gloves_percentage} %", "shoes": f"{shoes_percentage} %", "jacket": f"{jacket_percentage} %" } }) response_data = { "date_wise_counts": date_wise_counts, "month_wise_counts": formatted_month_wise_counts } return Response(response_data) import io from django.http import HttpResponse from openpyxl import Workbook from rest_framework import viewsets from rest_framework.permissions import IsAuthenticated from .models import Alert from django.core.mail import EmailMessage class ExportToExcelView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): # Get all alerts alerts = Alert.objects.all() # Calculate day-wise and month-wise data day_wise_data = {} month_wise_data = {} for alert in alerts: date_str = alert.created_at.strftime('%Y-%m-%d') month_str = alert.created_at.strftime('%Y-%m') if date_str not in day_wise_data: day_wise_data[date_str] = {'total': 0, 'helmet': 0, 'glasses': 0, 'gloves': 0, 'shoes': 0, 'jacket': 0} if month_str not in month_wise_data: month_wise_data[month_str] = {'total': 0, 'helmet': 0, 'glasses': 0, 'gloves': 0, 'shoes': 0, 'jacket': 0} day_wise_data[date_str]['total'] += 1 month_wise_data[month_str]['total'] += 1 if 'Helmet: no' in alert.alert_message: day_wise_data[date_str]['helmet'] += 1 month_wise_data[month_str]['helmet'] += 1 if 'Glasses: no' in alert.alert_message: day_wise_data[date_str]['glasses'] += 1 month_wise_data[month_str]['glasses'] += 1 if 'Gloves: no' in alert.alert_message: day_wise_data[date_str]['gloves'] += 1 month_wise_data[month_str]['gloves'] += 1 if 'Shoes: no' in alert.alert_message: day_wise_data[date_str]['shoes'] += 1 month_wise_data[month_str]['shoes'] += 1 if 'Jacket: no' in alert.alert_message: day_wise_data[date_str]['jacket'] += 1 month_wise_data[month_str]['jacket'] += 1 # Create a new workbook wb = Workbook() ws_day = wb.active ws_day.title = "Day Wise" ws_month = wb.create_sheet("Month Wise") # Write headers headers = ['Date', 'Total Alerts', 'Helmet Alerts', 'Helmet %', 'Glasses Alerts', 'Glasses %', 'Gloves Alerts', 'Gloves %', 'Shoes Alerts', 'Shoes %', 'Jacket Alerts', 'Jacket %'] ws_day.append(headers) ws_month.append(headers) # Write day-wise data for date_str, data in day_wise_data.items(): ws_day.append([ date_str, data['total'], data['helmet'], round((data['helmet'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['glasses'], round((data['glasses'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['gloves'], round((data['gloves'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['shoes'], round((data['shoes'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['jacket'], round((data['jacket'] / data['total']) * 100) if data['total'] > 0 else 'N/A', ]) # Write month-wise data for month_str, data in month_wise_data.items(): ws_month.append([ month_str, data['total'], data['helmet'], round((data['helmet'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['glasses'], round((data['glasses'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['gloves'], round((data['gloves'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['shoes'], round((data['shoes'] / data['total']) * 100) if data['total'] > 0 else 'N/A', data['jacket'], round((data['jacket'] / data['total']) * 100) if data['total'] > 0 else 'N/A', ]) # Generate a unique file name using a timestamp timestamp = timezone.now().strftime('%Y_%m_%d_%H%M%S') file_name = f'undetected_data_{timestamp}.xlsx' # Create the response with a unique file name response = HttpResponse(content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet') response['Content-Disposition'] = f'attachment; filename={file_name}' # Save the workbook to the response wb.save(response) return response import io from django.db.models import Count, Q from django.db.models.functions import TruncDate from django.http import HttpResponse from django.utils import timezone from rest_framework import viewsets from rest_framework.permissions import IsAuthenticated from reportlab.lib import colors from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle class ExportToPDFView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): # Get all alerts alerts = Alert.objects.all() total_alerts = alerts.count() # Calculate day-wise data day_wise_data = ( alerts .annotate(day=TruncDate('created_at')) .values('day') .annotate( total=Count('id'), helmet=Count('id', filter=Q(alert_message__icontains='Helmet: no')), glasses=Count('id', filter=Q(alert_message__icontains='Glasses: no')), gloves=Count('id', filter=Q(alert_message__icontains='Gloves: no')), shoes=Count('id', filter=Q(alert_message__icontains='Shoes: no')), jacket=Count('id', filter=Q(alert_message__icontains='Jacket: no')), ) .order_by('day') ) # Prepare data for PDF table data = [['Date', 'Total Alerts', 'Helmet (Count, %)', 'Glasses (Count, %)', 'Gloves (Count, %)', 'Shoes (Count, %)', 'Jacket (Count, %)']] for day_data in day_wise_data: total = day_data['total'] row = [ day_data['day'].strftime('%Y-%m-%d'), str(total), f"{day_data['helmet']} ({round((day_data['helmet'] / total) * 100, 2)} %)" if total else "0 (0 %)", f"{day_data['glasses']} ({round((day_data['glasses'] / total) * 100, 2)} %)" if total else "0 (0 %)", f"{day_data['gloves']} ({round((day_data['gloves'] / total) * 100, 2)} %)" if total else "0 (0 %)", f"{day_data['shoes']} ({round((day_data['shoes'] / total) * 100, 2)} %)" if total else "0 (0 %)", f"{day_data['jacket']} ({round((day_data['jacket'] / total) * 100, 2)} %)" if total else "0 (0 %)" ] data.append(row) # Prepare PDF buffer buffer = io.BytesIO() # Create the PDF object p = SimpleDocTemplate(buffer, pagesize=letter, topMargin=40, bottomMargin=40, leftMargin=40, rightMargin=40) # Define custom styles styles = getSampleStyleSheet() custom_heading_style = ParagraphStyle( name='CustomHeadingStyle', parent=styles['Heading1'], fontSize=18, textColor=colors.HexColor('#1f77b4'), # Custom color for text (hex color code) alignment=1, # Center alignment (0=left, 1=center, 2=right) spaceAfter=20 # Space after the heading ) custom_table_style = TableStyle([ ('BACKGROUND', (0, 0), (-1, 0), colors.cadetblue), ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), ('ALIGN', (0, 0), (-1, -1), 'CENTER'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('BOTTOMPADDING', (0, 0), (-1, 0), 12), ('BACKGROUND', (0, 1), (-1, -1), colors.beige), ('GRID', (0, 0), (-1, -1), 1, colors.black), ('FONTNAME', (0, 1), (-1, -1), 'Helvetica'), ('FONTSIZE', (0, 1), (-1, -1), 8), ]) # Add content content = [Paragraph('Month / Day wise Alert Data', custom_heading_style)] table = Table(data) table.setStyle(custom_table_style) content.append(table) # Add footer footer_style = ParagraphStyle('FooterStyle', alignment=1, fontSize=10, textColor=colors.black) content.append(Spacer(1, 20)) content.append(Paragraph(f'Total Alerts: {total_alerts}', footer_style)) content.append(Paragraph(f'Generated on: {timezone.now().strftime("%Y-%m-%d %H:%M:%S")}', footer_style)) # Build PDF p.build(content) # PDF response buffer.seek(0) response = HttpResponse(buffer, content_type='application/pdf') response['Content-Disposition'] = 'attachment; filename="day_wise_alert_data.pdf"' return response import csv from django.http import HttpResponse from rest_framework.views import APIView from rest_framework.permissions import IsAuthenticated from django.db.models import Count, Q from django.db.models.functions import TruncDate, TruncMonth class ExportToCSVView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): # Prepare the CSV response response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename=undetected_event_data.csv' csv_writer = csv.writer(response) # Write header csv_writer.writerow(['Time Period', 'Total Alerts', 'Helmet Alerts', 'Helmet %', 'Glasses Alerts', 'Glasses %', 'Gloves Alerts', 'Gloves %', 'Shoes Alerts', 'Shoes %', 'Jacket Alerts', 'Jacket %']) # Function to calculate and write data def calculate_and_write_data(time_period, alerts): total_alerts = alerts.count() helmet_alerts = alerts.filter(Q(alert_message__icontains='Helmet: no')).count() glasses_alerts = alerts.filter(Q(alert_message__icontains='Glasses: no')).count() gloves_alerts = alerts.filter(Q(alert_message__icontains='Gloves: no')).count() shoes_alerts = alerts.filter(Q(alert_message__icontains='Shoes: no')).count() jacket_alerts = alerts.filter(Q(alert_message__icontains='Jacket: no')).count() # Calculate percentages helmet_percentage = round((helmet_alerts / total_alerts) * 100, 2) if total_alerts != 0 else 0 glasses_percentage = round((glasses_alerts / total_alerts) * 100, 2) if total_alerts != 0 else 0 gloves_percentage = round((gloves_alerts / total_alerts) * 100, 2) if total_alerts != 0 else 0 shoes_percentage = round((shoes_alerts / total_alerts) * 100, 2) if total_alerts != 0 else 0 jacket_percentage = round((jacket_alerts / total_alerts) * 100, 2) if total_alerts != 0 else 0 # Write data to CSV csv_writer.writerow([time_period, total_alerts, helmet_alerts, helmet_percentage, glasses_alerts, glasses_percentage, gloves_alerts, gloves_percentage, shoes_alerts, shoes_percentage, jacket_alerts, jacket_percentage]) # Calculate day-wise data day_wise_data = ( Alert.objects .annotate(day=TruncDate('created_at')) .values('day') .annotate(total=Count('id')) .order_by('day') ) for day_data in day_wise_data: calculate_and_write_data(day_data['day'].strftime('%Y-%m-%d'), Alert.objects.filter(created_at__date=day_data['day'])) # Calculate month-wise data month_wise_data = ( Alert.objects .annotate(month=TruncMonth('created_at')) .values('month') .annotate(total=Count('id')) .order_by('month') ) for month_data in month_wise_data: calculate_and_write_data(month_data['month'].strftime('%Y-%m'), Alert.objects.filter(created_at__month=month_data['month'].month)) return response # views.py import csv import pandas as pd from rest_framework import generics from rest_framework.permissions import IsAuthenticated from django.http import HttpResponse from django_filters.rest_framework import DjangoFilterBackend from .models import Alert from .serializers import AlertSerializer import django_filters from reportlab.lib.pagesizes import letter from reportlab.pdfgen import canvas from io import BytesIO class AlertFilter(django_filters.FilterSet): camera = django_filters.CharFilter(field_name="camera__camera_name", lookup_expr='icontains') camera_events = django_filters.CharFilter(field_name="camera_events__camera_event", lookup_expr='icontains') created_at = django_filters.DateFromToRangeFilter() is_event_detect = django_filters.BooleanFilter() class Meta: model = Alert fields = ['camera', 'camera_events', 'created_at', 'is_event_detect', 'is_enabled'] class AlertReportView(generics.ListAPIView): queryset = Alert.objects.all() serializer_class = AlertSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend] filterset_class = AlertFilter def get(self, request, *args, **kwargs): format = request.GET.get('formet', None) queryset = self.filter_queryset(self.get_queryset()) if format == 'csv': return self.generate_csv(queryset) elif format == 'excel': return self.generate_excel(queryset) elif format == 'pdf': return self.generate_pdf(queryset) else: return super().get(request, *args, **kwargs) def generate_csv(self, queryset): response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="alerts.csv"' writer = csv.writer(response) writer.writerow(['Camera', 'Event', 'Created At', 'Is Event Detect', 'Is Enabled']) for alert in queryset: writer.writerow([alert.camera.camera_name, alert.camera_events.camera_event, alert.created_at, alert.is_event_detect, alert.is_enabled]) return response def generate_excel(self, queryset): response = HttpResponse(content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet') response['Content-Disposition'] = 'attachment; filename="alerts.xlsx"' df = pd.DataFrame(list(queryset.values('camera__camera_name', 'camera_events__camera_event', 'created_at', 'is_event_detect', 'is_enabled'))) df.columns = ['Camera', 'Event', 'Created At', 'Is Event Detect', 'Is Enabled'] with BytesIO() as b: writer = pd.ExcelWriter(b, engine='openpyxl') df.to_excel(writer, index=False, sheet_name='Alerts') writer.save() response.write(b.getvalue()) return response def generate_pdf(self, queryset): buffer = BytesIO() p = canvas.Canvas(buffer, pagesize=letter) width, height = letter p.drawString(100, height - 50, "Alert Report") p.drawString(100, height - 100, "Camera") p.drawString(200, height - 100, "Event") p.drawString(300, height - 100, "Created At") p.drawString(400, height - 100, "Is Event Detect") p.drawString(500, height - 100, "Is Enabled") y = height - 150 for alert in queryset: p.drawString(100, y, alert.camera.camera_name) p.drawString(200, y, alert.camera_events.camera_event) p.drawString(300, y, str(alert.created_at)) p.drawString(400, y, str(alert.is_event_detect)) p.drawString(500, y, str(alert.is_enabled)) y -= 30 p.showPage() p.save() buffer.seek(0) response = HttpResponse(buffer, content_type='application/pdf') response['Content-Disposition'] = 'attachment; filename="alerts.pdf"' return response class TopFourAlertDataView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = AlertSerializer queryset = Alert.objects.all() lookup_field = 'camera_event_id' pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user organization_id = request.query_params.get('organization_id', None) if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) camera_event_obj = None if queryset: camera_event_obj = queryset.filter(organization__organization_user = user, is_enabled=True).order_by('-created_at')[:4] if organization_id: camera_event_obj = camera_event_obj.filter(organization__id = organization_id, is_enabled = True) if camera_event_obj: page = self.paginate_queryset(camera_event_obj) if page is not None: serializer = AlertSerializer(page,many=True,context={'request': request}) serializer = self.get_paginated_response(serializer.data) else: serializer = AlertSerializer(camera_event_obj,many=True,context={'request': request}) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Camera Event List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({'status': False,'message': 'No Alerts Found.','data':[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) from django.utils import timezone as tz class AlertGetView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = AlertSerializer queryset = Alert.objects.all() pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_alert'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) search_query = request.query_params.get('search', '') start_date = request.query_params.get('start_date', '') end_date = request.query_params.get('end_date', '') organization_id = request.query_params.get('organization_id', '') location_id = request.query_params.get('location_id', '') area_id = request.query_params.get('area_id', '') camera_id = request.query_params.get('camera_id', '') event_id = request.query_params.get('camera_event_id','') alert_message = request.query_params.get('alert_message','') use_case_detection = request.query_params.get('use_case_detection', '') alert_obj = None # if user.groups.filter(name='Admin').exists(): if queryset: alert_obj = queryset.filter(organization__organization_user = user).order_by('-created_at') # elif user.groups.filter(name='Manager').exists(): # alert_obj = queryset.filter(camera__location__location_user = user).order_by('-created_at') if start_date and end_date: if start_date > end_date: return Response({'status': False, 'message': "Start date should be less then end date"}) date_format = '%Y-%m-%d' # start_date,end_date, error_message = validate_dates(start_date, end_date) # if error_message: # return Response({'status': False, 'message': error_message,"data":[]}) start_date = datetime.strptime(start_date, date_format).date() end_date = datetime.strptime(end_date, date_format).date() if start_date == end_date: alert_obj = alert_obj.filter(created_at__date=start_date).order_by('-created_at') else: alert_obj = alert_obj.filter(created_at__date__gte=start_date, created_at__date__lte=end_date).order_by('-created_at') elif start_date: date_format = '%Y-%m-%d' start_date = datetime.strptime(start_date, date_format).date() if isinstance(start_date, date): alert_obj = alert_obj.filter(created_at__date=start_date).order_by('-created_at') else: return Response({"status": False, "message": "Enter correct format of start_date","data":str(e)}) if organization_id: alert_obj = alert_obj.filter(organization__id = organization_id) if location_id: alert_obj = alert_obj.filter(location__id = location_id) if area_id: alert_obj = alert_obj.filter(area__id = area_id) if camera_id: alert_obj = alert_obj.filter(camera__id = camera_id) if event_id: alert_obj = alert_obj.filter(camera_events__id = event_id) if alert_message: alert_obj = alert_obj.filter(alert_message__icontains = alert_message) if not alert_obj: return Response({'status': False,'message': "No data found!","data":[]}) if search_query: search_query = search_query.strip().lower() alert_obj = alert_obj.filter(Q(alert_message__icontains = search_query)| Q(location__loc_name__icontains = search_query) | Q(camera_events__camera_event__icontains = search_query) | Q(organization__org_name__icontains = search_query) | Q(created_at__icontains = search_query)).order_by('-created_at') if use_case_detection: alert_obj = alert_obj.filter(camera_events__camera_event = use_case_detection).order_by('-created_at') if alert_obj: page = self.paginate_queryset(alert_obj) if page is not None: serializer = AlertSerializer(page,many=True,context={'request': request}) serializer = self.get_paginated_response(serializer.data) else: serializer = AlertSerializer(alert_obj,many=True,context={'request': request}) count = serializer.data['count'] limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Alert List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': serializer.data['results']}) else: return Response({"status": True, "message":"No Alert Found","data":[]}) except Exception as e: return Response({"status": False, "message": "Something went wrong","data":str(e)}) import pandas as pd class ExportReportCSV(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = AlertSerializer queryset = Alert.objects.all() pagination_class = StandardResultsSetPagination def send_report_email(self, user_email, csv_file, org_name, message): genrated_date = datetime.today().strftime("%d %B, %Y") subject = f'{org_name} Report - {genrated_date}' from_email= settings.EMAIL_HOST_USER email = EmailMessage(subject, message, from_email, to=[user_email]) email.attach_file(csv_file) email.send() def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user search_query = request.query_params.get('search', '') start_date = request.query_params.get('start_date', '') end_date = request.query_params.get('end_date', '') organization_id = request.query_params.get('organization_id', '') location_id = request.query_params.get('location_id', '') camera_id = request.query_params.get('camera_id', '') area_id = request.query_params.get('area_id', '') event_id = request.query_params.get('camera_event_id','') alert_message = request.query_params.get('alert_message','') use_case_detection = request.query_params.get('use_case_detection', '') alert_obj = None # if user.groups.filter(name='Admin').exists(): if queryset: alert_obj = queryset.filter(organization__organization_user = user).order_by('-created_at') # elif user.groups.filter(name='Manager').exists(): # alert_obj = queryset.filter(camera__location__location_user = user).order_by('-created_at') if start_date and end_date: date_format = '%Y-%m-%d' # start_date,end_date, error_message = validate_dates(start_date, end_date) # if error_message: # return Response({'status': False, 'message': error_message,"data":[]}) start_date = datetime.strptime(start_date, date_format).date() end_date = datetime.strptime(end_date, date_format).date() # start_datetime = tz.make_aware(datetime.combine(start_date, datetime.min.time())) # end_datetime = tz.make_aware(datetime.combine(end_date, datetime.min.time())) if start_date == end_date: alert_obj = alert_obj.filter(created_at__date=start_date).order_by('-created_at') else: alert_obj = alert_obj.filter(created_at__date__gte=start_date, created_at__date__lte=end_date).order_by('-created_at') elif start_date: date_format = '%Y-%m-%d' start_date = datetime.strptime(start_date, date_format).date() if isinstance(start_date, date): alert_obj = alert_obj.filter(created_at__date=start_date).order_by('-created_at') else: return Response({"status": False, "message": "Enter correct format of start_date","data":str(e)}) if organization_id: alert_obj = alert_obj.filter(organization__id = organization_id) if location_id: alert_obj = alert_obj.filter(location__id = location_id) if area_id: alert_obj = alert_obj.filter(area__id = area_id) if camera_id: alert_obj = alert_obj.filter(camera__id = camera_id) if event_id: alert_obj = alert_obj.filter(camera_events__id = event_id) if alert_message: alert_obj = alert_obj.filter(alert_message__icontains = alert_message) if not alert_obj: return Response({'status': False,'message': "No data found!","data":[]}) if use_case_detection: alert_obj = alert_obj.filter(camera_events__camera_event = use_case_detection).order_by('-created_at') if alert_obj.exists(): page = self.paginate_queryset(alert_obj) if page is not None: serializer = AlertSerializer(page,many=True,context={'request': request}) serializer = self.get_paginated_response(serializer.data) else: serializer = AlertSerializer(alert_obj,many=True,context={'request': request}) # result_data = serializer.data['results'] # data_lst = [] # for res_data in result_data: # filter_data = { # "Organization":res_data['organization_name'], # "Location":res_data['location_name'], # "Detected Camera Name":res_data['camera_name'], # "Event Name":res_data['camera_events_name'], # "Alert Message":res_data['alert_message'], # "Created On":res_data["created_at"], # } # data_lst.append(filter_data) sl_data = AlertSerializer(alert_obj,many=True,context={'request': request}) columns = ['camera_name', 'camera_events_name', 'frame', 'location_name', 'organization_name', 'alert_message', 'created_at'] data =pd.DataFrame(sl_data.data,columns=columns) # print(data) file_path = settings.MEDIA_ROOT+'/report_csv/report_data.csv' data.to_csv(file_path, index=False) if sl_data: check = Organization.objects.filter(organization_user = user).first() if check.email_notification_is_enabled: if start_date and end_date: message = f'I hope this email finds you well.\n\n Please find attached the comprehensive report of the records from {start_date} to {end_date} which you requested. This report includes detailed data and analysis, which we believe will provide you with valuable insights.\n\nThank you.' else: message = 'I hope this email finds you well.\n\n Please find attached the comprehensive report of the overall records you requested. This report includes detailed data and analysis, which we believe will provide you with valuable insights.\n\nThank you.' self.send_report_email(user.email, file_path, check.org_name, message) response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="alerts.csv"' csv_file_writer = csv.writer(response) header_row = ['Organization', 'Camera Location', 'Created At', 'Event ID', 'Alert Message', 'frame'] csv_file_writer.writerow(header_row) for result in serializer.data['results']: row = [ result['organization_name'], result['location_name'], result['created_at'], result['camera_events_name'] if result['camera_events_name'] else '', result['alert_message'], result['frame'] ] csv_file_writer.writerow(row) return Response({"status": True, "message":"File generated Successfully.","data":settings.URL+"/media/report_csv/report_data.csv"}) else: return Response({"status": True, "message":"No Alert Found","data":[]}) except Exception as e: return Response({"status": False, "message": "Something went wrong","data":str(e)}) # views.py # from rest_framework.views import APIView # from rest_framework.response import Response # from rest_framework import status # import cv2 # from ultralytics import YOLO # from django.core.files.storage import FileSystemStorage # from django.conf import settings # import cv2 # import tempfile # from django.core.files.base import ContentFile from django.core.files.storage import default_storage # def detect_objects_in_roi(roi_frame): # # fps = 30 # # fourcc = cv2.VideoWriter_fourcc(*'mp4v') # # out = cv2.VideoWriter("output.mp4", fourcc, fps, (roi_frame.shape[1], roi_frame.shape[0])) # Output video filename, codec, fps, frame size # model = YOLO('best.pt') # results = model(roi_frame)[0] # classes = [ # 'Helmet: yes', 'Glasses: yes', 'Gloves: yes', 'Shoes: yes', 'Person', # 'Helmet: no', 'Glasses: no', 'Gloves: no', 'Shoes: no' # ] # detacted_classes = [] # for result in results: # x1, y1, x2, y2, conf, class_idx = result.boxes.data.tolist()[0] # class_idx = int(class_idx) # class_label = classes[class_idx] # detacted_classes.append(class_label) # return detacted_classes # def handle(file_path): # cap = cv2.VideoCapture(file_path) # total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # print(f"Total number of frames: {total_frames}") # # Read frames from the RTSP stream # ret, frame = cap.read() # detactions = [] # skip = True # # Initialize loop variable # counter = 0 # while counter < total_frames: # counter += 100 # # Skip one iteration # if skip: # skip = False # Reset the skip variable # continue # Skip the rest of the loop # # Capture frame-by-frame # ret, frame = cap.read() # if ret: # detected_objects = detect_objects_in_roi(frame) # for i in detected_objects: # if i in ['Helmet: no', 'Glasses: no', 'Gloves: no', 'Shoes: no']: # detactions.append({"detection":i, "frame":frame}) # else: # break # return detactions # class RunScriptOnVideoApiView(APIView): # def post(self, request, *args, **kwargs): # video_file = request.FILES.get('video_file') # if not video_file: # return Response({'error': 'No video file provided'}, status=400) # path = default_storage.save('tmp/some_video.mp4', ContentFile(video_file.read())) # tmp_file = os.path.join(settings.MEDIA_ROOT, path) # detections = handle(tmp_file) # for detection in detections: # # Convert the frame to a format Django can handle (like JPEG) # _, buffer = cv2.imencode('.jpg', detection["frame"]) # content_file = ContentFile(buffer.tobytes(),name='detection.jpg') # # Create a new Alert instance # alert = Alert( # alert_message=detection["detection"], # frame=content_file, # camera=Camera.objects.first(), # camera_events=CameraEvent.objects.first() # ) # alert.save() # # save camera inchareg # alert.camera_incharge.add(CustomUser.objects.first()) # # Clean up the temporary video file if not needed # os.remove(tmp_file) # # Here, use 'tmp_file' as the path to your video in your processing logic # # Implement your video processing and detection logic here # serializer = VideoUploadSerializer(alert,context={'request': request}) # return Response({"status": True,'message': 'Alert Generated!', 'data': serializer.data}) # ======================================================================================================================== # def post(self, request, *args, **kwargs): # # Check if a video file is in the request # if 'video_file' not in request.FILES: # return Response({'message': 'No video file provided.'}, status=status.HTTP_400_BAD_REQUEST) # # Get the video file from the request # video_file = request.FILES['video_file'] # # Save the video file in the media folder # fs = FileSystemStorage() # filename = fs.save(video_file.name, video_file) # video_path = fs.url(filename) # # Full path to the saved video file # full_video_path = os.path.join(settings.MEDIA_ROOT, filename) # if not full_video_path: # return Response({'message': 'Please provide a video URL.'}, status=status.HTTP_400_BAD_REQUEST) # try: # cap = cv2.VideoCapture(full_video_path) # if not cap.isOpened(): # return Response({'message': 'Failed to open the video.'}, status=status.HTTP_400_BAD_REQUEST) # total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # detactions = [] # skip = True # # Initialize loop variable # counter = 0 # while counter < total_frames: # counter += 1 # # Skip one iteration # if skip: # skip = False # Reset the skip variable # continue # Skip the rest of the loop # # Capture frame-by-frame # ret, frame = cap.read() # if ret: # detected_objects = self.detect_objects_in_roi(frame) # for i in detected_objects: # if 'no' in i: # self.create_alert(i, frame) # else: # break # cap.release() # cv2.destroyAllWindows() # return Response({'message': 'Script executed successfully', 'detactions': detactions}) # except Exception as e: # return Response({'message': 'Something went wrong', 'error': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) # def detect_objects_in_roi(self, roi_frame): # # Replace this section with your detection logic # model = YOLO('best.pt') # results = model(roi_frame)[0] # classes = [ # 'Helmet: yes', 'Glasses: yes', 'Gloves: yes', 'Shoes: yes', 'Person', # 'Helmet: no', 'Glasses: no', 'Gloves: no', 'Shoes: no' # ] # detected_classes = [] # for result in results: # x1, y1, x2, y2, conf, class_idx = result.boxes.data.tolist()[0] # class_idx = int(class_idx) # class_label = classes[class_idx] # detected_classes.append(class_label) # return detected_classes # def create_alert(self, alert_message, frame): # # Create a temporary file # # with tempfile.NamedTemporaryFile(delete=False, suffix= '.jpg') as temp_file: # # # Save the frame to the temporary file # # cv2.imwrite(temp_file.name, frame) # # # Open the temporary file in read mode # # temp_file.seek(0) # # django_file = ContentFile(temp_file.read(), name='frame.jpg') # # # Save the file using FileSystemStorage # # fs = FileSystemStorage() # # frame_name = fs.save(django_file.name, django_file) # # frame_url = fs.url(frame_name) # # Rest of your code # camera = Camera.objects.first() # camera_event = CameraEvent.objects.first() # camera_incharge = CustomUser.objects.first() # alert = Alert( # camera=camera, # camera_events=camera_event, # # frame=frame_url, # alert_message=alert_message, # camera_location = camera_event.camera.area.location, # ) # alert.save() # alert.camera_incharge.add(camera_incharge) # # Delete the temporary file # # os.unlink(temp_file.name) # ================================================================================================================== class UserOrganizationLoactionAreaIdView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CustomUserSerializer def list(self, request, *args, **kwargs): try: user = self.request.user if not user: return Response({'status': False, 'message': 'You cannot see user.', 'data': []}) organization_id = request.query_params.get('organization_id') location_id = request.query_params.get('location_id') area_id = request.query_params.get('area_id') if organization_id: if not Organization.objects.filter(id=organization_id).exists(): return Response({'status': False, 'message': 'Organization not found.', 'data': []}) if location_id: if not Location.objects.filter(id=location_id).exists(): return Response({'status': False, 'message': 'Location not found.', 'data': []}) if area_id: if not Area.objects.filter(id=area_id).exists(): return Response({'status': False, 'message': 'Area not found.', 'data': []}) if organization_id and location_id and area_id: # Filter by project, phase, and tower users = CustomUser.objects.filter(organization__id=organization_id, location__id=location_id, area__id=area_id) elif organization_id and location_id: # Filter by project and phase users = CustomUser.objects.filter(organization__id=organization_id, location__id=location_id) elif organization_id and area_id: # Filter by project and tower users = CustomUser.objects.filter(organization__id=organization_id, area__id=area_id) elif location_id and area_id: # Filter by phase and tower users = CustomUser.objects.filter(location__id=location_id, area__id=area_id) elif organization_id: # Filter by project users = CustomUser.objects.filter(organization__id=organization_id) elif location_id: # Filter by phase users = CustomUser.objects.filter(location__id=location_id) elif area_id: # Filter by tower users = CustomUser.objects.filter(area__id=area_id) else: return Response({'status': False, 'message': 'Please provide a project, phase, or tower.', 'data': []}) if users.exists(): serializer = CustomUserSerializer(users, many=True, context={'request': request}) return Response({'status': True, 'message': 'User list successfully.', 'data': serializer.data}) else: return Response({'status': False, 'message': 'No user found.', 'data': []}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong.', 'data': str(e)}) class OrganizationLocationAreaIdView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = OrganizationSerializer def list(self, request, *args, **kwargs): try: user = self.request.user if not user: return Response({'status': False, 'message': 'You cannot see organization.', 'data': []}) organization_id = request.query_params.get('organization_id') location_id = request.query_params.get('location_id') area_id = request.query_params.get('area_id') camera_id = request.query_params.get('camera_id') if organization_id: if not Organization.objects.filter(id=organization_id).exists(): return Response({'status': False, 'message': 'Organization not found.', 'data': []}) if location_id: if not Location.objects.filter(id=location_id).exists(): return Response({'status': False, 'message': 'Location not found.', 'data': []}) if area_id: if not Area.objects.filter(id=area_id).exists(): return Response({'status': False, 'message': 'Area not found.', 'data': []}) if camera_id: if not Camera.objects.filter(id=camera_id).exists(): return Response({'status': False, 'message': 'Camera not found.', 'data': []}) if organization_id and location_id and area_id: # Filter by project, phase, and tower organizations = Organization.objects.filter(id=organization_id, location__id=location_id, area__id=area_id) elif organization_id and location_id: # Filter by project and phase organizations = Organization.objects.filter(id=organization_id, location__id=location_id) elif organization_id and area_id: # Filter by project and tower organizations = Organization.objects.filter(id=organization_id, area__id=area_id) elif location_id and area_id: # Filter by phase and tower organizations = Organization.objects.filter(location__id=location_id, area__id=area_id) elif organization_id: # Filter by project organizations = Organization.objects.filter(id=organization_id) elif location_id: # Filter by phase organizations = Organization.objects.filter(location__id=location_id) elif area_id: # Filter by tower organizations = Organization.objects.filter(area__id=area_id) else: return Response({'status': False, 'message': 'Please provide a project, phase, or tower.', 'data': []}) if organizations.exists(): serializer = OrganizationSerializer(organizations, many=True) return Response({'status': True, 'message': 'Organization list successfully.', 'data': serializer.data}) else: return Response({'status': False, 'message': 'No organization found.', 'data': []}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong.', 'data': str(e)}) class OrganizationLocationAreaCameraUserView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] serializer_class = CustomUserSerializer queryset = CustomUser.objects.all() filter_backends = [filters.SearchFilter] # Add the filter backend for search functionality search_fields = ['first_name', 'last_name'] def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user: return Response({'status': False, 'message': 'You cannot see User.', 'data': []}) organization_id = request.query_params.get('organization_id') location_id = request.query_params.get('location_id') area_id = request.query_params.get('area_id') camera_id = request.query_params.get('camera_id') if queryset: queryset = queryset.exclude(groups__name = "super_admin").exclude(is_superuser = True) if organization_id and location_id and area_id and camera_id: # Filter by project, phase, tower, and camera users = queryset.filter(organization__id=organization_id, location__id=location_id, area__id=area_id, camera__id=camera_id) elif organization_id and location_id and area_id: # Filter by project, phase, and tower users = queryset.filter(organization__id=organization_id, location__id=location_id, area__id=area_id) elif organization_id and location_id: # Filter by project and phase users = queryset.filter(organization__id=organization_id, location__id=location_id) elif organization_id and area_id: # Filter by project and tower users = queryset.filter(organization__id=organization_id, area__id=area_id) elif location_id and area_id: # Filter by phase and tower users = queryset.filter(location__id=location_id, area__id=area_id) elif organization_id: # Filter by project users = queryset.filter(organization__id=organization_id) elif location_id: # Filter by phase users = queryset.filter(location__id=location_id) elif area_id: # Filter by tower users = queryset.filter(area__id=area_id) else: return Response({'status': False, 'message': 'Please provide a project, phase, or tower.', 'data': []}) if users.exists(): serializer = CustomUserSerializer(users, many=True, context={'request': request}) return Response({'status': True, 'message': 'User list successfully.', 'data': serializer.data}) else: return Response({'status': False, 'message': 'No user found.', 'data': []}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong.', 'data': str(e)}) import subprocess from django.http import HttpResponse import subprocess from django.http import HttpResponse # def stream_rtsp(request): # try: # # Replace 'rtsp://...' with your RTSP stream URL # rtsp_url = 'rtsp://192.168.1.3:554/user=admin&password=&channel=4&stream=0.sdp?' # # Replace '/path/to/output/stream.m3u8' with your desired output file path # output_file = '/home/savan/kalpesh/visnx-backend/Nettyfy_visnx/media/hls_file/hls_file_all/stream.m3u8' # # Start FFmpeg subprocess to transcode RTSP stream to HLS # process = subprocess.Popen(['ffmpeg', '-i', rtsp_url, '-c:v', 'copy', '-c:a', 'aac', '-strict', 'experimental', '-f', 'hls', '-hls_time', '4', '-hls_list_size', '6', '-hls_wrap', '10', output_file], stdout=subprocess.PIPE, stderr=subprocess.PIPE) # # Capture output and error streams for debugging # stdout, stderr = process.communicate() # print("FFmpeg stdout:", stdout.decode('utf-8')) # print("FFmpeg stderr:", stderr.decode('utf-8')) # base_url = 'http://192.168.1.23:8001/' # hls_url = base_url + output_file # print("HLS URL:======================", hls_url) # # Return HTTP response with the URL of the transcoded HLS stream # return HttpResponse(hls_url) # except Exception as e: # # Handle any exceptions and return an error response # print("Error:", e) # return HttpResponse("An error occurred while processing the RTSP stream.", status=500) # views.py from rest_framework import status from rest_framework.decorators import api_view from rest_framework.response import Response import ffmpeg @api_view(['POST']) def convert_rtsp_to_hls(request): if request.method == 'POST': rtsp_url = request.data.get('rtsp_url') # Assuming the RTSP URL is passed in the request data try: output_file_path = '/home/savan/kalpesh/visnx-backend/Nettyfy_visnx/media/hls_file/hls_file_all/stream.m3u8' ( ffmpeg .input(rtsp_url) .output(output_file_path, format='hls') .run() ) # Construct the base URL base_url = request.build_absolute_uri('/')[:-1] hls_file_url = os.path.join(base_url, output_file_path) return Response({'url': hls_file_url}, status=status.HTTP_200_OK) except ffmpeg.Error as e: return Response({'message': f'Error during conversion: {e.stderr.decode()}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) else: return Response({'message': 'Only POST requests are allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) import os from rest_framework.decorators import api_view from rest_framework.response import Response from rest_framework import status from django.conf import settings @api_view(['GET']) def get_hls_file(request): if request.method == 'GET': hls_file_path = os.path.join(settings.MEDIA_ROOT, 'hls_file', 'hls_file_all', 'stream.m3u8') # Construct the base URL base_url = request.build_absolute_uri('/')[:-1] hls_url = f"{base_url}/{hls_file_path}" return Response({'url': hls_url}, status=status.HTTP_200_OK) else: return Response({'message': 'Only GET requests are allowed'}, status=status.HTTP_405_METHOD_NOT_ALLOWED) from django.core.files.storage import FileSystemStorage from rest_framework.response import Response from rest_framework import status from rest_framework.views import APIView import os import subprocess from django.conf import settings from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from .models import CameraEvent, Camera, Alert from .serializers import CameraEventSerializer import cv2 import mediapipe as mp from django.core.files.base import ContentFile import datetime import pandas as pd mp_pose = mp.solutions.pose def detect_fall(keypoints, fall_threshold_y=0.3, fall_threshold_x=0.1): if mp_pose.PoseLandmark.NOSE.value not in keypoints or \ mp_pose.PoseLandmark.RIGHT_ANKLE.value not in keypoints or \ mp_pose.PoseLandmark.LEFT_ANKLE.value not in keypoints: return False head_y = keypoints[mp_pose.PoseLandmark.NOSE.value].y right_ankle_x = keypoints[mp_pose.PoseLandmark.RIGHT_ANKLE.value].x left_ankle_x = keypoints[mp_pose.PoseLandmark.LEFT_ANKLE.value].x # Ensure that extracted values are numeric if not all(isinstance(val, (int, float)) for val in [head_y, right_ankle_x, left_ankle_x]): return False fall_detected = (head_y - right_ankle_x) > fall_threshold_y and abs(right_ankle_x - left_ankle_x) < fall_threshold_x return fall_detected import cv2 import mediapipe as mp def run_fall_detection(video_file): cap = cv2.VideoCapture(video_file) pose = mp.solutions.pose.Pose(min_detection_confidence=0.5, min_tracking_confidence=0.5) fall_threshold_y = 0.3 fall_threshold_x = 0.1 frame_number = 0 fall_frames = [] while cap.isOpened(): ret, frame = cap.read() if not ret: break frame_number += 1 frame = cv2.resize(frame, (640, 480)) rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) results = pose.process(rgb_frame) if results.pose_landmarks: keypoints = {landmark: landmark_data for landmark, landmark_data in enumerate(results.pose_landmarks.landmark)} if detect_fall(keypoints, fall_threshold_y, fall_threshold_x): fall_frames.append({'frame_number': frame_number, 'frame': frame}) # Include frame_number roi_color = (0, 0, 255) # Red color head_coords = (int(keypoints[mp_pose.PoseLandmark.NOSE.value].x * frame.shape[1]), int(keypoints[mp_pose.PoseLandmark.NOSE.value].y * frame.shape[0])) right_ankle_coords = (int(keypoints[mp_pose.PoseLandmark.RIGHT_ANKLE.value].x * frame.shape[1]), int(keypoints[mp_pose.PoseLandmark.RIGHT_ANKLE.value].y * frame.shape[0])) cv2.rectangle(frame, head_coords, right_ankle_coords, roi_color, 2) message = f"Fall Detected @Frame {frame_number}" cv2.putText(frame, message, (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, roi_color, 2, cv2.LINE_AA) print(message) mp.solutions.drawing_utils.draw_landmarks(frame, results.pose_landmarks, mp.solutions.pose.POSE_CONNECTIONS) cv2.imshow('Fall Detection', frame) if cv2.waitKey(1) & 0xFF == ord('q'): break cap.release() cv2.destroyAllWindows() return fall_frames class FallDetectionViewSet(APIView): def post(self, request, *args, **kwargs): try: video_file = request.FILES.get('video_file') if not video_file: return Response({'message': 'No video file provided'}, status=status.HTTP_400_BAD_REQUEST) path = default_storage.save('tmp/some_video.mp4', ContentFile(video_file.read())) tmp_file = os.path.join(settings.MEDIA_ROOT, path) fall_detected_frames = run_fall_detection(tmp_file) # Check if "FallDetected" event exists in the CameraEvent table fall_event, _ = CameraEvent.objects.get_or_create(camera_event="FallDetected") for fall_frame in fall_detected_frames: frame_number = fall_frame['frame_number'] frame = fall_frame['frame'] # Convert the frame to a format Django can handle (like JPEG) _, buffer = cv2.imencode('.jpg', frame) content_file = ContentFile(buffer.tobytes(), name='detection.jpg') # Create a new Alert instance alert = Alert( alert_message=f"Fall detected", frame=content_file, camera=Camera.objects.first(), # Provide your camera instance here camera_events=fall_event # Provide your camera event instance here ) alert.save() # Save camera incharge (You may adjust this according to your model structure) alert.camera_incharge.add(CustomUser.objects.first()) return Response({'status': True, 'message': 'Fall detected in the video'}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'error': str(e)}) def fire_detection(file_path): cap = cv2.VideoCapture(file_path) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) detections = [] for counter in range(total_frames): ret, frame = cap.read() if not ret: break hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) lower_red = np.array([0, 120, 70]) upper_red = np.array([10, 255, 255]) mask = cv2.inRange(hsv, lower_red, upper_red) contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) if contours: x, y, w, h = cv2.boundingRect(np.vstack(contours)) cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 0, 255), 2) cv2.putText(frame, 'ALERT: FIRE', (10, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2, cv2.LINE_AA) message = f"Fire Detected" print(message) detections.append({"detection": message, "frame": frame.copy()}) # Save a copy of the frame cap.release() return detections class FireDetectionApiView(APIView): def post(self, request, *args, **kwargs): try: video_file = request.FILES.get('video_file') if not video_file: return Response({'message': 'No video file provided'}, status=status.HTTP_400_BAD_REQUEST) path = default_storage.save('tmp/some_video.mp4', ContentFile(video_file.read())) tmp_file = os.path.join(settings.MEDIA_ROOT, path) fire_detected = fire_detection(tmp_file) # Check if "FireDetection" event exists in the CameraEvent table fire_event, _ = CameraEvent.objects.get_or_create(camera_event="FireDetection") for detection in fire_detected: # Convert the frame to a format Django can handle (like JPEG) _, buffer = cv2.imencode('.jpg', detection["frame"]) content_file = ContentFile(buffer.tobytes(), name='detection.jpg') # Create a new Alert instance alert = Alert( alert_message=detection["detection"], frame=content_file, camera=Camera.objects.first(), camera_events=fire_event ) alert.save() # save camera incharge alert.camera_incharge.add(CustomUser.objects.first()) return Response({'status': True, 'message': 'Fire detected in the video'}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'error': str(e)}) def intrusion_detection(file_path, start_time_str, end_time_str): width = 480 height = 360 start_time = datetime.datetime.strptime(start_time_str, "%H:%M").time() end_time = datetime.datetime.strptime(end_time_str, "%H:%M").time() cap = cv2.VideoCapture(file_path) fgbg = cv2.createBackgroundSubtractorKNN() fixed_roi = [(50, 150), (50, 300), (200, 300), (200, 150)] # Initialize empty list for detections detections_list = [] frame_number = 0 while True: ret, frame = cap.read() if not ret: break # Break the loop if there are no more frames frame_number += 1 resized_frame = cv2.resize(frame, (width, height)) fgmask = fgbg.apply(resized_frame) kernel = np.ones((3, 3), np.uint8) # Adjust kernel size as needed opening = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel, iterations=2) closing = cv2.morphologyEx(opening, cv2.MORPH_CLOSE, kernel, iterations=2) detected_time = None # Initialize detected_time contours, _ = cv2.findContours(closing, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) for cnt in contours: area = cv2.contourArea(cnt) if area > 500: x, y, w, h = cv2.boundingRect(cnt) if fixed_roi[0][0] <= x <= fixed_roi[2][0] and fixed_roi[0][1] <= y <= fixed_roi[2][1]: current_time = datetime.datetime.now().time() if start_time <= current_time <= end_time: cv2.rectangle(resized_frame, (x, y), (x + w, y + h), (0, 255, 0), 2) cv2.putText(resized_frame, "Person Going Inside the Door (Green ROI)", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) detected_time = frame_number detection_message = {'detection': f"Person Detected", 'frame': resized_frame} detections_list.append(detection_message) else: cv2.rectangle(resized_frame, (x, y), (x + w, y + h), (0, 0, 255), 2) cv2.putText(resized_frame, "Intrusion Detected (Red ROI)", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) detected_time = frame_number detection_message = {'detection': f"Intrusion Detected", 'frame': resized_frame} detections_list.append(detection_message) cv2.polylines(resized_frame, [np.array(fixed_roi)], isClosed=True, color=(0, 0, 255), thickness=2) cv2.imshow("Intrusion Detection", resized_frame) key = cv2.waitKey(1) & 0xFF if key == ord('q'): break cap.release() cv2.destroyAllWindows() return detections_list from Nettyfy_visnx.settings import BASE_DIR class IntrusionDetectionApiView(APIView): def post(self, request, *args, **kwargs): try: video_file = request.FILES.get('video_file') start_time = request.data.get('start_time') end_time = request.data.get('end_time') if not video_file: return Response({'message': 'No video file provided'}, status=status.HTTP_400_BAD_REQUEST) path = default_storage.save('tmp/some_video.mp4', ContentFile(video_file.read())) tmp_file = os.path.join(settings.MEDIA_ROOT, path) intrusion_detected = intrusion_detection(tmp_file, start_time, end_time) intrusion_event, _ = CameraEvent.objects.get_or_create(camera_event="IntrusionDetected") for detection in intrusion_detected: # Convert the frame to a format Django can handle (like JPEG) _, buffer = cv2.imencode('.jpg', detection["frame"]) content_file = ContentFile(buffer.tobytes(), name='detection.jpg') # Create a new Alert instance alert = Alert( alert_message = detection["detection"], frame=content_file, camera=Camera.objects.first(), camera_events = intrusion_event ) alert.save() # save camera incharge alert.camera_incharge.add(CustomUser.objects.first()) return Response({'status': True, 'message': 'Intrusion detected in the video'}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'error': str(e)}) def danzer_detection(file_path): model = YOLO('model/yolov8s.pt') file = open(BASE_DIR/'coco.txt', 'r') data = file.read() class_list = data.split('/n') roi_x, roi_y, roi_h, roi_w = 300, 300, 389, 398 cap = cv2.VideoCapture(file_path) fourcc = cv2.VideoWriter_fourcc(*'XVID') out = cv2.VideoWriter('output video/PV_5.avi', fourcc, 20.0, (1000, 700)) frame_count = 0 detections_list = [] while True: ret, frame = cap.read() if not ret: break frame = cv2.resize(frame, (1000, 700)) result = model.predict(frame, conf=0.15) a = result[0].boxes.data px = pd.DataFrame(a).astype("float") cv2.rectangle(frame, (roi_x, roi_y), (roi_x + roi_w, roi_y + roi_h), (255, 0, 0), 2) danger_detect = False for index, row in px.iterrows(): x1, y1, x2, y2, d = map(int, row[:5]) c = class_list[d] if 'person' in c and roi_x < x1 < roi_x + roi_w and roi_y < y1 < roi_y + roi_h: cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 0, 255), 2) cv2.putText(frame, 'Person', (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 2) cv2.putText(frame, 'Danger Zone', (x1, y1 - 20), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) danger_detect = True if danger_detect: detection_message = { "detection": f"Detect in danger zone", "frame": frame } detections_list.append(detection_message) cv2.imshow("RGB", frame) out.write(frame) frame_count += 1 if cv2.waitKey(1) & 0XFF == ord('q'): break cap.release() out.release() cv2.destroyAllWindows() return detections_list class DanzerDetectionApiView(APIView): def post(self, request, *args, **kwargs): try: video_file = request.FILES.get('video_file') if not video_file: return Response({'message': 'No video file provided', 'status': False}) path = default_storage.save('tmp/some_video.mp4', ContentFile(video_file.read())) tmp_file = os.path.join(settings.MEDIA_ROOT, path) detections = danzer_detection(tmp_file) danzer_event, _ = CameraEvent.objects.get_or_create(camera_event="DanzerDetected") for detection in detections: # Convert the frame to a format Django can handle (like JPEG) _, buffer = cv2.imencode('.jpg', detection["frame"]) content_file = ContentFile(buffer.tobytes(), name='detection.jpg') # Create a new Alert instance alert = Alert( alert_message=detection["detection"], frame=content_file, camera=Camera.objects.first(), camera_events=danzer_event ) alert.save() # save camera incharge alert.camera_incharge.add(CustomUser.objects.first()) return Response({'status': True, 'message': 'Danzer detected in the video'}) except Exception as e: return Response({'status': False, 'message': 'Something went wrong', 'error': str(e)}) from datetime import datetime, timezone import random class AlertViewSet(viewsets.ModelViewSet): permission_classes = [AllowAny] serializer_class = AlertSerializer def create(self, request): camera = Camera.objects.first() user = CustomUser.objects.filter(first_name = "Visnx") entry_event, _ = CameraEvent.objects.get_or_create(camera_event='entry') exit_event, _ = CameraEvent.objects.get_or_create(camera_event='exit') today = datetime.now(timezone.utc) one_week_ago = today - timedelta(days=7) date_generated = [one_week_ago + timedelta(days=x) for x in range(0, 7)] start_hour = 10 end_hour = 22 created_alerts = [] for date in date_generated: num_alerts = random.randint(1, 10) for _ in range(num_alerts): random_hour = random.randint(start_hour, end_hour - 1) random_minute = random.randint(0, 59) random_second = random.randint(0, 59) alert_time = date.replace(hour=random_hour, minute=random_minute, second=random_second) event_type = random.choice(['entry', 'exit']) event = entry_event if event_type == 'entry' else exit_event alert_message = "Entry detected" if event_type == 'entry' else "Exit detected" alert = Alert.objects.create( camera=camera, camera_events=event, alert_message=alert_message, created_at=alert_time, is_event_detect=True, ) alert.camera_incharge.add(user.first()) created_alerts.append(alert) serializer = AlertSerializer(created_alerts, many=True, context={'request': request}) return Response({'status': True, 'message': 'Alert Created','data':serializer.data},status = status.HTTP_200_OK) class EntryExitCount(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Alert.objects.all() serializer_class = AlertSerializer pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user organization_id = request.query_params.get('organization_id', 0) location_id = request.query_params.get('location_id', 0) start_date = request.query_params.get('start_date', '') end_date = request.query_params.get('end_date', '') start_time = time(5, 0) end_time = time(17, 0) if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) if start_date: queryset = queryset.filter(created_at__date = datetime.strptime(start_date, "%Y-%m-%d").date()) else: queryset = queryset.filter( created_at__date = datetime.now().date(), is_enabled=True # Assuming you want to filter enabled alerts ) if organization_id: queryset = queryset.filter(camera__organization__id = organization_id) if location_id: queryset = queryset.filter(camera__location__id = location_id) duration_wise_counts = [] # month_wise_counts = defaultdict(lambda: defaultdict(int)) while start_time < end_time: duration = datetime.combine(date.today(), start_time) + timedelta(minutes=30) duration = duration.time() if start_date: duration_alert = queryset.filter( created_at__gte=datetime.combine(datetime.strptime(start_date, "%Y-%m-%d").date(), start_time, tzinfo=timezone.utc), created_at__lte=datetime.combine(datetime.strptime(start_date, "%Y-%m-%d").date(), duration, tzinfo=timezone.utc), is_enabled=True # Assuming you want to filter enabled alerts ) # queryset = queryset.filter(created_at__date = datetime.strptime(start_date, "%Y-%m-%d").date()) else: duration_alert = queryset.filter( created_at__gte=datetime.combine(datetime.now().date(), start_time, tzinfo=timezone.utc), created_at__lte=datetime.combine(datetime.now().date(), duration, tzinfo=timezone.utc), is_enabled=True # Assuming you want to filter enabled alerts ) entry_count = duration_alert.filter(camera__organization__organization_user = user, camera_events__camera_event__icontains = 'entry', is_enabled=True).order_by('-created_at').count() exit_count = duration_alert.filter(camera__organization__organization_user = user, camera_events__camera_event__icontains = 'exit', is_enabled=True).order_by('-created_at').count() # if organization_id: # entry_count = entry_count.filter(organization__id = organization_id, is_enabled = True).count() # exit_count = exit_count.filter(organization__id = organization_id, is_enabled = True).count() duration_wise_counts.append({ "duration": start_time.strftime('%H:%M')+"-"+duration.strftime('%H:%M'), "entry_count": entry_count, "exit_count":exit_count }) # Move to the next day start_time = datetime.combine(date.today(), start_time) + timedelta(minutes=30) start_time = start_time.time() if duration_wise_counts: return Response({'status': True, 'message': 'Entry Exit count Successfull.', 'data': duration_wise_counts}) else: return Response({'status': False, 'message': 'No data found!', 'data': []}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class QueueManagementAPI(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = QueueManagement.objects.all().order_by('-created_at') def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_queuemanagement'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) start_date = request.query_params.get('start_date', '') search_query = request.query_params.get('search', '') organization_id = request.query_params.get('organization_id', 0) location_id = request.query_params.get('location_id', 0) area_id = request.query_params.get('area_id', 0) # camera_id = request.query_params.get('camera_id', None) cameras = Camera.objects.filter(is_enabled = True) if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) queryset = queryset.filter(roi__camera__organization__organization_user = user) # queryset = queryset.filter(created_at__date = datetime.now().date()) if start_date: queryset = queryset.filter(created_at__date = datetime.strptime(start_date, "%Y-%m-%d").date()) if organization_id: cameras = cameras.filter(organization__id = organization_id) if location_id: cameras = cameras.filter(location__id = location_id) if area_id: cameras = cameras.filter(area__id = area_id) if cameras: queryset = queryset.filter(roi__camera__in = cameras).distinct() if search_query: search_query = search_query.strip().lower() queryset = queryset.filter(Q(roi__roi_name__icontains = search_query)| Q(currunt_queue_count__icontains = search_query)| Q(average_waiting_time__icontains = search_query)| Q(recomanded_staff_adjustment__icontains = search_query)| Q(action__icontains = search_query)).order_by('-created_at') if queryset: serializer = QueueManagementSerializer(queryset, many=True, context={'request': request}) # data_lst = [] # for querydata in queryset: # que_len = querydata.currunt_queue_count # wait_time = querydata.average_waiting_time # adjustment = querydata.recomanded_staff_adjustment # action = querydata.action # data = { # # "roi_data" : querydata.roi, # "checkout_point":querydata.roi.roi_name, # "currunt_queue_length":que_len, # "average_waiting_time":wait_time, # "recomanded_staff_adjustment":adjustment, # "action":action # } # data_lst.append(data) return Response({'status': True, 'message': 'Queue data Successfully.', 'data': serializer.data}) else: return Response({'status': False, 'message': 'No data found!', 'data': []}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class HeatMappingAPI(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = HeatMapping.objects.all().order_by('-created_at') serializer_class = HeatMappingSerializer pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_heatmapping'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) search_query = request.query_params.get('search', '') start_date = request.query_params.get('start_date', '') organization_id = request.query_params.get('organization_id', 0) location_id = request.query_params.get('location_id', 0) area_id = request.query_params.get('area_id', 0) # camera_id = request.query_params.get('camera_id', None) cameras = Camera.objects.filter(is_enabled = True) if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) queryset = queryset.filter(roi__camera__organization__organization_user = user) # queryset = queryset.filter(created_at__date = datetime.now().date()) if start_date: queryset = queryset.filter(created_at__date = datetime.strptime(start_date, "%Y-%m-%d").date()) if organization_id: cameras = cameras.filter(organization__id = organization_id) if location_id: cameras = cameras.filter(location__id = location_id) if area_id: cameras = cameras.filter(area__id = area_id) # if cameras: queryset = queryset.filter(roi__camera__in = cameras).distinct() if search_query: search_query = search_query.strip().lower() queryset = queryset.filter(roi__camera__area__area_name__icontains = search_query).order_by('-created_at') if queryset: # print(queryset) serializer = HeatMappingSerializer(queryset, many=True, context={'request': request}) return Response({'status': True, 'message': 'Heat Map list successfully.', 'data': serializer.data}) else: return Response({'status': False, 'message': 'Heat Map list not found!', 'data': []}) # return Response({'status':True, 'message': 'HeatMap data successfull.','data':queryset}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class DemographicDetailsAPI(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = DemographicDetails.objects.all().order_by('-created_at') serializer_class = DemographicDetailsSerializer pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_demographicdetails'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) search_query = request.query_params.get('search', '') start_date = request.query_params.get('start_date', '') organization_id = request.query_params.get('organization_id', 0) location_id = request.query_params.get('location_id', 0) area_id = request.query_params.get('area_id', 0) # camera_id = request.query_params.get('camera_id', None) cameras = Camera.objects.filter(is_enabled = True) if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) queryset = queryset.filter(roi__camera__organization__organization_user = user) # queryset = queryset.filter(created_at__date = datetime.now().date()) if start_date: queryset = queryset.filter(created_at__date = datetime.strptime(start_date, "%Y-%m-%d").date()) if organization_id: cameras = cameras.filter(organization__id = organization_id) if location_id: cameras = cameras.filter(location__id = location_id) if area_id: cameras = cameras.filter(area__id = area_id) if cameras: queryset = queryset.filter(roi__camera__in = cameras).distinct() if search_query: search_query = search_query.strip().lower() queryset = queryset.filter(age_range__icontains = search_query).order_by('-created_at') if queryset: # print(queryset) serializer = DemographicDetailsSerializer(queryset, many=True, context={'request': request}) return Response({'status': True, 'message': 'Demographics details list successfully.', 'data': serializer.data}) else: return Response({'status': False, 'message': 'Demographics details list not found!', 'data': []}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class ParkingAPI(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = Parking.objects.all().order_by('-created_at') serializer_class = ParkingSerializer pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm('myprofile.view_parking'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) start_date = request.query_params.get('start_date', '') organization_id = request.query_params.get('organization_id', 0) location_id = request.query_params.get('location_id', 0) area_id = request.query_params.get('area_id', 0) # camera_id = request.query_params.get('camera_id', None) cameras = Camera.objects.filter(is_enabled = True) if not user: return Response({'status': False, 'message': 'You can not access roi list', 'data': []}) queryset = queryset.filter(roi__camera__organization__organization_user = user) # queryset = queryset.filter(created_at__date = datetime.now().date()) if start_date: queryset = queryset.filter(created_at__date =start_date) if organization_id: cameras = cameras.filter(organization__id = organization_id) if location_id: cameras = cameras.filter(location__id = location_id) if area_id: cameras = cameras.filter(area__id = area_id) if cameras: queryset = queryset.filter(roi__camera__in = cameras).distinct() if queryset: # print(queryset) serializer = ParkingSerializer(queryset, many=True, context={'request': request}) return Response({'status': True, 'message': 'Parking details list successfully.', 'data': serializer.data}) else: return Response({'status': False, 'message': 'Parking details list not found!', 'data': []}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) # class EmployeePerson(viewsets.ModelViewSet): # permission_classes = [IsAuthenticated] # queryset = Person.objects.all() # serializer_class = PersonSerializer # pagination_class = StandardResultsSetPagination # def list(self, request, *args, **kwargs): # queryset = self.filter_queryset(self.get_queryset()) # try: # user = self.request.user # except Exception as e: # return Response({'status':False, 'message': 'Something went wrong','data':str(e)}) class AlertImageUploadView(APIView): def post(self, request, *args, **kwargs): serializer = ImageUploadSerializer(data=request.data) if serializer.is_valid(): image = serializer.validated_data['image'] # Assuming image is an InMemoryUploadedFile object save_path = str(settings.MEDIA_ROOT) + '/alert_images/' + image.name directory = os.path.dirname(save_path) if not os.path.exists(directory): os.makedirs(directory) with open(save_path, 'wb+') as destination: for chunk in image.chunks(): destination.write(chunk) return Response({"message": "Image uploaded successfully", "path": save_path}, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class DashboardAPI(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] def list(self, request, *args, **kwargs): try: user = self.request.user if not user.has_perm('myprofile.view_dashboardmanufacturing') or not user.has_perm('myprofile.view_dashboardretail'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) start_date = request.query_params.get('start_date', '') end_date = request.query_params.get('end_date', '') if start_date and end_date: start_date = datetime.strptime(start_date, '%Y-%m-%d').date() end_date = datetime.strptime(end_date, '%Y-%m-%d').date() if start_date > end_date: return Response({'status': False, 'message': "Start date should be less then end date."}) alert_count = Alert.objects.filter(created_at__date__gte = start_date, created_at__date__lte = end_date).count() else: alert_count = Alert.objects.filter(organization__organization_user = user, is_enabled=True).count() event_count = CameraEvent.objects.filter(organization__organization_user = user, is_enabled=True).count() camera_count = Camera.objects.filter(organization__organization_user = user).count() location_count = Location.objects.filter(organization__organization_user = user).count() count_data = { "alert_count": alert_count, "event_count": event_count, "camera_count": camera_count, "location_count": location_count } alert_obj = Alert.objects.filter(organization__organization_user = user, is_enabled=True).order_by('-created_at') if start_date and end_date: start_date_graph = start_date end_date_graph = end_date else: current_date = datetime.now().date() start_date_graph = current_date.replace(day=1) next_month = start_date_graph.replace(day=28) + timedelta(days=4) end_date_graph = (next_month - timedelta(days=next_month.day)) date_wise_counts = [] if end_date_graph - timedelta(days=6) == start_date_graph: current_date = start_date_graph while current_date <= end_date_graph: total_count = Alert.objects.filter(created_at__date=current_date).count() date_wise_counts.append({ "date": current_date.strftime('%a'), "total": total_count }) current_date += timedelta(days=1) elif end_date_graph - timedelta(days=31) > start_date_graph: current_date = start_date_graph.replace(day=1) while current_date <= end_date_graph: next_month = current_date.replace(day=28) + timedelta(days=4) last_day_of_month = next_month - timedelta(days=next_month.day) total_count = alert_obj.filter( created_at__year=current_date.year, created_at__month=current_date.month ).count() date_wise_counts.append({ "date": current_date.strftime('%b'), "total": total_count }) current_date = last_day_of_month + timedelta(days=1) else: current_date = start_date_graph while current_date <= end_date_graph: total_count = alert_obj.filter(created_at__date=current_date).count() date_wise_counts.append({ "date": current_date.strftime('%Y-%m-%d'), "total": total_count }) current_date += timedelta(days=1) event_queryset = CameraEvent.objects.filter(organization__organization_user = user, is_enabled=True) if start_date and end_date: if start_date == end_date: alert_obj = alert_obj.filter(created_at__date = start_date) else: alert_obj = alert_obj.filter(created_at__date__gte = start_date, created_at__date__lte = end_date) event_percentage_data = {} for event in event_queryset: event_percentage_data[event.camera_event] = round((alert_obj.filter(camera_events__camera_event = event.camera_event).count()/alert_obj.count() if alert_obj else 0)*100) if alert_obj: if start_date and end_date: top_alert_obj = alert_obj.filter(camera__organization__organization_user = user, created_at__date__gte = start_date, created_at__date__lte = end_date).order_by('-created_at') else: top_alert_obj = alert_obj top_alerts = top_alert_obj.filter(camera__organization__organization_user = user, is_enabled=True).order_by('-created_at')[:4] top_alert = AlertSerializer(top_alerts,many=True,context={'request': request}) top_alert_data = top_alert.data else: top_alert_data = [] camera = Camera.objects.filter(organization__organization_user = user) camera_data = CameraSerializer(camera,many=True) data = { 'alert_event_camera_location_count':count_data, 'date_wise_counts':date_wise_counts, 'undetected_event_percentage':event_percentage_data, 'top_four_alerts':top_alert_data, 'camera_list':camera_data.data, } return Response({'status': True, 'message': 'Dashboard data get successfully.', 'data': data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class AlertForPPEApiView(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] queryset = AlertPersonTracking.objects.all().order_by('-created_at') serializer_class = AlertPersonTrackingSerializer pagination_class = StandardResultsSetPagination def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) try: user = self.request.user if not user.has_perm("myprofile.view_alertpersontracking"): return Response({'status': False, 'message': "You don't have permission to perform this action"}) search_query = request.query_params.get('search', '') start_date = request.query_params.get('start_date', '') end_date = request.query_params.get('end_date', '') organization_id = request.query_params.get('organization_id', '') location_id = request.query_params.get('location_id', '') area_id = request.query_params.get('area_id', '') camera_id = request.query_params.get('camera_id', '') event_id = request.query_params.get('camera_event_id','') alert_message = request.query_params.get('alert_message','') person_id = request.query_params.get('person_id','') ppe_alert_obj = queryset # if queryset: # ppe_alert_obj = queryset.filter(camera__organization__organization_user__email = user).order_by('-created_at') if not ppe_alert_obj: return Response({'status': False, 'message': "No data found.", "data":[]}) if start_date and end_date: if start_date > end_date: return Response({'status': False, 'message': "Start date should be less then end date"}) date_format = '%Y-%m-%d' # start_date,end_date, error_message = validate_dates(start_date, end_date) # if error_message: # return Response({'status': False, 'message': error_message,"data":[]}) start_date = datetime.strptime(start_date, date_format).date() end_date = datetime.strptime(end_date, date_format).date() if start_date == end_date: ppe_alert_obj = ppe_alert_obj.filter(created_at__date=start_date).order_by('-created_at') else: ppe_alert_obj = ppe_alert_obj.filter(created_at__date__gte=start_date, created_at__date__lte=end_date).order_by('-created_at') elif start_date: date_format = '%Y-%m-%d' start_date = datetime.strptime(start_date, date_format).date() if isinstance(start_date, date): ppe_alert_obj = ppe_alert_obj.filter(created_at__date=start_date).order_by('-created_at') else: return Response({"status": False, "message": "Enter correct format of start_date","data":str(e)}) if person_id: ppe_alert_obj = ppe_alert_obj.filter(tracking_id = person_id) # import ipdb;ipdb.set_trace() if organization_id: ppe_alert_obj = ppe_alert_obj.filter(alerts__organization__id = organization_id).distinct() if location_id: ppe_alert_obj = ppe_alert_obj.filter(alerts__location__id = location_id).distinct() if area_id: ppe_alert_obj = ppe_alert_obj.filter(alerts__area__id = area_id).distinct() if camera_id: ppe_alert_obj = ppe_alert_obj.filter(alerts__camera__id = camera_id).distinct() if event_id: ppe_alert_obj = ppe_alert_obj.filter(alerts__camera_events__id = event_id).distinct() if alert_message: ppe_alert_obj = ppe_alert_obj.filter(alerts__alert_message__icontains = alert_message).distinct() if not ppe_alert_obj: return Response({'status': False,'message': "No data found!","data":[]}) if search_query: search_query = search_query.strip().lower() ppe_alert_obj = ppe_alert_obj.filter(Q(alerts__alert_message__icontains = search_query)| Q(alerts__location__loc_name__icontains = search_query) | Q(alerts__camera_events__camera_event__icontains = search_query) | Q(alerts__camera__camera_name__icontains = search_query) | Q(alerts__organization__org_name__icontains = search_query) | Q(created_at__icontains = search_query)).order_by('-created_at') if ppe_alert_obj: page = self.paginate_queryset(ppe_alert_obj) response_data = [] if page is not None: for data in page: response_obj = {'id':'', 'tracking_id':'', 'alerts':[]} serializer = AlertSerializer(data.alerts.all(), many=True, context={'request': request}) alert_data = serializer.data response_obj['id'] = data.id response_obj['tracking_id'] = data.tracking_id response_obj['alerts'] = alert_data response_data.append(response_obj) else: for data in ppe_alert_obj: response_obj = {'id':'', 'tracking_id':'', 'alerts':[]} serializer = AlertSerializer(data.alerts.all(), many=True, context={'request': request}) alert_data = serializer.data response_obj['id'] = data.id response_obj['tracking_id'] = data.tracking_id response_obj['alerts'] = alert_data response_data.append(response_obj) if response_data: # serializer = AlertPersonTrackingSerializer(ppe_alert_obj, many=True) count = ppe_alert_obj.count() limit = int(request.GET.get('page_size', 10)) return Response({"status": True, "message":"Alert List Successfully", 'total_page': (count + limit - 1) // limit, 'count':count, 'data': response_data}) else: return Response({"status": True, "message":"No data found.",'data':[]}) else: return Response({"status": True, "message":"No Alert Found","data":[]}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class IdBasedAlertsApi(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] lookup_field = 'alert_id' def list(self, request, *args, **kwargs): try: user = self.request.user alert_id = self.kwargs.get('alert_id') alert_obj = Alert.objects.filter(id = alert_id).first() if not alert_obj: return Response({'status': False, 'message': 'Alert not found!', 'data': []}) alert_serialized_obj = AlertSerializer(alert_obj,context={'request': request}) return Response({'status': True, 'message': 'Alert data successfully.', 'data': alert_serialized_obj.data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) class IdBasedPPEAlertsApi(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] lookup_field = 'alert_id' def list(self, request, *args, **kwargs): try: user = self.request.user alert_id = self.kwargs.get('alert_id') alert_obj = AlertPersonTracking.objects.filter(id = alert_id).first() if not alert_obj: return Response({'status': False, 'message': 'Alert not found!', 'data': []}) alert_serialized_obj = AlertPersonTrackingSerializer(alert_obj,context={'request': request}) return Response({'status': True, 'message': 'Alert data successfully.', 'data': alert_serialized_obj.data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) import tempfile class CameraIDForROI(viewsets.ModelViewSet): permission_classes = [IsAuthenticated] lookup_field = 'camera_id' def list(self, request, *args, **kwargs): try: user = self.request.user camera_id = self.kwargs.get('camera_id') camera_obj = Camera.objects.filter(id = camera_id).first() if not camera_obj: return Response({'status': False, 'message': 'Camera not found!', 'data': []}) cap = cv2.VideoCapture(camera_obj.rtsp_url) ret, frame = cap.read() cap.release() if not ret: return Response({'status':False, 'message': 'Camera Not woring.'}) height, width = frame.shape[:2] rois = Roi.objects.filter(camera__id = camera_id) roi_data = RoiSerializer(rois, many= True, context={'request': request}) # temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.jpg', dir=settings.MEDIA_ROOT+'/rtsp_frame_images') file_name = 'rtsp_temp_image.jpg' file_path = os.path.join(settings.MEDIA_ROOT, 'rtsp_frame_images', file_name) cv2.imwrite(file_path, frame) relative_path = os.path.relpath(file_path, settings.MEDIA_ROOT+'/rtsp_frame_images') image_url = os.path.join(settings.URL+settings.MEDIA_URL+'rtsp_frame_images', relative_path) data = { 'height': 700, 'width': 1200, 'image': image_url, 'roi_data':roi_data.data } return Response({'status': True, 'message': 'RTSP Image successfully.', 'data': data}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)}) import json class DeleteLogApi(viewsets.ModelViewSet): def list(self, request, *args, **kwargs): try: user = self.request.user start_date = request.GET.get('start_date') end_date = request.GET.get('end_date') status_code = request.GET.get('status_code') status_str = request.GET.get('status') if not user.has_perm('admin.view_logentry'): return Response({'status': False, 'message': "You don't have permission to perform this action"}) # import ipdb;ipdb.set_trace() log_dir = os.path.join(os.path.dirname(__file__), 'logs') delete_log_file = os.path.join(log_dir, 'api_logs/delete_log.json') if os.path.exists(delete_log_file): with open(delete_log_file, 'r') as f: try: delete_logs = json.load(f) except json.JSONDecodeError: delete_logs = [] else: delete_logs = [] if start_date and end_date: try: # Parse the date strings into datetime objects start_date = datetime.strptime(start_date, '%Y-%m-%d') end_date = datetime.strptime(end_date, '%Y-%m-%d') # Filter the logs by the date range delete_logs = [ log for log in delete_logs if start_date <= datetime.strptime(log['date'][:10], '%Y-%m-%d') <= end_date ] except ValueError: return Response({'status': False, 'message': 'Invalid date format. Use YYYY-MM-DD for both start_date and end_date.'}) filterd_list = [] if status_code: try: status_code = int(status_code) # Ensure it's an integer for log in delete_logs: if log['status_code'] == status_code: filterd_list.append(log) delete_logs = filterd_list except ValueError: return Response({'status': False, 'message': 'Invalid status code. It must be a number.'}) filterd_list = [] if status_str: for log in delete_logs: if log['status'] == status_str: filterd_list.append(log) delete_logs = filterd_list delete_logs = sorted(delete_logs,key=lambda x: datetime.strptime(x['date'], '%Y-%m-%d %H:%M:%S.%f'),reverse=True) # Pagination logic page = int(request.GET.get('page', 1)) # Default to page 1 limit = int(request.GET.get('page_size', 10)) # Default to 10 items per page count = len(delete_logs) # Total number of log entries # Calculate start and end indexes for pagination start = (page - 1) * limit end = start + limit paginated_logs = delete_logs[start:end] if paginated_logs: return Response({ 'status': True, 'message': 'Delete Log data successfully.', 'total_page': (count + limit - 1) // limit, 'count':count, 'data': paginated_logs }) else: return Response({'status': False, 'message': 'No Log Founds.', 'data': []}) except Exception as e: return Response({'status': False,'message': 'Something went wrong','data':str(e)})