from datetime import datetime from io import BytesIO from typing import Iterable, List, OrderedDict import sentry_sdk import stripe import xlsxwriter from django.conf import settings from django.forms import model_to_dict from django.http import HttpResponse from django.shortcuts import redirect from django.utils.decorators import method_decorator from drf_yasg.utils import swagger_auto_schema from forex_python.converter import CurrencyRates from rest_framework import status from rest_framework.decorators import action from rest_framework.mixins import ( CreateModelMixin, ListModelMixin, RetrieveModelMixin, UpdateModelMixin, ) from rest_framework.response import Response from rest_framework.viewsets import GenericViewSet from verify_trusted.companies.api.filters.payment import PaymentFilter from verify_trusted.companies.api.params import ( company_id_param, company_name_param, currency_param, date_after_param, date_before_param, status_param, ) from verify_trusted.companies.api.serializers import ( PaymentCompanySerializer, PaymentExportSerializer, PaymentListSerializer, PaymentSerializer, PaymentUpdateSerializer, ) from verify_trusted.companies.models import Payment from verify_trusted.users.api.paginations import PaymentResultsSetPagination from verify_trusted.users.api.params import user_param from verify_trusted.utils.swagger_schemas import NoDjangoFilerBackendFilterInspector from verify_trusted.companies.models import Company payment_filter_params = [ company_name_param, company_id_param, status_param, currency_param, user_param, date_before_param, date_after_param, ] stripe.api_key = settings.STRIPE_API_KEY price_usd = settings.PRICE_USD_ID price_gb = settings.PRICE_GBP_ID @method_decorator( name='list', decorator=swagger_auto_schema( manual_parameters=payment_filter_params, filter_inspectors=[NoDjangoFilerBackendFilterInspector], ), ) class CompanyPaymentViewSet( CreateModelMixin, RetrieveModelMixin, UpdateModelMixin, ListModelMixin, GenericViewSet, ): queryset = ( Payment.objects.prefetch_related('plan', 'company', 'company__user') .order_by('-create_date', '-modify_date') .all() ) serializer_class = PaymentSerializer pagination_class = PaymentResultsSetPagination filterset_class = PaymentFilter def get_serializer_class(self): if self.action == 'retrieve': return PaymentCompanySerializer if self.action == 'list': return PaymentListSerializer if self.action in ['update', 'partial_update']: return PaymentUpdateSerializer return super().get_serializer_class() def get_queryset(self): queryset = super().get_queryset() if self.action == 'list': is_superuser = self.request.user.is_superuser if not is_superuser: queryset = self.queryset.filter(company__user_id=self.request.user.id) return queryset def truncate(self, num, n): integer = int(num * (10 ** n)) / (10 ** n) return float(integer) def create(self, request, *args, **kwargs): try: serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) plan = serializer.validated_data['plan'] self.perform_create(serializer) payment = Payment.objects.get(id=serializer.data['id']) company = Company.objects.get(id=payment.company.pk) if payment.stripe_customer_id is None: # if payment.company.currency_symbol != 'GB': # customer = stripe.Customer.create(email=self.request.user.email, tax_exempt='exempt') # else: customer = stripe.Customer.create(email=self.request.user.email, tax_exempt='exempt') customer_id = customer['id'] else: customer_id = payment.stripe_customer_id # pm = stripe.PaymentMethod() total_amount = int(plan.price * 100) tax_amount = 0 # if ( # payment.company.currency_symbol is not None # and payment.company.currency_symbol.upper() != 'USD' # ): # currency = payment.company.currency_symbol # else: # currency = plan.currency if company.currency_symbol != 'USD': company.currency_symbol = 'USD' company.save() currency = plan.currency country_code = payment.company.country_code map = payment.company.map price_id = price_usd if map is not None or country_code is not None: # if country_code == 'GB': # tax_amount = total_amount * settings.STRIPE_UK_TAX / 100 # total_amount += int(tax_amount) # price_id = price_gb # el if payment.company.map is not None: for data in payment.company.map['address_components']: if 'country' in data['types'] and data['short_name'] == 'GB': # tax_amount = total_amount * settings.STRIPE_UK_TAX / 100 # total_amount += int(tax_amount) # price_id = price_gb break subscription = stripe.Subscription.create( customer=customer_id, items=[{ 'price': price_id, }], payment_behavior='default_incomplete', payment_settings={'save_default_payment_method': 'on_subscription'}, expand=['latest_invoice.payment_intent'], ) truncate_amount = self.truncate(total_amount / 100, 2) result_rate = truncate_amount try: if currency != 'USD': c = CurrencyRates() result_rate = c.convert(currency, 'USD', truncate_amount, date_obj=datetime.now()) payment.price_to_usd = result_rate except Exception as e: print(e) payment.price_to_usd = None payment.total_amount = truncate_amount payment.vat = self.truncate(tax_amount / 100, 2) payment.stripe_id = subscription.id payment.status = Payment.Status.PENDING payment.stripe_status = subscription.status payment.currency = subscription.currency payment.stripe_customer_id = customer_id # payment.stripe_payment_method_id = subscription.payment_method payment.save() response_data = model_to_dict(payment) response_data['client_secret'] = subscription.latest_invoice.payment_intent.client_secret response_data['id'] = payment.id response_data['customer_id'] = customer_id headers = self.get_success_headers(serializer.data) # noqa return Response( response_data, status=status.HTTP_201_CREATED, headers=headers ) except Exception as e: sentry_sdk.capture_exception(e) return Response({'error': str(e)}) @swagger_auto_schema(paginator_inspectors=[]) @action(detail=False, methods=['GET'], url_path='export-excel') def export_excel(self, request, *args, **kwargs): payments: Iterable[PaymentExportSerializer] = self.filter_queryset( self.get_queryset() ) data: List[OrderedDict] = PaymentExportSerializer(payments, many=True).data # create our spreadsheet. I will create it in memory with a StringIO output = BytesIO() workbook = xlsxwriter.Workbook(output) worksheet = workbook.add_worksheet() fields = PaymentExportSerializer().fields for i, field in enumerate(fields): worksheet.write(0, i, field) for r, history in enumerate(data, start=1): for c, field in enumerate(fields): worksheet.write(r, c, str(history.get(field))) workbook.close() # create a response response = HttpResponse(content_type='application/vnd.ms-excel') # tell the browser what the file is named filename = 'Payment Histories {}'.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