from rest_framework import viewsets from rest_framework.exceptions import ValidationError as DRFValidationError from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from .models import AddressMaster,AddressCodeRelationship from .serializers import AddressMasterSerializer,CustomerNameSerializer, CustomerMasterSerializer from rest_framework import generics from django.db import transaction from django.db import connection from django.db.models import Q, F,Exists, OuterRef,BooleanField, Value, Case, When, IntegerField from django.forms.models import model_to_dict from django.conf import settings import requests from . import serializers as customer_serializer from . import models as customer_models from . import services as customer_services from django.http import JsonResponse from django.core.paginator import Paginator from rest_framework import exceptions class CustomerListView(APIView): """ A POST API to fetch all customers or a specific customer based on the given customer_id. If customer_id is provided, fetch that customer; otherwise, fetch all customers. """ def post(self, request): # Fetch the customer_id from query parameters customer_id = request.data.get('customer_id') address_id = request.data.get('address_id') exact_address_id = request.data.get('exact_address_id') customer_name = request.data.get('customer_name') search_query = request.data.get('search') search_string = request.data.get('search_string') exact_id = request.data.get('exact_id') try: if address_id: address = AddressMaster.objects.filter(Id=address_id).first() serializer = AddressMasterSerializer(address) address_data = serializer.data elif exact_address_id: address = AddressMaster.objects.filter(ExactERPAddressId=exact_address_id).first() serializer = AddressMasterSerializer(address) address_data = serializer.data elif search_query: # Assuming 'search_query' is the value you want to pass, like 'TradeHaven' search_query_like = f"%{search_query}%" if search_query else None # Create the SQL query with direct parameter injection query = f""" SELECT a.Id AS AddressId, a.CustomerMasterId AS CustomerId, a.CustomerName, a.Type AS AddressType, a.CustomerNickName, a.AddressLine1, a.AddressLine2, a.City, a.Country, a.ExactERPAddressId FROM AddressMaster a WHERE a.IsDeleted = 0 AND ( '{search_query}' IS NULL OR '{search_query}' = '' OR ( a.CustomerName LIKE '{search_query_like}' OR a.Type LIKE '{search_query_like}' OR a.CustomerNickName LIKE '{search_query_like}' OR a.AddressLine1 LIKE '{search_query_like}' OR a.AddressLine2 LIKE '{search_query_like}' OR a.City LIKE '{search_query_like}' OR a.Country LIKE '{search_query_like}' ) ); """ # Execute the query directly with connection.cursor() as cursor: cursor.execute(query) result = cursor.fetchall() address_data = [] for row in result: address_data.append({ 'AddressId': row[0], 'AddressType': row[3], 'CustomerId': row[1], 'CustomerName': row[2], 'CustomerNickName': row[4], 'AddressLine1': row[5], 'AddressLine2': row[6], 'City': row[7], 'Country': row[8], 'ExactERPAddressId': row[9] }) elif search_string: search_query_like = f"%{search_string}%" if search_string else None query = f""" SELECT a.ExactERPAddressId FROM AddressMaster a WHERE a.IsDeleted = 0 AND ( '{search_string}' IS NULL OR '{search_string}' = '' OR ( a.CustomerName LIKE '{search_query_like}' OR a.Type LIKE '{search_query_like}' OR a.CustomerNickName LIKE '{search_query_like}' OR a.AddressLine1 LIKE '{search_query_like}' OR a.AddressLine2 LIKE '{search_query_like}' OR a.City LIKE '{search_query_like}' OR a.Country LIKE '{search_query_like}' ) ); """ # Execute the query directly with connection.cursor() as cursor: cursor.execute(query) address_data = cursor.fetchall() elif exact_id is not None: address_data = AddressMaster.objects.filter(ExactERPAddressId=exact_id).values('Id').first() elif customer_name is not None: address_data = AddressMaster.objects.filter(CustomerName=customer_name).values( 'Id', 'CustomerName', 'CustomerNickName') else: with connection.cursor() as cursor: cursor.execute(""" SELECT a.Id AS AddressId, a.CustomerMasterId AS CustomerId, a.CustomerName, a.CustomerNickName, a.AddressLine1, a.AddressLine2, a.City, a.Country, a.ExactERPAddressId FROM AddressMaster a WHERE a.IsDeleted = 0 """) address_data = [] for row in cursor: address_data.append({ 'AddressId': row[0], 'CustomerId': row[1], 'CustomerName': row[2], 'CustomerNickName': row[3], 'AddressLine1': row[4], 'AddressLine2': row[5], 'City': row[6], 'Country': row[7], 'ExactERPAddressId': row[8], 'SearchQuery':row[2]+" "+row[3] }) response_data = { "status_code": 200, "status": "success", "message": "Customer details fetched successfully", "data": address_data } return Response(response_data, status=status.HTTP_200_OK) except Exception as e: return Response( { "status_code": 500, "status": "error", "message": f"An error occurred: {str(e)}", "data": None }, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) class AddressMasterSearchView(APIView): def post(self, request): search_query = request.data.get('search_query', '').strip() if not search_query: return Response({"error": "Search query is required"}, status=status.HTTP_400_BAD_REQUEST) address_ids = AddressMaster.objects.filter( Q(CustomerName__icontains=search_query) | Q(CustomerNickName__icontains=search_query) | Q(AddressLine1__icontains=search_query) | Q(AddressLine2__icontains=search_query) | Q(City__icontains=search_query) | Q(State__icontains=search_query) | Q(Country__icontains=search_query) | Q(PostalCode__icontains=search_query), IsDeleted=False ).values_list('Id', flat=True) return Response({"address_ids": list(address_ids)}, status=status.HTTP_200_OK) class GetAddressView(APIView): def post(self, request): filter_data = request.data.get('filter_data') if not search_query: return Response({"error": "Search query is required"}, status=status.HTTP_400_BAD_REQUEST) address_ids = AddressMaster.objects.filter( Q(CustomerName__icontains=search_query) | Q(CustomerNickName__icontains=search_query) | Q(AddressLine1__icontains=search_query) | Q(AddressLine2__icontains=search_query) | Q(City__icontains=search_query) | Q(State__icontains=search_query) | Q(Country__icontains=search_query) | Q(PostalCode__icontains=search_query), IsDeleted=False ).values_list('Id', flat=True) return Response({"address_ids": list(address_ids)}, status=status.HTTP_200_OK) class CustomerNameListView(APIView): """ List all customer's names. """ def post(self, request, *args, **kwargs): customers = AddressMaster.objects.filter(IsDeleted=False, ExactERPAddressId=F('ParentShipTo')) # Fetch all customers print(customers) serializer = CustomerNameSerializer(customers, many=True) # Serialize the data response_data = { "status_code": 200, "status": "success", "message": "Customer details fetched successfully", "data": serializer.data } return Response(response_data, status=status.HTTP_200_OK) class RelatedCustomerListView(APIView): def post(self, request): # Check if 'address_id' is in the request body if 'address_id' not in request.data: return Response({ 'status': 'error', 'status_code': 400, 'message': 'address_id is required', }, status=400) address_id = request.data.get('address_id') if not address_id: return Response({ 'status': 'error', 'status_code': 400, 'message': 'address_id cannot be null or empty', }, status=400) # SQL query to fetch InvoiceToCode and OrderedByCode query = """ SELECT DISTINCT acr.InvoiceToCode, acr.OrderedByCode FROM pvsmaster.dbo.AddressCodeRelationship acr JOIN pvsmaster.dbo.AddressMaster am ON acr.ShipToCode = am.ExactERPAddressId WHERE am.Id = %s """ with connection.cursor() as cursor: cursor.execute(query, [address_id]) # Using parameterized query for security result1 = cursor.fetchall() # Separate data for InvoiceToCode and OrderedByCode invoice_address_data = [] ordered_by_address_data = [] if result1: # Iterate through the result and fetch additional address details for row in result1: # Fetch InvoiceToCode's corresponding AddressMaster try: # For InvoiceToCode # invoice_address = AddressMaster.objects.get(ExactERPAddressId=row[0]) # # Avoid adding duplicates by checking if already added # if invoice_address.ExactERPAddressId not in [addr['ExactERPAddressId'] for addr in # invoice_address_data]: # invoice_address_data.append(model_to_dict(invoice_address)) # Annotate AddressId as an alias for Id invoice_address = AddressMaster.objects.annotate(AddressId=F('Id')).get(ExactERPAddressId=row[0]) # Convert to dictionary and rename field if not already added if invoice_address.ExactERPAddressId not in [addr['ExactERPAddressId'] for addr in invoice_address_data]: address_dict = model_to_dict(invoice_address) address_dict['AddressId'] = address_dict.pop('Id') # Optional: Rename Id to AddressId in the dictionary invoice_address_data.append(address_dict) # Step 2: Fetch remaining addresses using raw SQL and append to the same list with connection.cursor() as cursor: cursor.execute(""" SELECT a.Id AS AddressId, a.CustomerMasterId AS CustomerId, a.CustomerName, a.CustomerNickName, a.AddressLine1, a.AddressLine2, a.City, a.Country, a.ExactERPAddressId FROM AddressMaster a WHERE a.IsDeleted = 0 """) for db_row in cursor: # Check if the address is already in the invoice_address_data to avoid duplicates if db_row[8] not in [addr['ExactERPAddressId'] for addr in invoice_address_data]: invoice_address_data.append({ 'AddressId': db_row[0], 'CustomerId': db_row[1], 'CustomerName': db_row[2], 'CustomerNickName': db_row[3], 'AddressLine1': db_row[4], 'AddressLine2': db_row[5], 'City': db_row[6], 'Country': db_row[7], 'ExactERPAddressId': db_row[8] }) except AddressMaster.DoesNotExist: continue # Skip if the AddressMaster is not found for InvoiceToCode # Fetch OrderedByCode's corresponding AddressMaster try: # For OrderedByCode # ordered_by_address = AddressMaster.objects.get(ExactERPAddressId=row[1]) # # Avoid adding duplicates by checking if already added # if ordered_by_address.ExactERPAddressId not in [addr['ExactERPAddressId'] for addr in # ordered_by_address_data]: # ordered_by_address_data.append(model_to_dict(ordered_by_address)) # Annotate AddressId as an alias for Id ordered_by_address = AddressMaster.objects.annotate(AddressId=F('Id')).get(ExactERPAddressId=row[1]) # Convert to dictionary and rename field if not already added if ordered_by_address.ExactERPAddressId not in [addr['ExactERPAddressId'] for addr in ordered_by_address_data]: address_dict = model_to_dict(ordered_by_address) address_dict['AddressId'] = address_dict.pop('Id') # Optional: Rename Id to AddressId in the dictionary ordered_by_address_data.append(address_dict) # Step 2: Fetch remaining addresses using raw SQL and append to the same list with connection.cursor() as cursor: cursor.execute(""" SELECT a.Id AS AddressId, a.CustomerMasterId AS CustomerId, a.CustomerName, a.CustomerNickName, a.AddressLine1, a.AddressLine2, a.City, a.Country, a.ExactERPAddressId FROM AddressMaster a WHERE a.IsDeleted = 0 """) for db_row in cursor: # Check if the address is already in the ordered_by_address_data to avoid duplicates if db_row[8] not in [addr['ExactERPAddressId'] for addr in ordered_by_address_data]: ordered_by_address_data.append({ 'AddressId': db_row[0], 'CustomerId': db_row[1], 'CustomerName': db_row[2], 'CustomerNickName': db_row[3], 'AddressLine1': db_row[4], 'AddressLine2': db_row[5], 'City': db_row[6], 'Country': db_row[7], 'ExactERPAddressId': db_row[8] }) except AddressMaster.DoesNotExist: continue # Skip if the AddressMaster is not found for OrderedByCode # Prepare the response data response_data = { "status_code": 200, "status": "success", "message": "Customer details fetched successfully", "data": { "InvoiceToAddressData": invoice_address_data, "OrderedByAddressData": ordered_by_address_data } } return Response(response_data, status=status.HTTP_200_OK) def get_product_articlecode(auth_token, product_id): api_url = settings.PRODUCT_API_URL + "products/" headers = { 'x-api-key': settings.API_KEY, 'x-content-type': 'application/json', 'x-auth-token': auth_token, } payload = {"product_id": product_id} try: response = requests.post(api_url, json=payload, headers=headers) if response.status_code == 200: data = response.json() return data.get('data', {}).get('ArticleCode', None) else: return None except requests.RequestException as e: # Handle exceptions, such as connection errors return None class DeliverToCustomerListView(APIView): def post(self, request): data = request.data required_fields = ["product_id"] # Validate required fields for field in required_fields: value = data.get(field) if value is None or (isinstance(value, str) and value.strip() == ''): return Response({ "status_code": 400, "status": "error", "message": f"The field '{field}' is required and cannot be null or empty." }, status=status.HTTP_400_BAD_REQUEST) product_id = data.get('product_id') article_code = get_product_articlecode(request.headers.get('x-auth-token'), product_id) search_query = data.get('search_query') if search_query: addresses = ( AddressMaster.objects .filter( Q(CustomerName__icontains=search_query) | Q(CustomerNickName__icontains=search_query) ) .annotate( AddressId=F('Id'), CustomerId=F('CustomerMasterId'), IsExisting=Exists( AddressCodeRelationship.objects.filter( ShipToCode=OuterRef('ExactERPAddressId'), ArticleCode=article_code ) ), IsDisabled=Value(False, output_field=BooleanField()), ContainsArticleInNickName=Case( When(CustomerNickName__icontains=article_code, then=Value(True)), default=Value(False), output_field=BooleanField() ), IsRelationFound=Case( When( Q(CustomerNickName__icontains=article_code) | Q(ExactERPAddressId__in=AddressCodeRelationship.objects.filter( ArticleCode=article_code ).values('ShipToCode')), then=Value(True) ), default=Value(False), output_field=BooleanField() ) ) .order_by('-ContainsArticleInNickName', '-IsExisting') .values( 'AddressId', 'CustomerId', 'CustomerName', 'CustomerNickName', 'AddressLine1', 'AddressLine2', 'City', 'Country', 'ExactERPAddressId', 'IsDisabled', 'IsRelationFound', 'PostalCode', 'CountryCode', 'CityMapping', 'ParentShipTo' ) ) else: addresses = ( AddressMaster.objects.filter( IsDeleted=False, ExactERPAddressId=F('ParentShipTo'), IsActive=True) .annotate( AddressId=F('Id'), CustomerId=F('CustomerMasterId'), IsExisting=Case( When(ExactERPAddressId__in=AddressCodeRelationship.objects.filter( ArticleCode=article_code ).values('ShipToCode'), then=Value(1)), default=Value(0), output_field=IntegerField() ), ContainsArticleInNickName=Case( When(CustomerNickName__icontains=article_code, then=Value(True)), default=Value(False), output_field=BooleanField() ), IsRelationFound=Case( When( Q(CustomerNickName__icontains=article_code) | Q(ExactERPAddressId__in=AddressCodeRelationship.objects.filter( ArticleCode=article_code ).values('ShipToCode')), then=Value(True) ), default=Value(False), output_field=BooleanField() ) ) .order_by('-ContainsArticleInNickName', '-IsExisting') .values( 'AddressId', 'CustomerId', 'CustomerName', 'CustomerNickName', 'AddressLine1', 'AddressLine2', 'City', 'Country', 'ExactERPAddressId', 'IsRelationFound', 'PostalCode', 'CountryCode', 'CityMapping', 'ParentShipTo' ) ) # Format response data response_data = list(addresses) for address in response_data: parts = [ address.get('CustomerName', ''), address.get('AddressLine1', ''), address.get('AddressLine2', ''), address.get('CityMapping', ''), address.get('Country', ''), str(address.get('ParentShipTo', '')), # str(address.get('ExactERPAddressId', '')), ] address['SearchQuery'] = ' '.join(part.strip() for part in parts if part).strip() final_data = { "status_code": 200, "status": "success", "message": "Customer details fetched successfully", "data": response_data } return Response(final_data, status=status.HTTP_200_OK) class RelatedAddressListView(APIView): def post(self, request): data = request.data required_fields = ["address_id", "product_id"] for field in required_fields: value = data.get(field) if value is None or (isinstance(value, str) and value.strip() == ''): return Response({ "status_code": 400, "status": "error", "message": f"The field '{field}' is required and cannot be null or empty." }, status=status.HTTP_400_BAD_REQUEST) address_id = data.get('address_id') product_id = data.get('product_id') article_code = get_product_articlecode(request.headers.get('x-auth-token'), product_id) # SQL query to fetch InvoiceToCode and OrderedByCode query = """ SELECT DISTINCT acr.InvoiceToCode, acr.OrderedByCode,acr.Id FROM AddressCodeRelationship acr JOIN AddressMaster am ON acr.ShipToCode = am.ExactERPAddressId WHERE am.Id = %s AND acr.ArticleCode = %s order by acr.Id desc """ with connection.cursor() as cursor: cursor.execute(query, [address_id, article_code]) result1 = cursor.fetchall() # Set to store unique ExactERPAddressId values for existing addresses existing_invoice_ids = set() existing_ordered_by_ids = set() invoice_address_data = [] ordered_by_address_data = [] if result1: for row in result1: invoice_to_code = row[0] ordered_by_code = row[1] # Fetch InvoiceToCode's corresponding AddressMaster try: invoice_address = AddressMaster.objects.annotate(AddressId=F('Id')).get(ExactERPAddressId=invoice_to_code) address_dict = model_to_dict(invoice_address) address_combined = ",".join(filter(None, [ invoice_address.AddressLine1 # invoice_address.AddressLine2, # invoice_address.City, # invoice_address.PostalCode, # invoice_address.Country ])) address_dict['AddressLine1'] = address_combined address_dict['AddressId'] = address_dict.pop('Id') # Set IsRelationFound to True address_dict['IsRelationFound'] = True address_dict['SearchQuery'] = ' '.join(filter(None, [ address_dict.get('CustomerNickName', ''), address_dict.get('AddressLine1', ''), address_dict.get('AddressLine2', ''), address_dict.get('CityMapping', ''), address_dict.get('Country', '') ])).strip() invoice_address_data.append(address_dict) existing_invoice_ids.add(invoice_to_code) # Store existing InvoiceTo address ID except AddressMaster.DoesNotExist: pass # Fetch OrderedByCode's corresponding AddressMaster try: ordered_by_address = AddressMaster.objects.annotate(AddressId=F('Id')).get(ExactERPAddressId=ordered_by_code) address_dict = model_to_dict(ordered_by_address) address_combined = ",".join(filter(None, [ ordered_by_address.AddressLine1, # ordered_by_address.AddressLine2, # ordered_by_address.City, # ordered_by_address.PostalCode, # ordered_by_address.Country ])) address_dict['AddressLine1'] = address_combined address_dict['AddressId'] = address_dict.pop('Id') address_dict['SearchQuery'] = ' '.join(filter(None, [ address_dict.get('CustomerNickName', ''), address_dict.get('AddressLine1', ''), address_dict.get('AddressLine2', ''), address_dict.get('CityMapping', ''), address_dict.get('Country', '') ])).strip() # Set IsRelationFound to True address_dict['IsRelationFound'] = True ordered_by_address_data.append(address_dict) existing_ordered_by_ids.add(ordered_by_code) # Store existing OrderedBy address ID except AddressMaster.DoesNotExist: pass # Fetch remaining addresses from AddressMaster (excluding existing InvoiceTo and OrderedBy) remaining_invoice_addresses = AddressMaster.objects.exclude(ExactERPAddressId__in=existing_invoice_ids).annotate( AddressId=F('Id')).values( 'AddressId', 'CustomerMasterId', 'CustomerName', 'CustomerNickName', 'AddressLine1', 'AddressLine2', 'City', 'CityMapping', 'Country', 'ExactERPAddressId', 'PostalCode', 'CountryCode' ) remaining_ordered_by_addresses = AddressMaster.objects.exclude(ExactERPAddressId__in=existing_ordered_by_ids).annotate( AddressId=F('Id')).values( 'AddressId', 'CustomerMasterId', 'CustomerName', 'CustomerNickName', 'AddressLine1', 'AddressLine2', 'City', 'CityMapping', 'Country', 'ExactERPAddressId', 'PostalCode', 'CountryCode' ) # Format remaining addresses with IsRelationFound = False remaining_invoice_address_data = [] for address in remaining_invoice_addresses: address_combined = ",".join(filter(None, [ address['AddressLine1'] # address['AddressLine2'], # address['City'], # address.get('PostalCode', ''), # Ensure PostalCode exists # address['Country'] ])) address['AddressLine1'] = address_combined # Set IsRelationFound to False address['IsRelationFound'] = False address['SearchQuery'] = ' '.join(filter(None, [ address.get('CustomerNickName', ''), address.get('AddressLine1', ''), address.get('AddressLine2', ''), address.get('CityMapping', ''), address.get('Country', '') ])).strip() remaining_invoice_address_data.append(address) remaining_ordered_by_address_data = [] for address in remaining_ordered_by_addresses: address_combined = ",".join(filter(None, [ address['AddressLine1'] # address['AddressLine2'], # address['City'], # address.get('PostalCode', ''), # Ensure PostalCode exists # address['Country'] ])) address['AddressLine1'] = address_combined # Set IsRelationFound to False address['IsRelationFound'] = False address['SearchQuery'] = ' '.join(filter(None, [ address.get('CustomerNickName', ''), address.get('AddressLine1', ''), address.get('AddressLine2', ''), address.get('CityMapping', ''), address.get('Country', '') ])).strip() remaining_ordered_by_address_data.append(address) # Combine existing and remaining addresses, keeping existing ones on top final_invoice_addresses = invoice_address_data + remaining_invoice_address_data final_ordered_by_addresses = ordered_by_address_data + remaining_ordered_by_address_data # Prepare the response data response_data = { "status_code": 200, "status": "success", "message": "Customer details fetched successfully", "data": { "InvoiceToAddressData": final_invoice_addresses, # Existing on top, remaining below "OrderedByAddressData": final_ordered_by_addresses # Existing on top, remaining below } } return Response(response_data, status=status.HTTP_200_OK) class UpdateAddressCodeRelationshipView(APIView): def post(self, request): product_article_code = request.data.get('article_code') ship_to = request.data.get('ship_to') ordered_by = request.data.get('ordered_by') invoice_to = request.data.get('invoice_to') # Validate inputs if not (product_article_code and ship_to and ordered_by and invoice_to): return Response({'error': 'Missing required data.'}, status=status.HTTP_400_BAD_REQUEST) # Delete existing record if present AddressCodeRelationship.objects.filter( ArticleCode=product_article_code, ShipToCode=ship_to, OrderedByCode=ordered_by, InvoiceToCode=invoice_to ).delete() # Create new record AddressCodeRelationship.objects.create( ArticleCode=product_article_code, ShipToCode=ship_to, OrderedByCode=ordered_by, InvoiceToCode=invoice_to ) return Response({'message': 'AddressCodeRelationship updated.'}, status=status.HTTP_200_OK) class UpdateAddressView(APIView): """API to update an AddressMaster record""" @transaction.atomic def post(self, request): request_data = request.data address_id = request_data.get('address_id') if not address_id: return Response({ "status_code": 400, "status": "error", "message": "The field 'address_id' is required and cannot be null or empty.", "data": None }, status=status.HTTP_400_BAD_REQUEST) try: address = AddressMaster.objects.get(Id=address_id) except AddressMaster.DoesNotExist: return Response({ "status_code": 404, "status": "error", "message": "Address not found.", "data": None }, status=status.HTTP_404_NOT_FOUND) # Remove address_id before passing to serializer update_data = request_data.copy() update_data.pop('address_id', None) try: update_serializer = customer_serializer.AddressMasterSerializer( instance=address, data=update_data, partial=True ) update_serializer.is_valid(raise_exception=True) update_serializer.save() return Response({ "status_code": 200, "status": "success", "message": "Address updated successfully", "data": update_serializer.data }, status=status.HTTP_200_OK) except DRFValidationError as ve: # Get first error field and message field, messages = list(ve.detail.items())[0] message_text = messages[0] if isinstance(messages, list) else messages return Response({ "status_code": 400, "status": "error", "message": f"The field '{field}' is required and cannot be null or empty. ({message_text})", "data": None }, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({ "status_code": 500, "status": "error", "message": "An unexpected error occurred.", "data": None }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) from .scripts import sync_address_master class SyncAddressFromExactView(APIView): """API to update an AddressMaster record from Exact """ @transaction.atomic def post(self, request): addition_list, update_list = sync_address_master.sync_address_data_table() return Response({ "status_code": 200, "status": "success", "message": "Address updated Sucessfully", "data_added": addition_list, "data_updated": update_list, "data": None }, status=status.HTTP_200_OK) class SyncCitymappingFromPricemaster(APIView): """API to update city name from TransportPriceMaster and Price master""" # Get transport price of sugar campaign def post(self, request): auth_token = request.headers.get('x-auth-token') address_id = request.data.get('address_id') transport_price = customer_services.get_all_price_master_data(auth_token,address_id) print(transport_price) return Response( { 'data': transport_price, "status_code": 200, }, status=status.HTTP_200_OK ) class AddressMasterBulkCreateView(APIView): """Api to migrate address master data from production""" def post(self, request, *args, **kwargs): serializer = AddressMasterSerializer(data=request.data, many=True) serializer.is_valid(raise_exception=True) # bulk_create AddressMaster.objects.bulk_create( [AddressMaster(**item) for item in serializer.validated_data], ignore_conflicts=True # avoids duplicate key errors ) return Response({"status": "success", "created": serializer.validated_data}, status=status.HTTP_201_CREATED) class MigrateAddressMasterView(APIView): """Api to migrate address master data from production""" def post(self, request, *args, **kwargs): addresses = AddressMaster.objects.all() serializer = AddressMasterSerializer(addresses, many=True) address_data = serializer.data return Response({"status": "success", "created": address_data}, status=status.HTTP_201_CREATED) class AddressListView(APIView): def get(self, request): """API to get infra by department""" address_id = request.GET.get('address_id', None) search_key = request.GET.get('search', None) page_size = request.GET.get('page_size', 20) page_number = request.GET.get('page_no', 1) if address_id: try: customer = customer_models.AddressMaster.objects.get(id=address_id, is_deleted=False) except customer_models.AddressMaster.DoesNotExist: raise exceptions.NotFound(detail="customer with the given ID not found.") serializer = customer_serializer.BaseAddressMasterSerializer(instance=customer) result = serializer.data else: customers = customer_models.AddressMaster.objects.all() if search_key: customers = customers.filter( Q(CustomerNickName__icontains=search_key) | Q(ExactERPAddressId__icontains=search_key) | Q(City__icontains=search_key) ) # Apply pagination paginator = Paginator(customers, page_size) page_obj = paginator.get_page(page_number) serializer = customer_serializer.BaseAddressMasterSerializer(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 AddressMasterERPView(APIView): @transaction.atomic def post(self, request): """ API to create or update address details from ERP. """ data = request.data address_id = data.get('address_id') ship_to = data.get('ExactERPAddressId') if ship_to: data['ParentShipTo']=data['ExactERPAddressId'] if address_id: # Update existing Company try: customer = AddressMaster.objects.get(Id=address_id, IsDeleted=False) serializer = CustomerMasterSerializer( instance=customer, data=data, partial=True) except AddressMaster.DoesNotExist: raise exceptions.NotFound(detail="address with the given ID not found.") else: # Create a new company serializer = CustomerMasterSerializer(data=data) # Validate serializer if not serializer.is_valid(): return Response({ "status_code": 400, "status": "error", "message": f"{serializer.errors}", "data": None }, status=status.HTTP_400_BAD_REQUEST) # Save the address serializer.save() return Response( { 'status_code': status.HTTP_200_OK, 'status': "success", 'message': 'Company saved successfully', 'data': serializer.data, }, status=status.HTTP_200_OK ) def get(self, request): """API to get infra by department""" address_id = request.GET.get('address_id', None) page_size = request.GET.get('page_size', 20) page_number = request.GET.get('page_no', 1) if address_id: try: customer = AddressMaster.objects.get(Id=address_id, IsDeleted=False) except AddressMaster.DoesNotExist: raise exceptions.NotFound(detail="Company with the given ID not found.") serializer = CustomerMasterSerializer(instance=customer) result = serializer.data else: customer = AddressMaster.objects.all() # Apply pagination paginator = Paginator(customer, page_size) page_obj = paginator.get_page(page_number) serializer = CustomerMasterSerializer(page_obj, many=True) result = { "total_records": paginator.count, "total_pages": paginator.num_pages, "current_page": page_obj.number, "data": 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 Customer """ address_id = request.GET.get('address_id') try: customer = AddressMaster.objects.get(Id=address_id, IsDeleted=0) customer.IsDeleted=True customer.save() response = { "status": "success", "status_code": status.HTTP_200_OK, "message": "Customer Deleted Successfully", 'data': "Customer 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)