from datetime import datetime 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.http import HttpResponse from django.utils.decorators import method_decorator from drf_yasg import openapi from drf_yasg.utils import swagger_auto_schema from rest_framework.decorators import action from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from rest_framework.viewsets import ReadOnlyModelViewSet from verify_trusted.companies.api.filters.changelog import ( ChangelogFilter, ReviewChangelogFilter, ) from verify_trusted.companies.api.params import date_after_param, date_before_param, company_name_param, status_param from verify_trusted.companies.api.serializers.changelog import ( CompanyHistorySerializer, ReviewHistorySerializer, ) from verify_trusted.companies.history_types import HistoryType, HistoryTypeSymbol from verify_trusted.companies.models import Company, Review from verify_trusted.users.api.paginations import ( CompanyHistoryResultsSetPagination, ReviewHistoryResultsSetPagination, ) from verify_trusted.users.api.params import ( author_param, company_param, name_param, user_param, email_config_param, ) from verify_trusted.utils.swagger_schemas import NoDjangoFilerBackendFilterInspector User = get_user_model() history_type_param = openapi.Parameter( 'type', openapi.IN_QUERY, description='Changelog type', type=openapi.TYPE_STRING, enum=[ HistoryType.CREATED, HistoryType.CHANGED, HistoryType.CLAIMED, ], ) history_search_params = [ name_param, user_param, history_type_param, date_before_param, date_after_param, ] review_history_search_params = [ company_param, author_param, history_type_param, date_before_param, date_after_param, ] export_params = [ company_name_param, date_before_param, date_after_param, ] @method_decorator( name='list', decorator=swagger_auto_schema( manual_parameters=history_search_params, filter_inspectors=[NoDjangoFilerBackendFilterInspector], ), ) class CompanyHistoryViewSet(ReadOnlyModelViewSet): serializer_class = CompanyHistorySerializer permission_classes = (IsAuthenticated,) pagination_class = CompanyHistoryResultsSetPagination filterset_class = ChangelogFilter lookup_field = 'history_id' def get_queryset(self): types = [HistoryTypeSymbol.CREATED, HistoryTypeSymbol.CHANGED, HistoryTypeSymbol.DELETED] queryset = ( Company.history.prefetch_related('user', 'history_user') .filter(history_type__in=types) .order_by('-history_date') ) user = self.request.user if not user.is_superuser: queryset = queryset.filter( id__in=user.companies.values_list('id', flat=True) ) return queryset def retrieve(self, request, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance) # https://django-simple-history.readthedocs.io/en/latest/querying_history.html#getting-previous-and-next-historical-record prev_record = instance.prev_record response = { 'selected_record': serializer.data, 'previous_record': self.get_serializer(prev_record).data if prev_record is not None else None, } if instance.logo and len(instance.logo) > 0: response['selected_record']['logo'] = 'https://%s/media/%s' % ( settings.AWS_S3_CUSTOM_DOMAIN, instance.logo, ) if prev_record is not None and prev_record.logo and len(prev_record.logo) > 0: response['previous_record']['logo'] = 'https://%s/media/%s' % ( settings.AWS_S3_CUSTOM_DOMAIN, prev_record.logo, ) # https://django-simple-history.readthedocs.io/en/latest/history_diffing.html if prev_record is not None: delta = instance.diff_against(prev_record) diff = { change.field: { 'old': change.old.url if len(change.old.name) > 0 else None, 'new': change.new.url, } if change.field == 'image' or change.field == 'logo' else {'old': change.old, 'new': change.new} for change in delta.changes } else: diff = None response['diff'] = diff return Response(response) @swagger_auto_schema(manual_parameters=export_params) @action(detail=False, methods=['GET'], url_path='export-excel') def export_excel(self, request: Request, *args, **kwargs): histories: Iterable[Company] = self.filter_queryset(self.get_queryset()) data: List[OrderedDict] = CompanyHistorySerializer(histories, many=True).data headers = [ 'Id', 'History Id', 'History Date', 'Action', 'History User', 'User', 'Domain', 'Company Name', 'Phone', 'Email', 'Logo', 'Description', 'Address', 'Is Verified', ] # 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, history in enumerate(data, start=1): worksheet.write_row( r, 0, [ history['id'], str(history['history_id']), history['history_date'], history['history_type'], str(history['history_user']), str(history['user']), history['url'], history['name'], history['phone'], history['email'], history['logo'], history['about'], history['address'], history['is_verified'], ], ) 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()) # return the response return response @method_decorator( name='list', decorator=swagger_auto_schema( manual_parameters=review_history_search_params, filter_inspectors=[NoDjangoFilerBackendFilterInspector], ), ) class ReviewHistoryViewSet(ReadOnlyModelViewSet): serializer_class = ReviewHistorySerializer permission_classes = (IsAuthenticated,) pagination_class = ReviewHistoryResultsSetPagination filterset_class = ReviewChangelogFilter lookup_field = 'history_id' def get_queryset(self): types = [ HistoryTypeSymbol.CREATED, HistoryTypeSymbol.CHANGED, HistoryTypeSymbol.DELETED, ] queryset = ( Review.history.prefetch_related('source', 'history_user') .filter(history_type__in=types) .order_by('-history_date') ) user = self.request.user if not user.is_superuser and user.companies is not None: queryset = queryset.filter( id__in=user.companies.values_list('id', flat=True) ) return queryset def retrieve(self, request, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance) # https://django-simple-history.readthedocs.io/en/latest/querying_history.html#getting-previous-and-next-historical-record prev_record = instance.prev_record response = { 'selected_record': serializer.data, 'previous_record': self.get_serializer(prev_record).data if prev_record is not None else None, } # https://django-simple-history.readthedocs.io/en/latest/history_diffing.html if prev_record is not None: delta = instance.diff_against(prev_record) diff = { change.field: {'old': change.old, 'new': change.new} for change in delta.changes } else: diff = None response['diff'] = diff return Response(response)