from datetime import datetime, timedelta from distutils.util import strtobool from io import BytesIO from typing import Iterable, List, OrderedDict import xlsxwriter from django.conf import settings from django.contrib.auth import get_user_model from django.core.mail import EmailMultiAlternatives from django.db.models import BooleanField, Case, Count, F, Q, Value, When from django.http import HttpResponse from django.utils import timezone from django.utils.decorators import method_decorator from django_rest_passwordreset.models import ( ResetPasswordToken, clear_expired, get_password_reset_lookup_field, get_password_reset_token_expiry_time, ) from django_rest_passwordreset.signals import reset_password_token_created from django_rest_passwordreset.views import ( HTTP_IP_ADDRESS_HEADER, HTTP_USER_AGENT_HEADER, _unicode_ci_compare, ) from drf_yasg.utils import swagger_auto_schema from rest_framework.decorators import action from rest_framework.permissions import IsAdminUser from rest_framework.response import Response from rest_framework.viewsets import ModelViewSet from config import celery_app from config.settings.base import DEFAULT_FROM_EMAIL from verify_trusted.companies.api.serializers import ( CompanyExportExcelSerializer, CompanyUpdateSerializer, ) from verify_trusted.companies.api.viewsets import ( BaseCompanyHistoryViewSet, BaseCompanyListViewSet, BaseCompanyRetrieveMixin, BaseCompanyViewSet, ) from verify_trusted.companies.export import ExportUserIntoExcelTask from verify_trusted.companies.models import Company, Subscription from verify_trusted.reviews.models import ReviewSource from verify_trusted.users.api.paginations import UserResultsSetPagination from verify_trusted.users.api.params import ( assign_param, company_param, end_date_param, start_date_param, user_param, username_param, ) from verify_trusted.users.api.serializers import ( AdminUpdateProfileSerializer, AdminUpdateUserSerializer, CreateUserSerializer, UserSerializer, ) from verify_trusted.widgets.models import Seal, Widget from verify_trusted.companies.tasks import export_company_task User = get_user_model() search_params = [ user_param, username_param, company_param, assign_param, ] @method_decorator( name='list', decorator=swagger_auto_schema(manual_parameters=search_params), ) class AdminUserViewSet(ModelViewSet): serializer_class = UserSerializer queryset = User.objects.prefetch_related('profile').all() permission_classes = [IsAdminUser] pagination_class = UserResultsSetPagination def get_queryset(self): queryset = super().get_queryset() if self.action == 'list': query_params = self.request.query_params user_filter = query_params.get(user_param.name, None) if user_filter: user_query = Q(id=user_filter) queryset = queryset.filter(user_query) company_filter = query_params.get(company_param.name, None) assign_filter = bool( strtobool(query_params.get(assign_param.name, 'false')) ) name_filter = query_params.get(username_param.name, None) if company_filter and assign_filter is True: if name_filter: filter_ = ( Q(companies__isnull=True) & ( Q(username__icontains=name_filter) | Q(name__icontains=name_filter) | Q(email__icontains=name_filter) ) ) | Q(companies__id=company_filter) else: filter_ = Q(companies__isnull=True) | Q( companies__id=company_filter ) queryset = ( queryset.filter(filter_) .annotate( is_owner=Case( When( companies__user_id=F('id'), companies__id=company_filter, then=Value(True), ), default=Value(False), output_field=BooleanField(), ) ) .order_by('-is_owner', '-id') ) return queryset if name_filter and not company_filter and not assign_filter: queryset = queryset.filter( Q(username__icontains=name_filter) | Q(email__icontains=name_filter) ) if assign_filter is True and not company_filter: if name_filter: filter_ = Q(companies__isnull=True) & ( Q(username__icontains=name_filter) | Q(name__icontains=name_filter) | Q(email__icontains=name_filter) ) else: filter_ = Q(companies__isnull=True) queryset = queryset.filter(filter_).order_by('-id') return queryset queryset = queryset.distinct() return queryset.order_by('-id') return queryset.order_by('-id') @swagger_auto_schema( method='GET', manual_parameters=[start_date_param, end_date_param], ) @action(methods=['GET'], detail=False, url_path='dashboard') def admin_dashboard(self, request): query_params = self.request.query_params start_date_filter = query_params.get(start_date_param.name, None) + ' 00:00:00' end_date_filter = query_params.get(end_date_param.name, None) + ' 23:59:59' start_date_filter = datetime.strptime(start_date_filter, '%Y-%m-%d %H:%M:%S') end_date_filter = ( datetime.strptime(end_date_filter, '%Y-%m-%d %H:%M:%S') if end_date_filter is not None else datetime.now() ) count_user_non_paying = Count( 'company_id', filter=Q( due_date__lt=end_date_filter.date(), due_date__gte=start_date_filter.date(), ), ) count_user_trial = Count( 'company_id', filter=Q( payment__isnull=True, due_date__gte=end_date_filter.date(), create_date__range=[start_date_filter, end_date_filter], ), ) count_user_paid = Count( 'company_id', filter=Q( payment__isnull=False, due_date__gte=end_date_filter.date(), create_date__range=[start_date_filter, end_date_filter], ), ) count_user_signed_up_set = Count( 'id', filter=Q(date_joined__range=[start_date_filter, end_date_filter]), ) # data = Subscription.objects.filter( # create_date__range=[start_date_filter, end_date_filter] # ) data = Subscription.objects.all() data = data.aggregate( count_user_non_paying=count_user_non_paying, count_user_trial=count_user_trial, count_user_paid=count_user_paid, ) user = User.objects.aggregate( count_user_signed_up_set=count_user_signed_up_set, count_total_user=Count('id'), ) response_data = { 'count_user_non_paying': data['count_user_non_paying'], 'count_user_trial': data['count_user_trial'], 'count_user_paid': data['count_user_paid'], 'count_user_signed_up_set': user['count_user_signed_up_set'], 'count_total_user': user['count_total_user'], } return Response(response_data) def get_serializer_class(self): if self.action == 'create': return CreateUserSerializer if self.action in ['update', 'partial_update']: return AdminUpdateUserSerializer if self.action == 'update_profile': return AdminUpdateProfileSerializer return super().get_serializer_class() def perform_destroy(self, instance): Company.objects.filter(user=instance).update(is_verified=False, user=None) super().perform_destroy(instance) def create(self, request, *args, **kwargs): create_user = super().create(request, *args, **kwargs) user = User.objects.filter( **{ '{}__iexact'.format( get_password_reset_lookup_field() ): create_user.data['email'] } ).first() password_reset_token_validation_time = get_password_reset_token_expiry_time() now_minus_expiry_time = timezone.now() - timedelta( hours=password_reset_token_validation_time ) clear_expired(now_minus_expiry_time) if user.eligible_for_reset() and _unicode_ci_compare( user.email, getattr(user, get_password_reset_lookup_field()) ): if user.password_reset_tokens.all().count() > 0: token = user.password_reset_tokens.all()[0] else: token = ResetPasswordToken.objects.create( user=user, user_agent=request.META.get(HTTP_USER_AGENT_HEADER, ''), ip_address=request.META.get(HTTP_IP_ADDRESS_HEADER, ''), ) reset_password_token_created.send( sender=self.__class__, instance=self, reset_password_token=token ) return create_user class AdminCompanyViewSet( ModelViewSet, BaseCompanyHistoryViewSet, BaseCompanyListViewSet, BaseCompanyRetrieveMixin, BaseCompanyViewSet, ): permission_classes = [IsAdminUser] def get_serializer_class(self): if self.action == 'update': return CompanyUpdateSerializer return super().get_serializer_class() @action(detail=True, methods=['PATCH']) def remove_user(self, request, *args, **kwargs): instance = self.get_object() instance.user = None instance.is_verified = False instance.save() return Response({"success": "success"}) def perform_destroy(self, instance): Widget.objects.filter(company_id=instance.id).delete() Seal.objects.filter(company_id=instance.id).delete() ReviewSource.objects.filter(company_id=instance.id).delete() Company.objects.filter(id=instance.id).delete() # instance.delete() # @celery_app.task() # def export_company_task(self, request, *args, **kwargs): # company: Iterable[CompanyExportExcelSerializer] = self.filter_queryset( # self.get_queryset() # ) # data: List[OrderedDict] = CompanyExportExcelSerializer(company, many=True).data # # headers = ['Id', 'Company Name', 'Email', 'Profile URL', 'Domain', 'Address'] # # # create our spreadsheet. I will create it in memory with a StringIO # output = BytesIO() # workbook = xlsxwriter.Workbook(output) # worksheet = workbook.add_worksheet() # for i, field in enumerate(headers): # worksheet.write(0, i, field) # for r, company in enumerate(data, start=1): # worksheet.write_row( # r, # 0, # [ # company['id'], # company['name'], # company['email'], # company['url'], # company['url_display'], # company['address'], # ], # ) # workbook.close() # # # create a response # # response = HttpResponse(content_type='application/vnd.ms-excel') # # # tell the browser what the file is named # filename = 'Company Changelogs {}'.format(str(datetime.today())) # # response['Content-Disposition'] = 'attachment;filename="{}.xlsx"'.format( # # filename # # ) # # # put the spreadsheet data into the response # # response.write(output.getvalue()) # if settings.IS_PRODUCTION: # send_to_email = ['info@verifytrusted.com'] # else: # send_to_email = settings.DEFAULT_SEND_TO_EMAIL # msg = EmailMultiAlternatives( # # title: # 'Active user on {title}'.format(title='Verify Trusted'), # # message: # 'Active User', # # from: # DEFAULT_FROM_EMAIL, # # to: # send_to_email, # bcc=['verifytrusted.support@sphinxjsc.com'], # ) # msg.attach('{}.xlsx'.format(filename), output.getvalue(), 'application/vnd.ms-excel') # msg.send() # # return the response # # return response # return Response(status=200) @swagger_auto_schema(paginator_inspectors=[]) @action(detail=False, methods=['GET'], url_path='export-excel') def export_excel(self, request, *args, **kwargs): query_params = self.request.query_params task = export_company_task.s(query_params) task.apply_async() # export = ExportUserIntoExcelTask() # export.export(query_params) return Response(status=200)