# ############ form application code ########### # from django.shortcuts import render # Create your views here. from rest_framework import viewsets from rest_framework.response import Response from rest_framework import status from rest_framework.decorators import action from rest_framework.permissions import IsAuthenticated from login_signup.custom_authentication import CustomJWTAuthentication from product.models import FavProduct from order.models import Cart # from login_signup.custom_authentication import CustomJWTAuthentication # from login_signup.custom_permissions import IsCustomUser from .models import * from .serializers import * from itertools import groupby from drf_spectacular.utils import extend_schema from django_filters.rest_framework import DjangoFilterBackend from django.utils import timezone from datetime import timedelta, datetime, time import django_filters from suscription.models import subscribe from rest_framework.permissions import AllowAny from Help_Support.decorators import cache_response from rest_framework.filters import SearchFilter import re from suscription.custom_pagination import CustomPagination # from rest_framework.viewsets import ViewSet from rest_framework.views import APIView from django.db.models import Q, Count, Case, When, IntegerField, Value, Avg, Sum, CharField from django.db.models.functions import Length, Random import json import hashlib import hmac from rest_framework import generics, permissions from django.db.models import Case, When, Value, IntegerField, Avg, F, ExpressionWrapper, OuterRef, Subquery from django.db.models.functions import Coalesce from django.db import IntegrityError from rest_framework.exceptions import NotFound import logging import pytz import random from django.core.cache import cache # from django.contrib.postgres.search import SearchQuery, SearchRank, SearchVector # Create your views here. logger = logging.getLogger(__name__) class CategoryFilter(django_filters.FilterSet): category_name = django_filters.CharFilter(field_name='category_name', lookup_expr='icontains') class Meta: model = Category fields = ['category_name'] class MainCategoryListing(generics.ListAPIView): serializer_class = MainCategoryListingSerializer def get_queryset(self): queryset = MainCategory.objects.all() category_id = self.request.query_params.get('category_id') if category_id is not None: queryset = queryset.filter(id=category_id) return queryset def get_serializer_class(self): if self.request.query_params.get('category_id') is not None: return MainCategoryWithPlansSerializer return self.serializer_class def check_permissions(self, request): category_id = request.query_params.get('category_id') if category_id is not None: self.permission_classes = [IsAuthenticated] for permission in self.get_permissions(): if not permission.has_permission(request, self): self.permission_denied( request, message="Authentication required", code=403 ) def list(self, request, *args, **kwargs): self.check_permissions(request) queryset = self.get_queryset() serializer = self.get_serializer(queryset, many=True, context={'request': request}) return Response({ "status": True, "message": "Main category listing fetch successfully", "data": serializer.data }) class CategoryPlanDurationUpdateView(generics.UpdateAPIView): permission_classes = [IsAuthenticated] queryset = CategoriesPlanDuration.objects.all() serializer_class = PlanDurationUpdateSerializer def update(self, request, *args, **kwargs): partial = kwargs.pop('partial', False) instance = self.get_object() required_fields = ['price_1_month', 'price_3_month', 'price_6_month', 'price_12_month', 'product_limit', 'boost_product_limit'] missing_fields = {field: ["This field is required."] for field in required_fields if field not in request.data} if missing_fields: return Response({ "status": False, "message": "Missing required fields", "errors": missing_fields # Returns an array for each missing field }, status=status.HTTP_400_BAD_REQUEST) price_fields = ['price_1_month', 'price_3_month', 'price_6_month', 'price_12_month'] updated_prices = {} for price_field in price_fields: try: updated_prices[price_field] = Decimal(str(request.data.get(price_field)).strip()) except (ValueError, TypeError, InvalidOperation): return Response({ "status": False, "message": "Invalid price value", "errors": {price_field: "Invalid decimal value."} }, status=status.HTTP_400_BAD_REQUEST) # instance.product_limit = request.data.get('product_limit', instance.product_limit) # instance.boost_product_limit = request.data.get('boost_product_limit', instance.boost_product_limit) # instance.status = request.data.get('status', instance.status) # instance.save() fields_updated = False if "product_limit" in request.data and request.data["product_limit"] != instance.product_limit: instance.product_limit = request.data["product_limit"] fields_updated = True if "boost_product_limit" in request.data and request.data["boost_product_limit"] != instance.boost_product_limit: instance.boost_product_limit = request.data["boost_product_limit"] fields_updated = True if fields_updated: instance.save() for price_field, new_price in updated_prices.items(): self.handle_price_update(instance, price_field, new_price) instance.refresh_from_db() response_data = PlanDurationUpdateSerializer(instance).data return Response({ "status": True, "message": "Plan duration updated successfully", "data": response_data }, status=status.HTTP_200_OK) def handle_price_update(self, instance, price_field, new_price): """Update price in DurationPrice model and assign new price reference.""" price_name = price_field.replace("price_", "").replace("_", " ") category_plan_id = instance.category_plan.id old_price_obj = DurationPrice.objects.filter( duration=instance, category_plan=category_plan_id, name=price_name ).first() if old_price_obj is None or old_price_obj.price != new_price: if old_price_obj: old_price_obj.deleted_at = now() old_price_obj.save() new_price_obj = DurationPrice.objects.create( duration=instance, category_plan=instance.category_plan, name=price_name, price=new_price ) setattr(instance, price_field, new_price_obj) instance.save() def update_main_category_ids(main_category_id=None): try: if main_category_id: main_category_obj = MainCategory.objects.filter(id=main_category_id).first() if main_category_obj: related_category_ids = Category.objects.filter(main_category=main_category_id).values_list('id', flat=True) main_category_obj.category_ids = ','.join(map(str, related_category_ids)) main_category_obj.save() all_categories = MainCategory.objects.filter(name="All Categories").first() if all_categories: all_category_ids = Category.objects.values_list('id', flat=True) all_categories.category_ids = ','.join(map(str, all_category_ids)) all_categories.save() except Exception as e: print(f"Error updating main category: {e}") class CategoryMVS(viewsets.ModelViewSet): # queryset = Category.objects.all() serializer_class = CategorySer filter_backends = [DjangoFilterBackend] filterset_class = CategoryFilter # pagination_class = CustomPagination def get_queryset(self): """ Return queryset based on whether the request comes from admin or buyer. """ user = self.request.user # If authenticated and not in buyer if user.is_authenticated and not user.is_buyer: return Category.objects.all() # Buyer (unauthenticated or non-admin) → only active categories return Category.objects.filter(active=True) def get_serializer_context(self): context = super().get_serializer_context() context['request'] = self.request return context def partial_update(self, request, pk=None): try: category = Category.objects.get(pk=pk) except Category.DoesNotExist: return Response({"error": "Category not found"}, status=status.HTTP_404_NOT_FOUND) serializer = self.get_serializer(category, data=request.data, partial=True,context={'request':request}) if serializer.is_valid(): serializer.save() if not category.active: logger.info(f"Category deactivated - ID: {category.id} ") try: products = Product.objects.filter(category=category) product_ids = list(products.values_list('id', flat=True)) if product_ids: # Find all cart items to delete carts_to_delete = Cart.objects.filter(product__in=products) cart_ids = list(carts_to_delete.values_list('id', flat=True)) if cart_ids: logger.info(f"Deleting {len(cart_ids)} carts with IDs {cart_ids} due to inactive category {category.id}") # Delete the cart items carts_to_delete.delete() logger.info(f"Successfully deleted carts for category {category.id}") else: logger.info(f"No cart items found to delete for category {category.id}") else: logger.info(f"No products found in category {category.id}, so no carts to delete.") except Exception as e: # Log the error but don't interrupt the response logger.error(f"Error removing products from carts for category {category.id}: {e}", exc_info=True) return Response({'msg':'category updated','data': serializer.data}, status=status.HTTP_200_OK) else: return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @cache_response() def retrieve(self, request, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance, context={'request': request}) # Retrieve all subcategories related to the category subcategories = instance.mainform_belong_to.order_by('-id') subcategory_serializer = SubCategorySer(subcategories, many=True,context={'request': request}) # Include subcategories in the response data response_data = serializer.data response_data['subcategories'] = subcategory_serializer.data return Response(response_data) @cache_response() def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()).order_by('order') main_category = request.query_params.get('main_category') if main_category: if main_category != "5": queryset = queryset.filter(main_category=main_category) # Create custom sorting using Case-When to give priority to the top categories # page = self.paginate_queryset(queryset) # if page is not None: # serializer = self.get_serializer(page, many=True, context={'request': request}) # return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True, context={'request': request}) return Response({'msg':'list product of category','data':serializer.data}) @action(detail=False, methods=['get'], url_path='by-category-id/(?P[^/.]+)') @cache_response() def list_by_category_id(self, request, category_id=None): # Ensure category_id is passed as part of the URL path if not category_id: return Response({'msg': 'Category ID is required'}, status=status.HTTP_400_BAD_REQUEST) # Try to fetch the category based on category_id category = self.get_queryset().filter(id=category_id).first() # print('-------||', self.get_serializer().__class__.__name__) if not category: return Response({'msg': 'Category not found'}, status=status.HTTP_404_NOT_FOUND) # Serialize the category data category_serializer = self.get_serializer(category, context={'request': request}) # Access related subcategories subcategories = category.mainform_belong_to.order_by('-id') subcategory_serializer = SubCategorySerializer(subcategories, many=True, context={'request': request}) # Combine the data response_data = category_serializer.data response_data['subcategories'] = subcategory_serializer.data return Response({'msg': 'Category list fetched successfully', 'data': response_data}, status=status.HTTP_200_OK) def create(self, request, *args, **kwargs): data = request.data.copy() # Make a mutable copy of the request data # data['icon'] = 'e360/category/icon/man.png' # if 'icon' not in data or not data['icon']: # return Response({'error': 'Please select category image'}, status=status.HTTP_400_BAD_REQUEST) serializer = self.get_serializer(data=data, context={'request': request}) serializer.is_valid(raise_exception=True) # if 'main_category' not in data or not data['main_category']: # return Response({'error': 'Please select category plan'}, status=status.HTTP_400_BAD_REQUEST) main_category_id = data.get('main_category') try: main_category_obj = MainCategory.objects.get(id=main_category_id) except MainCategory.DoesNotExist: return Response({'error': 'Selected main category does not exist'}, status=status.HTTP_400_BAD_REQUEST) # Proceed with category creation category = serializer.save() update_main_category_ids(main_category_id) # serializer = self.get_serializer(data=data,context={'request': request}) # serializer.is_valid(raise_exception=True) # self.perform_create(serializer) headers = self.get_success_headers(serializer.data) return Response({'msg': 'Product category created', 'data': serializer.data}, status=status.HTTP_201_CREATED, headers=headers) class CategoryViewSet(viewsets.ViewSet): permission_classes = [IsAuthenticated] serializer_class = CategorySerializer def create(self, request): if not request.user.is_superuser: return Response({"status":False, "error": "Only Admin Has this permission"}, status=status.HTTP_403_FORBIDDEN) category_id = request.data.get('category_id') if not category_id: return Response({"status":False, "error": "Category ID is required"}, status=status.HTTP_400_BAD_REQUEST) category = Category.objects.filter(id=category_id).first() if not category: return Response({"status":False, "error": "Category Not Found"}, status=status.HTTP_400_BAD_REQUEST) # category = get_object_or_404(Category, id=category_id) # category = get_object_or_404(Category, id=category_id) main_category_id = category.main_category.id if category.main_category else None category.deleted_at = datetime.now() category.save() # subcategories = SubCategory.objects.filter(category=category_id) # subcategories.update(deleted_at=None) # Update main category update_main_category_ids(main_category_id) return Response({"status":True, "message": "Category deleted successfully"}, status=status.HTTP_200_OK) class WebsiteCategoryViewSet(viewsets.ModelViewSet): queryset = WebsiteCategory.objects.all() serializer_class = WebsiteCategorySer # pagination_class = CustomPagination def create(self, request, *args, **kwargs): serializer = WebsiteCategorySer(data = request.data) serializer.is_valid(raise_exception=True) str_having_digits = serializer.validated_data.get('category_list') if not WebsiteCategory.objects.exists(): WebsiteCategory.objects.create(category_list = str_having_digits) else: obj = WebsiteCategory.objects.first() obj.category_list = str_having_digits obj.save() characters = str_having_digits.split(',') categories_list = [int(char) for char in characters] categories_objs_list = [Category.objects.get(id = category_id) for category_id in categories_list] for cat in categories_objs_list: category_ser_list = [CategorySer(cat).data for cat in categories_objs_list] return Response({'msg':'WebsiteCategory', 'data':category_ser_list},status=status.HTTP_200_OK) @cache_response() def list(self, request, *args, **kwargs): obj = WebsiteCategory.objects.first() if obj == None: return Response({'error':'No category exist in database'},status=status.HTTP_400_BAD_REQUEST) cat_list = obj.category_list characters = cat_list.split(',') categories_id_list = [int(char) for char in characters] category_obj_list = [Category.objects.get(id = category_id) for category_id in categories_id_list] # page = self.paginate_queryset(category_obj_list ) # if page is not None: # serializer = self.get_serializer(page, many=True) # return self.get_paginated_response(serializer.data) category_list_ser = [CategorySer(category_obj).data for category_obj in category_obj_list] return Response({'msg':'Get api run seccessfully', 'data':category_list_ser},status=status.HTTP_200_OK) class SubCategoryFilter(django_filters.FilterSet): subcategory_name = django_filters.CharFilter(field_name='subcategory_name', lookup_expr='icontains') category = django_filters.CharFilter(field_name='category__category_name', lookup_expr='icontains') class Meta: model = SubCategory fields = ['subcategory_name', 'category'] class SubCategoryMVS(viewsets.ModelViewSet): # queryset = SubCategory.objects.all().order_by('-id') serializer_class = SubCategorySer filter_backends = [DjangoFilterBackend] filterset_class = SubCategoryFilter # pagination_class = CustomPagination def get_queryset(self): # Use select_related for foreign key relationships return SubCategory.objects.select_related('category').order_by('-id') def partial_update(self, request, pk=None): try: subcategory = SubCategory.objects.get(pk=pk) except SubCategory.DoesNotExist: return Response({"error": "subCategory not found"}, status=status.HTTP_404_NOT_FOUND) serializer = self.get_serializer(subcategory, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response({'msg':'updated','data': serializer.data}, status=status.HTTP_200_OK) else: return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @cache_response() def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()).order_by('-id') page = self.paginate_queryset(queryset) # if page is not None: # serializer = self.get_serializer(page, many=True, context={'request': request}) # return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True, context={'request': request}) return Response({'msg': 'List of active subcategories', 'data': serializer.data}) def create(self, request, *args, **kwargs): data = request.data.copy() # Make a mutable copy of the request data if 'icon' not in data or not data['icon']: return Response({'error': 'Please select subcategory image'}, status=status.HTTP_400_BAD_REQUEST) subcategory_name = data.get('subcategory_name') category_id = data.get('category') if SubCategory.objects.filter(subcategory_name=subcategory_name, category_id=category_id).exists(): return Response({'error': 'This subcategory already exists for the selected category.'}, status=status.HTTP_400_BAD_REQUEST) serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) self.perform_create(serializer) headers = self.get_success_headers(serializer.data) return Response({'msg': 'Product subcategory created', 'data': serializer.data}, status=status.HTTP_201_CREATED, headers=headers) from django.shortcuts import get_object_or_404 from django.db.models import Case, When, BooleanField, Value,F class ProductMVS(viewsets.ModelViewSet): # queryset = Product.objects.all() # serializer_class = ProductSer # authentication_classes=[CustomJWTAuthentication] # permission_classes = [IsAuthenticated] pagination_class = CustomPagination # filter_backends = [DjangoFilterBackend] # filterset_class = ProductFilter def get_queryset(self): # Use select_related for foreign key relationships # return Product.objects.select_related('category', 'sub_category','user').order_by('-id') user = self.request.user # If authenticated and not in buyer if user.is_authenticated and not user.is_buyer: return Product.objects.select_related('category', 'sub_category','user').order_by('-id') # Buyer (unauthenticated or non-admin) → only active categories return Product.objects.filter(category__active=True).select_related('category', 'sub_category','user').order_by('-id') def get_serializer_class(self): is_buyer = self.request.query_params.get('is_buyer', False) if is_buyer: return BuyerProductSer return ProductSer @cache_response() def list(self, request): # Retrieve filter parameters from request is_buyer = request.query_params.get('is_buyer') category_id = request.query_params.get('category_id') sub_category= request.query_params.get('sub_category') original_price = request.query_params.get('original_price') discount_price = request.query_params.get('discount_price') discount_min = request.query_params.get('discount_price_min') discount_max = request.query_params.get('discount_price_max') ratings = request.query_params.getlist('rating') user_id = request.query_params.get('user_id') buyer_id = request.query_params.get('buyer_id') global_keyword = request.query_params.get('search') payment_status= request.query_params.get('payment_status') queryset = Product.objects.all().order_by('-id') if is_buyer: queryset= queryset.filter(payment_status='paid' , availability_status = 'available', status='active', category__active=True).exclude(user__status='inactive') queryset = queryset.order_by( Case( When(user__user_plan__plan__name='Gold', then=Value(1)), When(user__user_plan__plan__name='Silver', then=Value(2)), When(user__user_plan__plan__name='Free', then=Value(3)), default=Value(4), # Fallback to ensure any other plans are after gold, silver, free output_field=IntegerField(), ), '-id' # Secondary sorting by ID descending ) if category_id: queryset = queryset.filter(category_id=category_id) if sub_category: queryset = queryset.filter(sub_category_id=sub_category) if discount_min: queryset = queryset.filter(discount_price__gte=discount_min) if discount_max: queryset = queryset.filter(discount_price__lte=discount_max) if ratings: queryset = queryset.filter(product_review__rating__in=ratings) if user_id: queryset = queryset.filter(user_id=user_id) if payment_status: queryset= queryset.filter(payment_status=payment_status) if discount_price: # New: Filter by price if provided queryset = queryset.filter(discount_price=discount_price) if buyer_id: buyer = get_object_or_404(User, id=buyer_id) fav_product_ids = FavProduct.objects.filter(user=buyer).values_list('product_id', flat=True) queryset = queryset.annotate( is_wishlist=Case( When(id__in=fav_product_ids, then=Value(True)), default=Value(False), output_field=BooleanField() ) ) if global_keyword: cleaned_global_keyword = global_keyword.strip().lower() global_keyword = cleaned_global_keyword.replace(" ", "") keyword_regex = r'(?i)(^|,\s*)' + re.escape(cleaned_global_keyword) + r'(\s*,|$)' queryset = queryset.filter( Q(sub_category__tags__iregex=keyword_regex) | Q(title__icontains=global_keyword) ) serializer_class = BuyerProductSer if is_buyer else ProductSer page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = ProductSer(queryset, many=True,context={'request':request}) return Response({'msg':'list of products form','data':serializer.data},status=status.HTTP_200_OK) @cache_response() def retrieve(self, request, pk=None): try: # product = Product.objects.get(pk=pk) user = self.request.user # If authenticated and not in buyer if user.is_authenticated and not user.is_buyer: product = Product.objects.get(pk=pk) else : product = Product.objects.filter(category__active=True).get(pk=pk) except Product.DoesNotExist: return Response({"error": "Product not found"}, status=status.HTTP_404_NOT_FOUND) # user_id = request.query_params.get('user_id') user_id = request.user.id if user_id: if request.user.is_buyer: if product.viewed_user_list: viewed_user_list = product.viewed_user_list.split(',') else: viewed_user_list = [] if str(user_id) not in viewed_user_list: product.views += 1 viewed_user_list.append(str(user_id)) product.viewed_user_list = ','.join(viewed_user_list) product.save() print('User added to viewed list.') else: print('User already in viewed list.') # viewed_user_list = product.viewed_user_list.split(',') # if user_id not in viewed_user_list: # # Increment views and update viewed_user_list # product.views += 1 # if product.viewed_user_list: # product.viewed_user_list += f',{user_id}' # else: # product.viewed_user_list = user_id # product.save() # serializer = self.get_serializer(product) serializer = self.get_serializer(product, context={'request': request,'user': request.user}) return Response({'msg':'list of product','data':serializer.data},status=status.HTTP_200_OK) def create(self, request, *args, **kwargs): user_id = request.data.get('user') if user_id: try: user = User.objects.get(id=user_id) except User.DoesNotExist: return Response({'msg': 'User does not exist.'}, status=status.HTTP_400_BAD_REQUEST) else: return Response({'msg': 'User ID is required.'}, status=status.HTTP_400_BAD_REQUEST) # Get the user's subscription plan # subscription=subscribe.objects.get(user=user_id) # if not subscription: # return Response({'msg': 'User does not have an active subscription.'}, status=status.HTTP_400_BAD_REQUEST) # plan_name = subscription.plan.name # product_limit = subscription.plan.limit if plan_name in ['Silver', 'Free'] else float('inf') # # Count the user's existing products # user_product_count = user.count # user_status=subscription.is_expired # # Check if the user has reached the product limit # # if int(user_product_count) >= product_limit: # # return Response({'msg': 'Product limit reached. Upgrade your plan to add more products.'}, status=status.HTTP_400_BAD_REQUEST) # if user_status is True: # return Response({'msg':'Subscription plan expired please upgrade your plan'}) # payment= request.data.get('payment_status') # if payment == "paid" : # user.count += 1 # user.save() #Get the subscription for user's categorywise # product_subscription_plan = PlanPurchase.objects.filter(vendor=user_id, payment_status='successful', status='active').order_by('-id').first() # if not product_subscription_plan: # return Response({'msg': 'User does not have an active subscription.'}, status=status.HTTP_400_BAD_REQUEST) # current_date = now().date() # if current_date > product_subscription_plan.end_date: # return Response({'msg': 'User subscription plan expired. Please upgrade your plan.'}, status=status.HTTP_400_BAD_REQUEST) # if product_subscription_plan.products_added >= product_subscription_plan.product_limit : # return Response({'msg': 'Product limit reached. Upgrade your plan to add more products.'}, status=status.HTTP_400_BAD_REQUEST) # payment= request.data.get('payment_status') # if payment == 'paid' : # product_subscription_plan.product_increment(1) # product_subscription_plan.products_added += 1 # product_subscription_plan.save() # payment_status = request.data.get('payment_status') # if payment_status and payment_status=="paid": # request.data['active_at'] = now() serializer = self.get_serializer(data=request.data, context={'request': request}) serializer.is_valid(raise_exception=True) self.perform_create(serializer) return Response({'msg':'Product created successfully ','data':serializer.data},status=status.HTTP_201_CREATED) # def perform_create(self, serializer, uploaded_images): # product = serializer.save() # uploaded_images = self.request.FILES.getlist('uploaded_images') # for image in uploaded_images: # ProImage.objects.create(product=product, images=image) def partial_update(self, request, pk=None): try: product = Product.objects.get(pk=pk) except Product.DoesNotExist: return Response({"error": "Product not found"}, status=status.HTTP_404_NOT_FOUND) if request.data.get('status') and request.data.get('status') == 'inactive' : if product: Cart.objects.filter(product=pk).update(deleted_at=datetime.now()) serializer = self.get_serializer(product, data=request.data, partial=True,context={'request':request}) serializer.is_valid(raise_exception=True) serializer.save() return Response({'data':'product updated','data':serializer.data}, status=status.HTTP_200_OK) # def destroy(self, request, *args, **kwargs): # instance = self.get_object() # user = instance.user # # Check if the payment status is 'paid' # if instance.payment_status == 'paid': # user.count -= 1 # self.perform_destroy(instance) # user.save() # return Response({'msg': 'Product deleted successfully'}, status=status.HTTP_204_NO_CONTENT) def destroy(self, request, *args, **kwargs): try: instance = self.get_object() user = instance.user if instance.payment_status == 'paid' and instance.status == 'active': # user.count = max(user.count - 1, 0) # Ensure it doesn't go negative product_subscription_plan = PlanPurchase.objects.filter( vendor=user, payment_status= PlanPurchase.PaymentStatus.SUCCESS, status=PlanPurchase.Status.Active ).order_by('-id').first() product_subscription_plan.products_added = max(0, product_subscription_plan.products_added - 1) product_subscription_plan.save() # Try to delete the object self.perform_destroy(instance) user.save() return Response({'msg': 'Product deleted successfully'}, status=status.HTTP_204_NO_CONTENT) except NotFound: return Response({'error': 'Product not found'}, status=status.HTTP_404_NOT_FOUND) except IntegrityError: return Response({ 'error': 'This product cannot be deleted because it is referenced by other records.' }, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({ 'error': 'An unexpected error occurred', 'details': str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class ProductBoostAPIView(APIView): permission_classes = [IsAuthenticated] def post(self, request, product_id): try: vendor = request.user product = Product.objects.get(id=product_id, user=vendor) is_boosted = request.data.get("is_boosted", None) if is_boosted is None: return Response({ "status": False, "message": "'is_boosted' field is required (true or false)." }, status=status.HTTP_400_BAD_REQUEST) if is_boosted == True : if product.availability_status != 'available' or product.status != 'active' : return Response({ "status": False, "message": "Boosting is only available for activated products. Activate it and retry." }, status=status.HTTP_400_BAD_REQUEST) subscription = PlanPurchase.objects.filter(payment_status=PlanPurchase.PaymentStatus.SUCCESS, vendor=vendor).order_by('-id').first() if not subscription or not subscription.plan_duration or not subscription.plan_duration.boost_product_limit: return Response({ "status": False, "message": "Please purchase a subscription plan to boost your product." }, status=status.HTTP_400_BAD_REQUEST) total_boosted = Product.objects.filter(user=vendor, payment_status="paid", availability_status="available", status="active", is_boosted=True ).count() if total_boosted >= subscription.plan_duration.boost_product_limit: return Response({ "status": False, "message": "Boosted product limit exceeded. Please purchase another plan or deactivate an existing boosted product." }, status=status.HTTP_400_BAD_REQUEST) product.is_boosted = is_boosted product.save() return Response({ "status": True, "message": f"Product boosted status updated." }, status=status.HTTP_200_OK) except Product.DoesNotExist: return Response({ "status": False, "message": "Product not found." }, status=status.HTTP_404_NOT_FOUND) except Exception as e: logger.exception(f"Boost update failed for vendor {request.user.id}: {str(e)}") return Response({ "status": False, "message": "Something went wrong. Please try again later." }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class ProductMVSListing(viewsets.ViewSet): serializer_class = ProductListingSer pagination_class = CustomPagination def list(self, request, *args, **kwargs): category_id = request.query_params.get('category_id') # product_id = request.query_params.get('product_id') subcategory_id= request.query_params.get('subcategory_id') min_price= request.query_params.get('min_price') max_price= request.query_params.get('max_price') sort_by= request.query_params.get('sort_by') # search= request.query_params.get('search') # products = Product.objects.filter(payment_status='paid' , availability_status = 'available', status='active').exclude(user__status='inactive').select_related('category', 'sub_category','user').order_by('-created_at') # queryset = Product.objects.filter( # payment_status="paid", # availability_status="available", # status="active" # ).exclude( # user__status="inactive" # ).select_related( # "category", "sub_category", "user" # ).prefetch_related( # "product_review" # ).annotate( # average_rating=Avg("product_review__rating"), # boost_priority=Case( # When( # Q(is_boosted=True) & # Q(user__plan_purchases__status=PlanPurchase.Status.Active) & # Q(user__plan_purchases__payment_status=PlanPurchase.PaymentStatus.SUCCESS) & # Q(user__plan_purchases__plan_duration__boost_product_limit__isnull=False), # then=F("user__plan_purchases__plan_duration__boost_product_limit") # ), # default=Value(0), # output_field=IntegerField() # ), # ordering_priority=Case( # When(is_boosted=True, then=Value(1)), # default=Value(0), # output_field=IntegerField() # ), # rand=Random() # ).order_by( # "-ordering_priority", # Boosted first # "-boost_priority", # Within boosted, highest priority first # "rand" # Random within group # ) # Random Boosted Product Query Start latest_boost_limit_subquery = PlanPurchase.objects.filter( vendor=OuterRef('user'), status=PlanPurchase.Status.Active, payment_status=PlanPurchase.PaymentStatus.SUCCESS, plan_duration__boost_product_limit__isnull=False ).order_by('-plan_duration__boost_product_limit').values('plan_duration__boost_product_limit')[:1] queryset = Product.objects.filter( payment_status="paid", availability_status="available", status="active", category__active=True ).exclude( user__status="inactive" # ).select_related( # "category", "sub_category", "user" # ).prefetch_related( # "product_review" ).annotate( boost_priority=Case( When(is_boosted=True, then=Coalesce( Subquery(latest_boost_limit_subquery, output_field=IntegerField()), Value(0) )), default=Value(0), output_field=IntegerField() ), is_boosted_flag=Case( When(is_boosted=True, then=Value(1)), default=Value(0), output_field=IntegerField() ), rand=Random(), average_rating=Avg("product_review__rating") ) # .order_by( # '-is_boosted_flag', # '-boost_priority', # 'rand' # ).distinct() # Random Boosted Product Query End if category_id: queryset = queryset.filter(category_id=category_id) if subcategory_id: queryset = queryset.filter(sub_category_id=subcategory_id) if min_price: queryset = queryset.filter(discount_price__gte=min_price) if max_price: queryset = queryset.filter(discount_price__lte=max_price) if sort_by: sort_by = int(sort_by) if sort_by == 1: queryset = queryset.order_by('-average_rating') # queryset = queryset.order_by('-product_review__rating') elif sort_by == 2: queryset = queryset.order_by('discount_price') elif sort_by == 3: queryset = queryset.order_by('-discount_price') else: queryset = queryset.order_by( '-boost_priority', '-is_boosted_flag', 'rand' ) if not queryset.exists(): return Response( { 'status': False, 'message': 'No product found.', 'data': [] }, status=status.HTTP_200_OK ) paginator = self.pagination_class() paginated_products = paginator.paginate_queryset(queryset, request, view=self) serializer = self.serializer_class(paginated_products, many=True, context={'request': request, 'user': request.user}) return paginator.get_paginated_response(serializer.data) class ProductListing(viewsets.ViewSet): permission_classes = [IsAuthenticated] serializer_class = ProductListingSer pagination_class = CustomPagination def list(self, request, *args, **kwargs): if not request.user.is_superuser or not request.user.otp_verified: return Response( {'detail': 'Only Admin Can Access This Endpoint'}, status=status.HTTP_403_FORBIDDEN ) payment_status = request.query_params.get('payment_status') search = request.query_params.get('search') start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') queryset = Product.objects.all() if payment_status: queryset= queryset.filter(payment_status=payment_status) if start_date and end_date: # # Parse to ensure correct format # try: # start_dt = datetime.strptime(start_date, '%Y-%m-%d') # end_dt = datetime.strptime(end_date, '%Y-%m-%d') + timedelta(days=1) # include full end day # queryset = queryset.filter(created_at__range=(start_dt, end_dt)) # logger.info(f'Final query: { queryset.query}') # except ValueError: # logger.info('Invalid datetime format.') tz = pytz.timezone("Africa/Dar_es_Salaam") try: start_dt = tz.localize(datetime.combine(datetime.strptime(start_date, '%Y-%m-%d').date(), time.min)) end_dt = tz.localize(datetime.combine(datetime.strptime(end_date, '%Y-%m-%d').date(), time.max)) queryset = queryset.filter(created_at__range=(start_dt, end_dt)) logger.info(f'Final query: {queryset.query}') except ValueError: logger.info('Invalid datetime format.') if search: exact_match_qs = queryset.filter( Q(title__icontains=search) | Q(user__full_name__icontains=search) | Q(category__category_name__icontains=search) | Q(sub_category__subcategory_name__icontains=search) ) search_cleaner = re.sub(r'[^a-zA-Z0-9\s]', '', search) search_words = search_cleaner.split() word_match_qs = Product.objects.none() if len(search_words) > 1: word_match_query = Q() for word in search_words: word_match_query |= ( Q(title__icontains=word) | Q(user__full_name__icontains=word) | Q(category__category_name__icontains=word) | Q(sub_category__subcategory_name__icontains=word) ) word_match_qs = queryset.filter(word_match_query).exclude(id__in=exact_match_qs) else: exact_match_qs = exact_match_qs.order_by('-id') queryset = list(exact_match_qs) + list(word_match_qs) else : queryset = queryset.order_by('-id') paginator = self.pagination_class() paginated_buyer = paginator.paginate_queryset(queryset, request, view=self) serializer = self.serializer_class( paginated_buyer, many=True, context={'request': request, 'user': request.user} ) return paginator.get_paginated_response(serializer.data) class ProductSimilarSearchListing(viewsets.ViewSet) : pagination_class = CustomPagination def list(self, request): keyword = request.query_params.get('keyword') if not keyword: return Response({ 'status': False, 'message': 'Please search any keyword.', 'data': [] }, status=status.HTTP_404_NOT_FOUND) try: products = Product.objects.filter(category__active=True).all() categories = Category.objects.filter(active=True).all() subcategories = SubCategory.objects.filter(category__active=True).all() if keyword: # products = products.filter(title__icontains=keyword).distinct() query = ''' SELECT *, MATCH(title) AGAINST (%s IN NATURAL LANGUAGE MODE) AS relevance FROM product_product WHERE MATCH(title) AGAINST (%s IN NATURAL LANGUAGE MODE) ORDER BY relevance DESC; ''' products = Product.objects.raw(query, [keyword, keyword]) # vector = SearchVector('title') # query = SearchQuery(keyword) # products = Product.objects.annotate( # rank=SearchRank(vector, query) # ).filter(rank__gt=0).order_by('-rank') # keyword = re.sub(r'\s+', ' ',keyword.lower()) # words = keyword.split() print('-----------------') # regex_pattern_full = r"\b" + r"\s+".join(re.escape(word) for word in words) + r"\b" # Full sequence match # regex_pattern_partial = ( # r"\b" + r"\s+".join(re.escape(word) for word in words[1:]) + r"\b" # if len(words) > 1 else None # ) # Partial match ignoring first word # # Base query for filtering products # base_query = Q() # for word in words: # base_query |= Q(title__icontains=word) # products = Product.objects.filter(base_query) # # Define When conditions separately # when_conditions = [ # When(title__iexact=keyword, then=Value(1)), # When(title__iregex=regex_pattern_full, then=Value(2)), # When(base_query, then=Value(4)) # This ensures base query words get priority # ] # # Add the partial regex condition only if it's valid # if regex_pattern_partial: # when_conditions.append(When(title__iregex=regex_pattern_partial, then=Value(3))) # # Annotate priority and order results # products = products.annotate( # priority=Case( # *when_conditions, # Unpacking the When conditions # default=Value(5), # output_field=IntegerField(), # ) # ).order_by("priority", "title").distinct() # print('-----------------|||||') # regex_pattern_full = r"\b" + r"\s+".join(re.escape(word) for word in words) + r"\b" # Full sequence match # regex_pattern_partial = r"\b" + r"\s+".join(re.escape(word) for word in words[1:]) + r"\b" if len(words) > 1 else None # Partial match ignoring first word # print('-------------|||----') # base_query = Q() # for word in words: # base_query |= Q(title__icontains=word) # print('-------------|||----////') # products = Product.objects.filter(base_query) # print('-------------|||--||--////') # products = products.annotate( # priority=Case( # When(title__iexact=keyword, then=Value(1)), # When(title__iregex=regex_pattern_full, then=Value(2)), # When(title__iregex=regex_pattern_partial, then=Value(3)) if regex_pattern_partial else Value(3), # When(base_query, then=Value(4)), # default=Value(5), # output_field=IntegerField(), # ) # ).order_by("priority", "title").distinct() print('-----------------|||||') # regex_pattern = r".*" + r".*".join(re.escape(word) for word in keyword.split()) + r".*" # words = keyword.split() # base_query = Q() # for word in words: # base_query |= Q(title__icontains=word) # products = Product.objects.filter(base_query) # products = products.annotate( # priority=Case( # When(title__iregex=regex_pattern, then=Value(1)), # Full sequence match # When(title__icontains=keyword, then=Value(2)), # Contains full keyword anywhere # default=Value(3), # Other partial matches # output_field=IntegerField(), # ) # ).order_by("priority", "title").distinct() print('-----------------products', products) categories = categories.filter(category_name__icontains=keyword).distinct() subcategories = SubCategory.objects.filter( Q(tags__icontains=keyword) | Q(subcategory_name__icontains=keyword) ).distinct() if not products and not categories and not subcategories: return Response( { 'status': False, 'message': 'No results found for the provided keyword.', 'data': [] }, status=status.HTTP_404_NOT_FOUND ) # Apply pagination only on products paginator = self.pagination_class() paginated_products = paginator.paginate_queryset(products, request, view=self) print('------|||||||//////') product_serializer = ProductSimilarListingSer(paginated_products, many=True, context={'request': request}) category_serializer = CategorySer(categories, many=True, context={'request': request}) subcategory_serializer = SubCategorySer(subcategories, many=True, context={'request': request}) print('------|||||||') return Response({ 'status': True, 'message': 'Search results retrieved successfully.', 'data': { "count": paginator.page.paginator.count, "next": paginator.get_next_link(), "previous": paginator.get_previous_link(), "categories": category_serializer.data, "subcategories": subcategory_serializer.data, "products": product_serializer.data, } }, status=status.HTTP_200_OK) except Exception as e: return Response({ 'error': str(e), 'message': 'An error occurred while processing the request.', 'data': [] }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class ProductSearchListing(viewsets.ViewSet): # serializer_class = ProductListingSer pagination_class = CustomPagination def list(self, request): keyword = request.query_params.get('keyword') pagination_check = request.query_params.get('page_size') if not keyword: return Response({ 'status': False, 'message': 'Please search any keyword.', 'data': [] }, status=status.HTTP_404_NOT_FOUND) try: products = Product.objects.all() # categories = Category.objects.all() # subcategories = SubCategory.objects.all() if keyword: # products = products.filter(title__icontains=keyword).distinct() latest_boost_limit_subquery = PlanPurchase.objects.filter( vendor=OuterRef('user'), status=PlanPurchase.Status.Active, payment_status=PlanPurchase.PaymentStatus.SUCCESS, plan_duration__boost_product_limit__isnull=False ).order_by('-plan_duration__boost_product_limit').values('plan_duration__boost_product_limit')[:1] products = Product.objects.filter( Q(title__icontains=keyword) | Q(sub_category__tags__icontains=keyword), payment_status="paid", availability_status="available", status="active", category__active=True ).exclude( user__status="inactive" ).annotate( match_priority=Case( When(title__icontains=keyword, then=Value(2)), When(sub_category__tags__icontains=keyword, then=Value(1)), default=Value(0), output_field=IntegerField() ), # boost_priority=Coalesce(Subquery(latest_boost_limit_subquery, output_field=IntegerField()), Value(0)), # is_boosted=Case( # When(boost_priority__gt=0, then=Value(1)), # default=Value(0), # output_field=IntegerField() # ), boost_priority=Case( When(is_boosted=True, then=Coalesce(Subquery(latest_boost_limit_subquery, output_field=IntegerField()), Value(0))), default=Value(0), output_field=IntegerField() ), sponsor_tag=Case( When(boost_priority__gt=0, then=Value("Sponsored")), default=Value(None), output_field=CharField() ), rand=Random(), average_rating=Avg("product_review__rating") ).order_by( "-match_priority", "-boost_priority", "rand", "-created_at" ).distinct() # categories = categories.filter(category_name__icontains=keyword).distinct() subcategories = SubCategory.objects.filter( Q(tags__icontains=keyword) | Q(subcategory_name__icontains=keyword), category__active=True ).distinct() # if not products and not categories and not subcategories: if not products and not subcategories: return Response( { 'status': False, 'message': 'No results found for the provided keyword.', 'data': [] }, status=status.HTTP_404_NOT_FOUND ) if pagination_check : paginator = self.pagination_class() paginated_products = paginator.paginate_queryset(products, request, view=self) product_serializer = ProductSearchListingSer(paginated_products, many=True, context={'request': request}) else : product_serializer = ProductSearchListingSer(products, many=True, context={'request': request}) # product_serializer = ProductListingSer(paginated_products, many=True, context={'request': request}) # product_serializer = ProductSearchListingSer(paginated_products, many=True, context={'request': request}) # category_serializer = CategorySer(categories, many=True, context={'request': request}) subcategory_serializer = SubCategorySer(subcategories, many=True, context={'request': request}) if pagination_check : return Response({ 'status': True, 'message': 'Search results retrieved successfully.', 'data': { "count": paginator.page.paginator.count, "next": paginator.get_next_link(), "previous": paginator.get_previous_link(), # "categories": category_serializer.data, "subcategories": subcategory_serializer.data, "products": product_serializer.data, } }, status=status.HTTP_200_OK) else : return Response({ 'status': True, 'message': 'Search results retrieved successfully.', 'data': { # "categories": category_serializer.data, "subcategories": subcategory_serializer.data, "products": product_serializer.data, } }, status=status.HTTP_200_OK) except Exception as e: return Response({ 'error': str(e), 'message': 'An error occurred while processing the request.', 'data': [] }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) # def list(self, request): # keyword = request.query_params.get('keyword') # try: # products = Product.objects.all() # keywords = keyword.split() # if len(keywords) > 1: # category_filter = Q() # for key in keywords: # category_filter |= Q(category_name__icontains=key) # categories_ids = categories.filter(category_filter).values_list('id', flat=True).distinct() # if categories_ids.exists(): # product_filter = Q() # for key in keywords: # product_filter |= Q(title__icontains=key) # remaining_keyword = " ".join(keywords[1:]) # exact_match_filter = Q(title__icontains=remaining_keyword) # products = products.filter(product_filter, category_id__in=categories_ids).annotate( # match_count=Sum( # Case( # *[When(title__icontains=key, then=1) for key in keywords], # default=0, # output_field=IntegerField() ) ) ) # exact_match_products = products.filter(exact_match_filter) # other_products = products.exclude(exact_match_filter).order_by('-match_count') # exclude_ids = set(exact_match_products.values_list('id', flat=True)) | set(other_products.values_list('id', flat=True)) # other_match_products = Product.objects.filter(product_filter).exclude(id__in=exclude_ids) # products = list(exact_match_products) + list(other_products) + list(other_match_products) # subcategories = [] # else: # products = products.filter(title__icontains=keyword).distinct() # subcategories = [] # categories = categories.filter(category_name__icontains=keyword).distinct() class VendorProductListMVS(viewsets.ModelViewSet): # queryset = Product.objects.all() serializer_class = ProductSer authentication_classes = [CustomJWTAuthentication] # pagination_class = CustomPagination def get_queryset(self): # Use select_related for foreign key relationships return Product.objects.filter(active=True).select_related('category', 'subcategory','user').order_by('-id') def create(self, request, *args, **kwargs): vendor_id= request.user.id product_obj = Product.objects.filter(user_id = vendor_id) product_ser = ProductSer(product_obj, many=True) return Response({'data':product_ser.data},status=status.HTTP_200_OK) @cache_response() def list(self, request, *args, **kwargs): vendor_id = request.user.id # Filter products by user and payment_status = "paid" product_obj = Product.objects.filter(user_id=vendor_id, payment_status="paid") # Check if the filtered queryset is empty or not if not product_obj.exists(): return Response({'error': 'No products found', 'data': []}, status=status.HTTP_404_NOT_FOUND) # page = self.paginate_queryset(product_obj) # if page is not None: # serializer = self.get_serializer(page, many=True) # return self.get_paginated_response(serializer.data) product_ser = ProductSer(product_obj, many=True) return Response({'data': product_ser.data}, status=status.HTTP_200_OK) class ListProductViewSet(viewsets.ModelViewSet): # queryset = Product.objects.all() serializer_class = ListProductSer # pagination_class = CustomPagination def get_queryset(self): # Use select_related for foreign key relationships return Product.objects.select_related('category', 'subcategory','user').order_by('-id') @cache_response() def list(self, request, *args, **kwargs): queryset = self.get_queryset().order_by('-id') serializer = self.get_serializer(queryset, many=True) discount_price = [] # page = self.paginate_queryset(queryset) # if page is not None: # serializer = self.get_serializer(page, many=True) # return self.get_paginated_response(serializer.data) for ser_obj in serializer.data: discount_price.append(ser_obj['discount_price']) # return Response(serializer.data) return Response({'msg':'list of products','price':discount_price},status=status.HTTP_200_OK) class NewArrivalProductViewSet(viewsets.ViewSet): def list(self, request): products = Product.objects.filter(payment_status='paid' , availability_status = 'available', status='active',category__active=True).select_related('category', 'sub_category','user').exclude(user__status='inactive').order_by('-created_at')[:6] if not products.exists(): return Response( { 'status': False, 'message': 'No product found.', 'data': [] }, status=status.HTTP_404_NOT_FOUND ) serialized_data = BuyerProductSer(products, many=True, context={'request': request,'user': request.user}).data return Response( { 'status': True, 'message': 'Product data fetched successfully.', 'data': serialized_data }, status=status.HTTP_200_OK ) # class NewArrivalProductAPIView(APIView): # def get(self, request, format=None): # products = Product.objects.filter( # payment_status='paid', # availability_status='available', # status='active' # ).select_related('category', 'sub_category', 'user').exclude(user__status='inactive').order_by('-created_at')[:6] # if not products.exists(): # return Response( { 'status': False, 'message': 'No product found.', 'data': [] }, status=status.HTTP_404_NOT_FOUND ) # serialized_data = BuyerProductSer(products, many=True, context={'request': request}).data # return Response( { 'status': True, 'message': 'Product data fetched successfully.', 'data': serialized_data }, status=status.HTTP_200_OK ) class FavProductViewSet(viewsets.ModelViewSet): # queryset = FavProduct.objects.all() serializer_class = FavProductSer # pagination_class = CustomPagination def get_queryset(self): # Use select_related for foreign key relationships return FavProduct.objects.select_related('product','user').filter( product__availability_status='available', product__status='active', product__category__active=True ).order_by('-id') @cache_response() def list(self, request, *args, **kwargs): queryset = self.get_queryset().order_by('-id') # page = self.paginate_queryset(queryset) # if page is not None: # serializer = self.get_serializer(page, many=True) # return self.get_paginated_response(serializer.data) serializer = FavProductSer(queryset, many=True,context={'request':request}) return Response({'msg':'list of favourite products','data':serializer.data}) def create(self, request): serializer = FavProductSer(data = request.data,context={'request':request}) serializer.is_valid(raise_exception = True) user = serializer.validated_data['user'] if user.is_buyer==False and user.email != 'csvjoy@gmail.com': return Response({'status':False,'error': 'you do not have permission','data':[]}, status=status.HTTP_403_FORBIDDEN) product = serializer.validated_data['product'] existing_fav = FavProduct.objects.filter(user = user, product = product).first() if existing_fav: existing_fav.delete() return Response({'status':True, 'msg': 'Product removed from favorites','data':[]}, status=status.HTTP_200_OK) else: serializer.save() return Response({'status':True,'msg': 'Product added to favorites','data':serializer.data}, status=status.HTTP_201_CREATED) @extend_schema(operation_id='retrieve_products_by_user_id') @action(detail=False, methods=['get'], url_path='(?P[^/.]+)') @cache_response() def list_by_user(self, request, user_id=None): queryset = self.get_queryset().filter(user_id=user_id) page = self.paginate_queryset(queryset) # if page is not None: # serializer = self.get_serializer(page, many=True) # return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True,context={'request':request}) # for product in serializer.data: # product['product']['is_wishlist'] = True return Response({'msg': 'Favorite products list fetched successfully', 'data': serializer.data}, status=status.HTTP_200_OK) class ReviewViewSet(viewsets.ModelViewSet): # queryset = Review.objects.all() # permission_classes = [IsAuthenticated] serializer_class = ReviewSer # pagination_class = CustomPagination def get_queryset(self): queryset = Review.objects.select_related('product','user').order_by('-id') product_id = self.request.query_params.get('product', None) if product_id is not None: queryset = queryset.filter(product_id=product_id) return queryset def create(self, request, *args, **kwargs): user = request.user or None product = request.data.get('product', None) if product: product = request.data.get('product', None) if user : order_item = request.data.get('order_item', None) if order_item : if not OrderItem.objects.filter(buyer=user.id, product=product, is_received=True, id=order_item).exists(): return Response( {"status": False, "message": "You have not received this product yet.", "data": []}, status=status.HTTP_400_BAD_REQUEST ) if Review.objects.filter(product=product, user=user.id, order_item=order_item).exists() : return Response( {"status": False, "message": "You have already submitted a review for this order.", "data": []}, status=status.HTTP_400_BAD_REQUEST ) else : if not OrderItem.objects.filter(buyer=user.id, product=product, is_received=True).exists(): return Response( {"status": False, "message": "You have not purchased this product yet.", "data": []}, status=status.HTTP_400_BAD_REQUEST ) if Review.objects.filter(user=user, product=product).exists(): return Response( {"status": False, "message": "You have already submitted a review for this product.", "data": []}, status=status.HTTP_400_BAD_REQUEST ) else : user = request.data.get('user', None) if not OrderItem.objects.filter(buyer=user, product=product, is_received=True).exists(): return Response( {"status": False, "message": "You have not purchased this product yet.", "data": []}, status=status.HTTP_400_BAD_REQUEST ) # if Review.objects.filter(user=user, product=product).exists(): # return Response( # {"status": False, "message": "You have already submitted a review for this product.", "data": []}, # status=status.HTTP_400_BAD_REQUEST # ) serializer = self.get_serializer(data=request.data) if serializer.is_valid(): serializer.save() return Response( {"status": "success", "message": "Review created successfully.", "data": serializer.data}, status=status.HTTP_201_CREATED ) return Response( {"status": "error", "message": "Review creation failed.", "errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST ) def update(self, request, *args, **kwargs): return Response("working on update",status=status.HTTP_200_OK) class HelpFullViewSet(viewsets.ModelViewSet): # queryset = HelpFull.objects.all() serializer_class = HelpFullSer # pagination_class = CustomPagination def get_queryset(self): # Use select_related for foreign key relationships return Product.objects.select_related('review','user').order_by('-id') def create(self, request): serializer = self.get_serializer(data = request.data) serializer.is_valid(raise_exception = True) user = serializer.validated_data['user'] review = serializer.validated_data['review'] helpfull = serializer.validated_data['helpfull'] hf_obj = HelpFull.objects.filter(user = user, review = review).first() if hf_obj: if hf_obj.helpfull == helpfull: hf_obj.delete() return Response({'msg':'HelpFull is removed due to again click'}, status = 200) else: hf_obj.helpfull = False return Response({'msg':'HelpFull is removed'}) else: if helpfull == False: return Response({'msg':'Review is already marked as not HelpFull'},status=status.HTTP_200_OK) helpfull_obj = HelpFull.objects.create(user = user, review = review, helpfull = helpfull) ser_obj = HelpFullSer(helpfull_obj) return Response({'msg':'This review is HelpFull', 'data':ser_obj.data}, status=status.HTTP_200_OK) class FormMVS(viewsets.ModelViewSet): # queryset = Form.objects.all() serializer_class = FormSer # pagination_class = CustomPagination def get_queryset(self): # Use select_related for foreign key relationships return Form.objects.select_related('subcategory').order_by('-id') @cache_response() def list(self, request, *args, **kwargs): sub_category_id = request.query_params.get('sub_category') if sub_category_id is not None: queryset = self.get_queryset().filter(subcategory=sub_category_id) if queryset.exists(): serializer = self.get_serializer(queryset.first()) return Response(serializer.data,status=status.HTTP_200_OK) else: return Response({'message': 'No objects found for the specified subcategory.'},status=status.HTTP_400_BAD_REQUEST) else: # return Response({'message': 'Subcategory ID is required as a query parameter.'}, status=400) queryset = self.get_queryset() ##page = self.paginate_queryset(queryset) # if page is not None: # serializer = self.get_serializer(page, many=True) # return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return Response(serializer.data,status=status.HTTP_200_OK) class DynamicFormMVS(viewsets.ModelViewSet): serializer_class = DynamicFormSer pagination_class = CustomPagination # Optional: Use if you want pagination def get_queryset(self): # Optimize queries with select_related return DynamicForm.objects.select_related('category', 'subcategory').order_by('-id') @cache_response() def list(self, request, *args, **kwargs): category_id = request.query_params.get('category_id') sub_category_id = request.query_params.get('sub_category_id') if category_id and sub_category_id: queryset = self.get_queryset().filter(category=category_id, subcategory=sub_category_id).order_by('-id') if queryset.exists(): serializer = self.get_serializer(queryset, many=True) return Response({ 'message': 'Forms found for the specified category and subcategory', 'data': serializer.data }, status=status.HTTP_200_OK) else: return Response({'message': 'The specified category and subcategory do not have an associated form.'}, status=status.HTTP_400_BAD_REQUEST) elif category_id: queryset = self.get_queryset().filter(category=category_id).order_by('-id') if queryset.exists(): serializer = self.get_serializer(queryset, many=True) return Response({ 'message': 'Forms found for the specified category', 'data': serializer.data }, status=status.HTTP_200_OK) else: return Response({'message': 'The specified category does not have an associated form.'}, status=status.HTTP_400_BAD_REQUEST) elif sub_category_id: queryset = self.get_queryset().filter(subcategory=sub_category_id).order_by('-id') if queryset.exists(): serializer = self.get_serializer(queryset, many=True) return Response({ 'message': 'Forms found for the specified subcategory', 'data': serializer.data }, status=status.HTTP_200_OK) else: return Response({'message': 'The specified subcategory does not have an associated form.'}, status=status.HTTP_400_BAD_REQUEST) else: queryset = self.get_queryset().order_by('-id') serializer = self.get_serializer(queryset, many=True) return Response({ 'message': 'List of dynamic forms', 'data': serializer.data }, status=status.HTTP_200_OK) class OfferModelViewSet(viewsets.ModelViewSet): # queryset = OfferModel.objects.all() serializer_class = OfferModelSerializer authentication_classes = [CustomJWTAuthentication] permission_classes = [IsAuthenticated] # pagination_class = CustomPagination def get_queryset(self): # Use select_related for foreign key relationships return Product.objects.select_related('buyer', 'seller','product').order_by('-id') def get_serializer_class(self): if self.action == 'create': return OfferModelCreateSerializer return OfferModelSerializer @cache_response() def list(self, request, *args, **kwargs): queryset = self.get_queryset().order_by('-id') # page = self.paginate_queryset(queryset) # if page is not None: # serializer = self.get_serializer(page, many=True) # return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return Response({'msg':'list of offers for respective vendor','data':serializer.data}) def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) self.perform_create(serializer) headers = self.get_success_headers(serializer.data) return Response({'msg': 'offer is successfully send to respective vendor '}, status=status.HTTP_200_OK, headers=headers) class BuyerOfferView(viewsets.ViewSet): permission_classes = [IsAuthenticated] serializer_class = OfferModelSerializer pagination_class = CustomPagination def list(self, request, *args, **kwargs): if not request.user.otp_verified or request.user.is_vendor: return Response( {'detail': 'Only Buyer Can Access This Endpoint'}, status=status.HTTP_403_FORBIDDEN ) queryset = OfferModel.objects.filter(buyer_id=request.user.id).all() queryset = queryset.order_by('-id') paginator = self.pagination_class() paginated_buyer = paginator.paginate_queryset(queryset, request, view=self) serializer = self.serializer_class( paginated_buyer, many=True, context={'request': request, 'user': request.user} ) return paginator.get_paginated_response(serializer.data) class ProductOfferViewSet(viewsets.ViewSet): @cache_response() def list(self, request): products = Product.objects.all() response_data = [] for product in products: offers = OfferModel.objects.filter(product=product).order_by('-id') if offers.exists(): offer_serializer = OfferModelSerializer(offers, many=True) product_serializer = ProductSer(product,context={'request':request}) # report_serializer = ReportSerializer(reports, many=True) # product_serializer = ProductSdetailserializer(product) product_data = { 'product': product_serializer.data, 'offers': offer_serializer.data } response_data.append(product_data) response_data = sorted(response_data, key=lambda x: x['product']['id'], reverse=True) return Response(response_data, status=status.HTTP_200_OK) class BulkUpdateCategoryViewSet(viewsets.ViewSet): def create(self, request): data = request.data # Expecting an array of objects errors = [] for item in data: category_id = item.get('id') # Ensure the ID is provided for updating if not category_id: errors.append({"error": "ID is required for updating."}) continue try: category_instance = Category.objects.get(id=category_id) except Category.DoesNotExist: errors.append({"error": f"Answer with ID {category_id} does not exist."}) continue serializer = CategorySer(category_instance, data=item, partial=True) if serializer.is_valid(): serializer.save() else: errors.append(serializer.errors) if errors: return Response({"errors": errors}, status=status.HTTP_400_BAD_REQUEST) return Response({"message": "All category updated successfully."}, status=status.HTTP_200_OK) class SponsoredProductAPIView(generics.ListAPIView): serializer_class = ProductListingSer def list(self, request, *args, **kwargs): # queryset = Product.objects.filter( # payment_status="paid", # availability_status="available", # status="active", # is_boosted=True # ).exclude( # user__status="inactive" # ).order_by('?')[:10] product_ids = list(Product.objects.filter( payment_status="paid", availability_status="available", status="active", category__active=True, is_boosted=True ).exclude( user__status="inactive" ).values_list('id', flat=True)) random_ids = random.sample(product_ids, min(len(product_ids), 10)) queryset = Product.objects.filter(id__in=random_ids) if not queryset.exists(): return Response( { 'status': False, 'message': 'No product found.', 'data': [] }, status=status.HTTP_200_OK ) serializer = self.serializer_class( queryset, many=True, context={'request': request, 'user': request.user} ) return Response({ "status": True, 'message': 'Sponsored products fetched successfully.', "data": serializer.data }) class RandomProductAPIView(generics.ListAPIView): serializer_class = ProductListingSer def get_queryset(self): try: # Filter valid products and fetch 10 random ones queryset = ( Product.objects .filter( payment_status="paid", availability_status="available", status="active", category__active=True ) .exclude(user__status="inactive") .order_by(Random()) .distinct()[:10] ) return queryset except Exception as e: logger.exception("Error while fetching random products: %s", e) return Product.objects.none() def list(self, request, *args, **kwargs): queryset = self.get_queryset() if not queryset.exists(): return Response( { "status": False, "message": "No product found.", "data": [] }, status=status.HTTP_200_OK ) serializer = self.serializer_class( queryset, many=True, context={'request': request, 'user': request.user} ) return Response({ "status": True, "message": "Random products fetched successfully.", "data": serializer.data }) class BestSellingProductAPIView(generics.ListAPIView): serializer_class = ProductListingSer queryset = Product.objects.none() # required by DRF, but overridden in `list()` def list(self, request, *args, **kwargs): """ Retrieve the top 10 best-selling products that are active, paid, and available. Uses caching to improve performance and error handling for reliability. """ try: # cache_key = "best_selling_products" # cached_data = cache.get(cache_key) # if cached_data: # logger.info("Best selling products fetched from cache.") # return Response({ # "status": True, # "message": "Best selling products fetched successfully (cached).", # "data": cached_data # }, status=status.HTTP_200_OK) # Fetch top 10 products based on total quantity sold top_products = ( Product.objects .filter( order_items__isnull=False, payment_status="paid", availability_status="available", status="active", category__active=True ) .annotate(total_sold=Sum('order_items__quantity')) .order_by('-total_sold') .distinct()[:10] ) if not top_products: logger.info("No top-selling products found.") return Response({ "status": True, "message": "No top-selling products found.", "data": [] }, status=status.HTTP_200_OK) serializer = self.get_serializer(top_products, many=True, context={'request': request}) serialized_data = serializer.data # Cache the result for 5 minutes (adjust time as needed) # cache.set(cache_key, serialized_data, timeout=300) logger.info("Best selling products fetched and cached.") return Response({ "status": True, "message": "Best selling products fetched successfully.", "data": serialized_data }, status=status.HTTP_200_OK) except Exception as e: logger.exception("Error fetching best selling products.") return Response({ "status": False, "message": "An error occurred while fetching best selling products.", "error": str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)