import base64 from datetime import timedelta from re import search import os from django.conf import settings from django.http import FileResponse from django.db import transaction from django.utils.timezone import now from rest_framework.views import APIView from rest_framework import status from django.http import JsonResponse from rest_framework.response import Response from django.core.paginator import Paginator from django.core.mail import EmailMessage from . import generate_final_data, populate_errors, CustomExcemption from rest_framework import exceptions from . import models as dec_models from . import serializer as dec_serializer from . import filters as dec_filters from . import constants as dec_constants from po_app.models import EmailLog from . import utils as dec_utils def get_declaration(declaration_id): """function to declaration from id""" try: declaration = dec_models.EUDeclaration.objects.get(id=declaration_id) except dec_models.EUDeclaration.DoesNotExist: raise exceptions.NotFound(detail=f"EU Declaration with ID {declaration_id} not found.") except Exception as e: raise exceptions.ValidationError(detail=f"An error occurred: {str(e)}") return declaration class CompanyView(APIView): @transaction.atomic def post(self, request): """ API to create or update Company details. """ data = request.data company_id = data.get('company_id') if company_id: # Update existing Company try: company = dec_models.Company.objects.get(id=company_id) serializer = dec_serializer.CompanySerializer( instance=company, data=data, partial=True) except dec_models.Company.DoesNotExist: raise exceptions.NotFound(detail="Company with the given ID not found.") else: # Create a new company serializer = dec_serializer.CompanySerializer(data=data) # Validate serializer if not serializer.is_valid(): return Response( { 'status_code': status.HTTP_404_NOT_FOUND, 'status': "error", 'message': populate_errors(serializer.errors), 'data': None, }, status=status.HTTP_404_NOT_FOUND ) # Save the company serializer.save() response = generate_final_data('DATA_SAVED') response['result'] = serializer.data # Return the saved data return Response( { 'status_code': status.HTTP_200_OK, 'status': "success", 'message': 'Company saved successfully', 'data': None, }, status=status.HTTP_200_OK ) def get(self, request): """API to get infra by department""" company_id = request.GET.get('company_id', None) page_size = request.GET.get('page_size', 20) page_number = request.GET.get('page_no', 1) if company_id: try: company = dec_models.Company.objects.get(id=company_id, is_deleted=False) except dec_models.Company.DoesNotExist: raise exceptions.NotFound(detail="Company with the given ID not found.") serializer = dec_serializer.CompanySerializer(instance=company) result = serializer.data else: search_key = request.GET.get('search', None) filter_data = { 'search_key': search_key, 'sort_column': request.GET.get('sort_column', None), 'sort_order': request.GET.get('sort_order', None), 'status': request.GET.get('status', None), } companies = dec_filters.company_filter(filter_data) # Apply pagination paginator = Paginator(companies, page_size) page_obj = paginator.get_page(page_number) serializer = dec_serializer.CompanySerializer(page_obj, many=True) result = { "total_records": paginator.count, "total_pages": paginator.num_pages, "current_page": page_obj.number, "companies": serializer.data } response_data = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data fetched successfully", 'data': result} return JsonResponse(response_data, status=status.HTTP_200_OK) @transaction.atomic def delete(self, request): """Function to delete Company """ company_id = request.GET.get('company_id') logged_user_id = request.login_user_id # attached from middleware try: company = dec_models.Company.objects.get(id=company_id) company.is_deleted=True company.deleted_by=logged_user_id company.save() response = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Company Deleted Successfully", 'data': "Company Deleted Successfully"} except: response = { "status": "error", "status_code": status.HTTP_301_MOVED_PERMANENTLY, "message": "Something went wrong", 'data': "Something went wrong"} return JsonResponse(response, status=status.HTTP_200_OK) class EUDeclarationView(APIView): """ View to create or update EU declarations """ @transaction.atomic def post(self, request): """ API to create or update EU declarations. """ data = request.data declaration_id = data.get('declaration_id') if declaration_id: # Update existing declaration try: declaration = dec_models.EUDeclaration.objects.get(id=declaration_id) declaration.status = dec_constants.ACTIVE declaration.save() serializer = dec_serializer.EUDeclarationSerializer( instance=declaration, data=data, partial=True, context={'request': request}) except dec_models.EUDeclaration.DoesNotExist: raise exceptions.NotFound(detail="declaration with the given ID not found.") else: # Create a new declaration serializer = dec_serializer.EUDeclarationSerializer(data=data) # Validate serializer if not serializer.is_valid(): return Response( { 'status_code': status.HTTP_404_NOT_FOUND, 'status': "error", 'message': populate_errors(serializer.errors), 'data': None, }, status=status.HTTP_404_NOT_FOUND ) # Save the declaration if serializer.is_valid(): serializer.save() response = generate_final_data('DATA_SAVED') response['result'] = serializer.data # Return the saved data return Response( { 'status_code': status.HTTP_200_OK, 'status': "success", 'message': 'declarations saved successfully', 'data': serializer.data, }, status=status.HTTP_200_OK ) def get(self, request): """API to get infra by department""" declaration_id = request.GET.get('declaration_id', None) company_id = request.GET.get('company_id', None) start_date = request.GET.get('start_date', None) end_date = request.GET.get('end_date', None) page_size = request.GET.get('page_size', 20) page_number = request.GET.get('page_number', 1) if declaration_id: try: declarations = dec_models.EUDeclaration.objects.get(id=declaration_id) except dec_models.EUDeclaration.DoesNotExist: raise exceptions.NotFound(detail="Company with the given ID not found.") declaration_data = dec_serializer.EUDeclarationSerializer(instance=declarations).data else: search_key = request.GET.get('search', None) filter_data = { 'search_key': search_key, 'company_id': company_id, 'start_date': start_date, 'end_date': end_date } declarations = dec_filters.declaration_filter(filter_data) serializer = dec_serializer.EUDeclarationSerializer(declarations, many=True) # Apply pagination paginator = Paginator(declarations, page_size) page_obj = paginator.get_page(page_number) declaration_data = { "total_records": paginator.count, "total_pages": paginator.num_pages, "current_page": page_obj.number, "result": serializer.data } result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data fetched successfully", 'data': declaration_data} return JsonResponse(result, status=status.HTTP_200_OK) @transaction.atomic def delete(self, request): """Function to delete Company """ declaration_id = request.GET.get('declaration_id') logged_user_id = request.login_user_id # attached from middleware try: declaration = dec_models.EUDeclaration.objects.get(id=declaration_id) declaration.is_deleted = True declaration.deleted_by = logged_user_id declaration.save() response = { "status": "success", "status_code": status.HTTP_200_OK, "message": "EUD Deleted Successfully", "data": "EUD Deleted Successfully" } except dec_models.EUDeclaration.DoesNotExist: response = { "status": "error", "status_code": status.HTTP_404_NOT_FOUND, "message": "EUD not found", "data": None } except Exception as e: response = { "status": "error", "status_code": status.HTTP_500_INTERNAL_SERVER_ERROR, "message": str(e), "data": str(e) } return JsonResponse(response, status=status.HTTP_200_OK) class GetEUDRenewalDataView(APIView): def get(self, request): """function to get EUD renewal data""" declaration_id = request.GET.get('declaration_id', None) if not declaration_id: raise exceptions.ValidationError( { 'message': f'declaration_id required', 'status_code': status.HTTP_400_BAD_REQUEST, 'status': "error", 'data': None, }) try: declaration = dec_models.EUDeclaration.objects.get(id=declaration_id) except dec_models.EUDeclaration.DoesNotExist: raise exceptions.NotFound(detail="declaration with the given ID not found.") customer = declaration.company.company_name to, cc = dec_utils.eud_recipient_mails(declaration_id) logged_user_mail = request.login_user_email cc.append(logged_user_mail) subject = f"{customer} - Customer Statement regarding precursor use" body = "Your Customer Statement regarding precursor use is about to expire. Please fill out the document attached and sent us back, filled in and signed." response_data = { "to": to, "cc": cc, "subject": subject, "body": body } result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data fetched successfully", 'data': response_data} return JsonResponse(result, status=status.HTTP_200_OK) class ServeEUDDocument(APIView): def get(self, request): # Construct the file path file_name = 'CustomerPrecursorUseStatement.pdf' file_path = os.path.join(settings.BASE_DIR, 'media', 'pdfs', file_name) # Check if the file exists if not os.path.exists(file_path): return Response({ "status_code": 404, "status": "error", "message": "File not found" }, status=status.HTTP_404_NOT_FOUND) try: # Serve the file as a response response = FileResponse(open(file_path, 'rb'), content_type='application/pdf') response['Content-Disposition'] = f'inline; filename="{file_name}"' return response except Exception as e: return Response({ "status_code": 500, "status": "error", "message": f"Error serving file: {str(e)}" }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class SendEUDEmailAPIView(APIView): """Api send email""" def post(self, request): # Extract data from request subject = request.data.get("subject", "No Subject") body = request.data.get("body", "") # Can be HTML content sender_email = os.getenv('EMAIL_HOST_USER') recipient_email = request.data.get("recipient_email", []) cc_email = request.data.get("cc_email", []) attachments = request.data.get("attachments", []) # List of base64 encoded PDFs initiated_by = request.data.get("initiated_by", "") declaration_id = request.data.get("declaration_id", "") email_type = request.data.get("type", None) company_id = request.data.get("company_id", None) file_description = request.data.get("file_description", None) logged_user_id = request.login_user_id # attached from middleware if not type: return Response({ "status_code": 400, "status": "error", "message": f"type required " }, status=status.HTTP_400_BAD_REQUEST) # Validation checks if email_type=='eud': required_fields = [subject, body, recipient_email, declaration_id] else: required_fields = [subject, body, recipient_email, company_id] missing_fields = [f for f in required_fields if f in [None, '']] if missing_fields: return Response({ "status_code": 400, "status": "error", "message": f"The following fields are required: {', '.join(missing_fields)}" }, status=status.HTTP_400_BAD_REQUEST) if email_type=='eud': declaration = get_declaration(declaration_id) recipient_id = declaration_id else: recipient_id = None if request.is_test_user: recipient_email = ["thowfeek@tycaninc.com", "ashmi.ma@tycaninc.com", "ashik.os@tycaninc.com"] cc_email = [] # cc_email = [] # Create EmailMessage object email = EmailMessage(subject, body, sender_email, recipient_email, cc=cc_email) email.content_subtype = "html" # Set content type to HTML attachment_names = [] # Handle attachments (Base64 encoded PDFs) for attachment in attachments: file_name = attachment.get("name", "document.pdf") file_data = attachment.get("data") # Base64 encoded content if file_data: decoded_file = base64.b64decode(file_data) if email_type=='shared' and company_id: file = dec_utils.save_base64_file(file_data, file_name=file_name) description = file_description if file_description else file_name shared_doc= dec_models.SharedDocument.objects.create( company_id=company_id, file_path=file, description=description, file_name=file_name, shared_by=logged_user_id) recipient_id = shared_doc.id email.attach(file_name, decoded_file, "application/pdf") attachment_names.append(file_name) if email_type=='eud': file_name = 'Customer’s precursor use statement.docx' file_path = os.path.join(settings.BASE_DIR, 'media', 'pdfs', file_name) if os.path.exists(file_path): email.attach_file(file_path) else: print(f"⚠️ File not found: {file_path}") # Send email and check response email_sent_count = email.send() email_log = EmailLog.objects.create( sender=sender_email, recipient=", ".join(recipient_email), cc=", ".join(cc_email), subject=subject, body=body, sent_at=now(), status="success" if email_sent_count > 0 else "failed", triggered_by=initiated_by, type=email_type, receipient_id=recipient_id, attachment=attachment_names, email_response=email_sent_count ) if email_type=='eud': dec_utils.update_contact_history(recipient_email, cc_email, declaration) dec_utils.update_company_contact_history( recipient_email, cc_email, declaration.company.id) declaration.status = dec_constants.REQUEST_SEND declaration.save() history = declaration.update_declaration_history(request_user_id=logged_user_id) elif email_type=='shared': dec_utils.update_company_contact_history( recipient_email, cc_email, company_id) result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Email sent successfully", 'data': None} return JsonResponse(result, status=status.HTTP_200_OK) class NotifyAdminEUDView(APIView): """class to notify admin about upcoming eud""" def get(self, request): company = dec_models.Company.objects.filter( latest_expiry__lte=now() + timedelta(days=14), latest_expiry__gte=now(), is_deleted=False).order_by('latest_expiry') if company.count()==1: first_company = company.first() message = f"EUDs of {first_company.company_name} expires on {first_company.latest_expiry.strftime('%d/%m/%Y')}." expiring = True elif company.count() > 1: first_company = company.first().company_name message = f"EUDs of {first_company} and {company.count()-1} others will expires soon." expiring = True else: message = "No EUDs are expiring in the next 14 days." expiring = False data = {'message': message, 'expiring': expiring} result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Email sent successfully", 'data': data} return JsonResponse(result, status=status.HTTP_200_OK) class CompanyDocumentView(APIView): def get(self, request): """API to get company details""" company_id = request.GET.get('company_id', None) page_size = request.GET.get('page_size', 20) page_number = request.GET.get('page_no', 1) if company_id: try: company = dec_models.Company.objects.get(id=company_id, is_deleted=False) except dec_models.Company.DoesNotExist: raise exceptions.NotFound(detail="Company with the given ID not found.") serializer = dec_serializer.BaseCompanySerializer(instance=company) result = serializer.data else: search_key = request.GET.get('search', None) filter_data = { 'search_key': search_key, 'sort_column': request.GET.get('sort_column', None), 'sort_order': request.GET.get('sort_order', None), 'status': request.GET.get('status', None), } companies = dec_filters.shared_company_filter(filter_data) # Apply pagination paginator = Paginator(companies, page_size) page_obj = paginator.get_page(page_number) serializer = dec_serializer.BaseCompanySerializer(page_obj, many=True) result = { "total_records": paginator.count, "total_pages": paginator.num_pages, "current_page": page_obj.number, "companies": serializer.data } response_data = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data fetched successfully", 'data': result} return JsonResponse(response_data, status=status.HTTP_200_OK) class SharedDocumentView(APIView): """ View to get list of shared documents """ def get(self, request): """API to get documents shared to Customer""" company_id = request.GET.get('company_id', None) start_date = request.GET.get('start_date', None) end_date = request.GET.get('end_date', None) page_size = request.GET.get('page_size', 20) page_number = request.GET.get('page_number', 1) search_key = request.GET.get('search', None) filter_data = { 'search_key': search_key, 'company_id': company_id, 'start_date': start_date, 'end_date': end_date } documents = dec_filters.shared_document_filter(filter_data) # Apply pagination paginator = Paginator(documents, page_size) page_obj = paginator.get_page(page_number) serializer = dec_serializer.SharedDocumentSerializer( page_obj.object_list, many=True ) document_data = { "total_records": paginator.count, "total_pages": paginator.num_pages, "current_page": page_obj.number, "result": serializer.data } result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data fetched successfully", 'data': document_data} return JsonResponse(result, status=status.HTTP_200_OK) def post(self, request): """API to get documents shared to Customer""" company_id = request.data.get('company_id', None) file = request.data.get('file', None) shared_doc= dec_models.SharedDocument.objects.create(company_id=company_id, file_path=file ) return JsonResponse({'file': 'updated successfully'}, status=status.HTTP_200_OK) class GetShareDocumentContentView(APIView): def get(self, request): """function to get EUD renewal data""" company_id = request.GET.get('company_id', None) if not company_id: raise exceptions.ValidationError( { 'message': f'company_id required', 'status_code': status.HTTP_400_BAD_REQUEST, 'status': "error", 'data': None, }) to, cc = dec_utils.get_company_recipient_mails(company_id) logged_user_mail = request.login_user_email cc.append(logged_user_mail) cc = list(set(cc)) subject = f"Company documents for your reference" response_data = { "to": to, "cc": cc, "subject": subject, } result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data fetched successfully", 'data': response_data} return JsonResponse(result, status=status.HTTP_200_OK) class SavedDocumentView(APIView): """ View to get list of Saved documents """ def get(self, request): """API to get documents shared to Customer""" company_id = request.GET.get('company_id', None) start_date = request.GET.get('start_date', None) end_date = request.GET.get('end_date', None) page_size = request.GET.get('page_size', 20) page_number = request.GET.get('page_number', 1) search_key = request.GET.get('search', None) filter_data = { 'search_key': search_key, 'company_id': company_id, 'start_date': start_date, 'end_date': end_date } documents = dec_filters.saved_document_filter(filter_data) # Apply pagination paginator = Paginator(documents, page_size) page_obj = paginator.get_page(page_number) serializer = dec_serializer.SavedDocumentSerializer( page_obj.object_list, many=True ) document_data = { "total_records": paginator.count, "total_pages": paginator.num_pages, "current_page": page_obj.number, "result": serializer.data } result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data fetched successfully", 'data': document_data} return JsonResponse(result, status=status.HTTP_200_OK) def post(self, request): """API to save documnets related to customer""" company_id = request.data.get('company_id', None) file = request.data.get('file', None) logged_user_id = request.login_user_id # attached from middleware required_fields = { "company_id": company_id, "file": file, } for field_name, field_value in required_fields.items(): if not field_value: raise exceptions.ValidationError( { "message": f"{field_name} is required", "status_code": status.HTTP_400_BAD_REQUEST, "status": "error", "data": None, } ) file_name = file.name shared_doc= dec_models.SavedDocument.objects.create( company_id=company_id, file_path=file, created_by=logged_user_id, file_name=file_name) result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data Saved Successfully", 'data': 'saved'} return JsonResponse(result, status=status.HTTP_200_OK) def delete(self, request): """API to delete documnets related to customer""" document_id = request.GET.get('document_id', None) if not document_id: raise exceptions.ValidationError( { 'message': f'document_id required', 'status_code': status.HTTP_400_BAD_REQUEST, 'status': "error", 'data': None, }) shared_doc= dec_models.SavedDocument.objects.get( id=document_id) shared_doc.is_deleted = True shared_doc.save() result = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Data deleted successfully", 'data': 'saved'} return JsonResponse(result, status=status.HTTP_200_OK)