mirror of
https://github.com/HamzaSha1/zod-backend.git
synced 2025-08-26 06:09:41 +00:00
Merge pull request #218 from KiwiTechLLC/ZBKADM-72
csv/excel export api
This commit is contained in:
dilipshrivastwa-kiwi
@ -1,13 +1,18 @@
|
||||
"""
|
||||
web_admin pagination file
|
||||
"""
|
||||
# third party imports
|
||||
from rest_framework.pagination import PageNumberPagination
|
||||
|
||||
from base.constants import NUMBER
|
||||
|
||||
|
||||
class CustomPageNumberPagination(PageNumberPagination):
|
||||
"""
|
||||
custom paginator class
|
||||
"""
|
||||
page_size = 10 # Set the desired page size
|
||||
# Set the desired page size
|
||||
page_size = NUMBER['ten']
|
||||
page_size_query_param = 'page_size'
|
||||
max_page_size = 100 # Set a maximum page size if needed
|
||||
# Set a maximum page size if needed
|
||||
max_page_size = NUMBER['hundred']
|
||||
|
||||
@ -1,12 +1,24 @@
|
||||
"""
|
||||
web_admin analytics serializer file
|
||||
"""
|
||||
# third party imports
|
||||
from rest_framework import serializers
|
||||
|
||||
# django imports
|
||||
from django.contrib.auth import get_user_model
|
||||
|
||||
# local imports
|
||||
from base.constants import USER_TYPE
|
||||
|
||||
from junior.models import JuniorPoints, Junior
|
||||
|
||||
USER = get_user_model()
|
||||
|
||||
|
||||
class JuniorLeaderboardSerializer(serializers.ModelSerializer):
|
||||
"""
|
||||
junior leaderboard serializer
|
||||
"""
|
||||
name = serializers.SerializerMethodField()
|
||||
first_name = serializers.SerializerMethodField()
|
||||
last_name = serializers.SerializerMethodField()
|
||||
@ -44,9 +56,80 @@ class JuniorLeaderboardSerializer(serializers.ModelSerializer):
|
||||
|
||||
|
||||
class LeaderboardSerializer(serializers.ModelSerializer):
|
||||
"""
|
||||
leaderboard serializer
|
||||
"""
|
||||
junior = JuniorLeaderboardSerializer()
|
||||
rank = serializers.IntegerField()
|
||||
|
||||
class Meta:
|
||||
"""
|
||||
meta class
|
||||
"""
|
||||
model = JuniorPoints
|
||||
fields = ('total_points', 'rank', 'junior')
|
||||
|
||||
|
||||
class UserCSVReportSerializer(serializers.ModelSerializer):
|
||||
"""
|
||||
user csv/xls report serializer
|
||||
"""
|
||||
phone_number = serializers.SerializerMethodField()
|
||||
user_type = serializers.SerializerMethodField()
|
||||
is_active = serializers.SerializerMethodField()
|
||||
date_joined = serializers.SerializerMethodField()
|
||||
|
||||
class Meta:
|
||||
"""
|
||||
meta class
|
||||
"""
|
||||
model = USER
|
||||
fields = ('first_name', 'last_name', 'email', 'phone_number', 'user_type', 'is_active', 'date_joined')
|
||||
|
||||
@staticmethod
|
||||
def get_phone_number(obj):
|
||||
"""
|
||||
:param obj: user object
|
||||
:return: user phone number
|
||||
"""
|
||||
if profile := obj.guardian_profile.all().first():
|
||||
return f"+{profile.country_code}{profile.phone}" \
|
||||
if profile.country_code and profile.phone else profile.phone
|
||||
elif profile := obj.junior_profile.all().first():
|
||||
return f"+{profile.country_code}{profile.phone}" \
|
||||
if profile.country_code and profile.phone else profile.phone
|
||||
else:
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_user_type(obj):
|
||||
"""
|
||||
:param obj: user object
|
||||
:return: user type
|
||||
"""
|
||||
if obj.guardian_profile.all().first():
|
||||
return dict(USER_TYPE).get('2').capitalize()
|
||||
elif obj.junior_profile.all().first():
|
||||
return dict(USER_TYPE).get('1').capitalize()
|
||||
else:
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_is_active(obj):
|
||||
"""
|
||||
:param obj: user object
|
||||
:return: user type
|
||||
"""
|
||||
if profile := obj.guardian_profile.all().first():
|
||||
return "Active" if profile.is_active else "Inactive"
|
||||
elif profile := obj.junior_profile.all().first():
|
||||
return "Active" if profile.is_active else "Inactive"
|
||||
|
||||
@staticmethod
|
||||
def get_date_joined(obj):
|
||||
"""
|
||||
:param obj: user obj
|
||||
:return: formatted date
|
||||
"""
|
||||
date = obj.date_joined.strftime("%d %b %Y")
|
||||
return date
|
||||
|
||||
@ -3,6 +3,10 @@ web_admin analytics view file
|
||||
"""
|
||||
# python imports
|
||||
import datetime
|
||||
import csv
|
||||
import io
|
||||
import pandas as pd
|
||||
import xlsxwriter
|
||||
|
||||
# third party imports
|
||||
from rest_framework.viewsets import GenericViewSet
|
||||
@ -16,15 +20,17 @@ from django.db.models import Count
|
||||
from django.db.models.functions import TruncDate
|
||||
from django.db.models import F, Window
|
||||
from django.db.models.functions.window import Rank
|
||||
from django.http import HttpResponse
|
||||
|
||||
# local imports
|
||||
from account.utils import custom_response
|
||||
from base.constants import PENDING, IN_PROGRESS, REJECTED, REQUESTED, COMPLETED, EXPIRED, DATE_FORMAT
|
||||
from base.constants import PENDING, IN_PROGRESS, REJECTED, REQUESTED, COMPLETED, EXPIRED, DATE_FORMAT, TASK_STATUS
|
||||
from guardian.models import JuniorTask
|
||||
from junior.models import JuniorPoints
|
||||
from web_admin.pagination import CustomPageNumberPagination
|
||||
from web_admin.permission import AdminPermission
|
||||
from web_admin.serializers.analytics_serializer import LeaderboardSerializer
|
||||
from web_admin.serializers.analytics_serializer import LeaderboardSerializer, UserCSVReportSerializer
|
||||
from web_admin.serializers.user_management_serializer import UserManagementListSerializer
|
||||
|
||||
USER = get_user_model()
|
||||
|
||||
@ -141,3 +147,103 @@ class AnalyticsViewSet(GenericViewSet):
|
||||
paginated_queryset = paginator.paginate_queryset(queryset, request)
|
||||
serializer = self.serializer_class(paginated_queryset, many=True)
|
||||
return custom_response(None, serializer.data)
|
||||
|
||||
@action(methods=['get'], url_name='export-excel', url_path='export-excel', detail=False)
|
||||
def export_excel(self, request):
|
||||
"""
|
||||
to export users count, task details and top juniors in csv/excel file
|
||||
:param request: start_date: date format (yyyy-mm-dd)
|
||||
:param request: end_date: date format (yyyy-mm-dd)
|
||||
:return:
|
||||
"""
|
||||
|
||||
response = HttpResponse(content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
|
||||
response['Content-Disposition'] = 'attachment; filename="ZOD_Bank_Analytics.xlsx"'
|
||||
|
||||
end_date = datetime.date.today()
|
||||
start_date = end_date - datetime.timedelta(days=6)
|
||||
|
||||
if request.query_params.get('start_date') and request.query_params.get('end_date'):
|
||||
start_date = datetime.datetime.strptime(request.query_params.get('start_date'), DATE_FORMAT)
|
||||
end_date = datetime.datetime.strptime(request.query_params.get('end_date'), DATE_FORMAT)
|
||||
|
||||
buffer = io.BytesIO() # Use BytesIO for binary data
|
||||
|
||||
# Create an XlsxWriter Workbook object
|
||||
workbook = xlsxwriter.Workbook(buffer)
|
||||
|
||||
# Add sheets
|
||||
sheets = ['Users', 'Assign Tasks', 'Juniors leaderboard']
|
||||
|
||||
for sheet_name in sheets:
|
||||
worksheet = workbook.add_worksheet(name=sheet_name)
|
||||
|
||||
# sheet 1 for Total Users
|
||||
if sheet_name == 'Users':
|
||||
user_qs = self.get_queryset()
|
||||
queryset = user_qs.filter(date_joined__range=(start_date, (end_date + datetime.timedelta(days=1))))
|
||||
serializer = UserCSVReportSerializer(queryset, many=True)
|
||||
|
||||
df_users = pd.DataFrame([
|
||||
{'Name': f"{user['first_name']} {user['last_name']}",
|
||||
'Email': user['email'], 'Phone Number': user['phone_number'],
|
||||
'User Type': user['user_type'], 'Status': user['is_active'],
|
||||
'Date Joined': user['date_joined']}
|
||||
for user in serializer.data
|
||||
])
|
||||
for idx, col in enumerate(df_users.columns):
|
||||
worksheet.write(0, idx, col) # Write header
|
||||
for row_num, row in enumerate(df_users.values, start=1):
|
||||
for col_num, value in enumerate(row):
|
||||
worksheet.write(row_num, col_num, value)
|
||||
|
||||
# sheet 2 for Assign Task
|
||||
elif sheet_name == 'Assign Tasks':
|
||||
assign_tasks = JuniorTask.objects.filter(
|
||||
created_at__range=[start_date, (end_date + datetime.timedelta(days=1))]
|
||||
).exclude(task_status__in=[PENDING, EXPIRED])
|
||||
|
||||
df_tasks = pd.DataFrame([
|
||||
{'Task Name': task.task_name, 'Task Status': dict(TASK_STATUS).get(task.task_status).capitalize()}
|
||||
for task in assign_tasks
|
||||
])
|
||||
|
||||
for idx, col in enumerate(df_tasks.columns):
|
||||
worksheet.write(0, idx, col) # Write header
|
||||
for row_num, row in enumerate(df_tasks.values, start=1):
|
||||
for col_num, value in enumerate(row):
|
||||
worksheet.write(row_num, col_num, value)
|
||||
|
||||
# sheet 3 for Juniors Leaderboard and rank
|
||||
elif sheet_name == 'Juniors leaderboard':
|
||||
queryset = JuniorPoints.objects.prefetch_related('junior', 'junior__auth').annotate(rank=Window(
|
||||
expression=Rank(),
|
||||
order_by=[F('total_points').desc(), 'junior__created_at']
|
||||
)).order_by('-total_points', 'junior__created_at')
|
||||
|
||||
df_leaderboard = pd.DataFrame([
|
||||
{
|
||||
'Junior Name': f"{junior.junior.auth.first_name} {junior.junior.auth.last_name}"
|
||||
if junior.junior.auth.last_name else junior.junior.auth.first_name,
|
||||
'Points': junior.total_points,
|
||||
'Rank': junior.rank
|
||||
}
|
||||
for junior in queryset
|
||||
])
|
||||
|
||||
for idx, col in enumerate(df_leaderboard.columns):
|
||||
worksheet.write(0, idx, col) # Write header
|
||||
for row_num, row in enumerate(df_leaderboard.values, start=1):
|
||||
for col_num, value in enumerate(row):
|
||||
worksheet.write(row_num, col_num, value)
|
||||
|
||||
# Close the workbook to save the content
|
||||
workbook.close()
|
||||
|
||||
# Reset the buffer position and write the content to the response
|
||||
buffer.seek(0)
|
||||
response.write(buffer.getvalue())
|
||||
buffer.close()
|
||||
|
||||
return response
|
||||
|
||||
|
||||
@ -72,9 +72,9 @@ class UserManagementViewSet(GenericViewSet, mixins.ListModelMixin,
|
||||
return custom_error_response(ERROR_CODE['2067'], status.HTTP_400_BAD_REQUEST)
|
||||
queryset = self.queryset
|
||||
if self.request.query_params.get('user_type') == dict(USER_TYPE).get('2'):
|
||||
queryset = queryset.filter(guardian_profile__user__id=kwargs['pk'])
|
||||
queryset = queryset.filter(id=kwargs['pk'])
|
||||
elif self.request.query_params.get('user_type') == dict(USER_TYPE).get('1'):
|
||||
queryset = queryset.filter(junior_profile__auth__id=kwargs['pk'])
|
||||
queryset = queryset.filter(id=kwargs['pk'])
|
||||
serializer = UserManagementDetailSerializer(queryset, many=True)
|
||||
return custom_response(None, data=serializer.data)
|
||||
|
||||
@ -89,12 +89,12 @@ class UserManagementViewSet(GenericViewSet, mixins.ListModelMixin,
|
||||
return custom_error_response(ERROR_CODE['2067'], status.HTTP_400_BAD_REQUEST)
|
||||
queryset = self.queryset
|
||||
if self.request.query_params.get('user_type') == dict(USER_TYPE).get('2'):
|
||||
user_obj = queryset.filter(guardian_profile__user__id=kwargs['pk']).first()
|
||||
user_obj = queryset.filter(id=kwargs['pk']).first()
|
||||
serializer = GuardianSerializer(user_obj.guardian_profile.all().first(),
|
||||
request.data, context={'user_id': kwargs['pk']})
|
||||
|
||||
elif self.request.query_params.get('user_type') == dict(USER_TYPE).get('1'):
|
||||
user_obj = queryset.filter(junior_profile__auth__id=kwargs['pk']).first()
|
||||
user_obj = queryset.filter(id=kwargs['pk']).first()
|
||||
serializer = JuniorSerializer(user_obj.junior_profile.all().first(),
|
||||
request.data, context={'user_id': kwargs['pk']})
|
||||
|
||||
@ -114,12 +114,12 @@ class UserManagementViewSet(GenericViewSet, mixins.ListModelMixin,
|
||||
return custom_error_response(ERROR_CODE['2067'], status.HTTP_400_BAD_REQUEST)
|
||||
queryset = self.queryset
|
||||
if self.request.query_params.get('user_type') == dict(USER_TYPE).get('2'):
|
||||
user_obj = queryset.filter(guardian_profile__user__id=kwargs['pk']).first()
|
||||
user_obj = queryset.filter(id=kwargs['pk']).first()
|
||||
obj = user_obj.guardian_profile.all().first()
|
||||
obj.is_active = False if obj.is_active else True
|
||||
obj.save()
|
||||
elif self.request.query_params.get('user_type') == dict(USER_TYPE).get('1'):
|
||||
user_obj = queryset.filter(junior_profile__auth__id=kwargs['pk']).first()
|
||||
user_obj = queryset.filter(id=kwargs['pk']).first()
|
||||
obj = user_obj.junior_profile.all().first()
|
||||
obj.is_active = False if obj.is_active else True
|
||||
obj.save()
|
||||
|
||||
Reference in New Issue
Block a user