from httpcore import request from rest_framework.views import APIView from rest_framework.response import Response from admin_panel.models import * from admin_panel.serializers import * from rest_framework import status from django.shortcuts import get_object_or_404 from django.core.mail import EmailMessage from .utils import get_client_ip, get_location from rest_framework.permissions import IsAuthenticated, AllowAny from admin_panel.rbac import RBACPermission from django.db.models import Count, OuterRef, Subquery from django.db.models.functions import TruncDate from datetime import timedelta, date import secrets #=====Soumya add=================== from django.db import IntegrityError from django.db import models # <-- ADD THIS from rest_framework_simplejwt.authentication import JWTAuthentication # <-- ADD THIS from django.utils import timezone from django.http import HttpResponse from rest_framework import viewsets from rest_framework.decorators import action from .utils import get_client_ip import re from urllib.parse import quote, unquote from django.core.mail import get_connection, EmailMessage from django.shortcuts import redirect, get_object_or_404 #-----------------------login------------------------------ from rest_framework_simplejwt.tokens import RefreshToken from django.contrib.auth import authenticate from .models import EmailTrackingEvent from openpyxl import Workbook import csv from openpyxl.styles import * class AdminLoginJWTView(APIView): def post(self, request): username = request.data.get("username") password = request.data.get("password") user = authenticate(username=username, password=password) if user is None: return Response({"error": "Invalid credentials"}, status=400) refresh = RefreshToken.for_user(user) user.last_active = timezone.now() user.save() return Response({ "access": str(refresh.access_token), "refresh": str(refresh) }) #--------------------------------------------Product Category API View--------------------------------------------# class ProductCategoryAPIView(APIView): module_name = "product_category" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): categories = ProductCategory.objects.filter( is_active=True, parent=None ) serializer = ProductCategorySerializer(categories, many=True) return Response(serializer.data) def post(self, request): serializer = ProductCategorySerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, pk): try: category = ProductCategory.objects.get(pk=pk) except ProductCategory.DoesNotExist: return Response({"error": "Category not found."}, status=404) serializer = ProductCategorySerializerPATCH(category, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, pk): try: category = ProductCategory.objects.get(pk=pk) except ProductCategory.DoesNotExist: return Response({"error": "Category not found."}, status=404) category.is_active = False category.save() return Response(status=204) #--------------------------------------------Product API View--------------------------------------------# class StockHandelAPIView(APIView): module_name = "product" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] # FIX 1: ADD GET METHOD def get(self, request, product_id): try: product = Product.objects.get(id=product_id) return Response({ "is_stock": product.is_stock }) except Product.DoesNotExist: return Response( {"error": "Product not found"}, status=status.HTTP_404_NOT_FOUND ) # FiX 2: CLEAN PATCH def patch(self, request, product_id): try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response( {"error": "Product not found"}, status=status.HTTP_404_NOT_FOUND ) is_stock = request.data.get('is_stock') # Optional safety if is_stock is None: return Response( {"error": "is_stock is required"}, status=status.HTTP_400_BAD_REQUEST ) product.is_stock = is_stock product.save() return Response({ "message": "Stock updated", "is_stock": product.is_stock }) from django.db.models import Q class ProductAPIView(APIView): module_name = "product" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, product_id=None): if product_id: try: product = Product.objects.get( id=product_id, is_active=True, is_draft=False ) serializer = ProductSerializer(product) return Response(serializer.data) except Product.DoesNotExist: return Response( {"error": "Product not found."}, status=404 ) products = Product.objects.filter( is_active=True, is_draft=False ) # ----------------------------- # Category Filter # ----------------------------- category = request.query_params.get("category") if category: try: parent_category = ProductCategory.objects.get( name__iexact=category, is_active=True ) category_ids = [parent_category.id] def get_children(cat): for child in cat.children.filter(is_active=True): category_ids.append(child.id) get_children(child) get_children(parent_category) products = products.filter(category_id__in=category_ids) except ProductCategory.DoesNotExist: return Response( {"error": "Category not found."}, status=404 ) serializer = ProductSerializer( products.order_by("-updated_at"), many=True ) return Response(serializer.data) def post(self, request): serializer = ProductSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, product_id): try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) serializer = ProductSerializerPOST( product, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, product_id): try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) product.is_active = False product.save() return Response(status=204) class draftProductAPIView(APIView): module_name = "product" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): products = Product.objects.filter(is_active=True,is_draft=True).order_by('-updated_at') serializer = ProductSerializer(products, many=True) return Response(serializer.data) #--------------------------------------------Product Platform API View--------------------------------------------# class PlatformAPIView(APIView): module_name = "product" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, product_id=None): if product_id: platforms = Platform.objects.filter(product__id=product_id, is_active=True) serializer = PlatformSerializer(platforms, many=True) return Response(serializer.data) else: platforms = Platform.objects.filter(is_active=True) serializer = PlatformSerializer(platforms, many=True) return Response(serializer.data) def post(self, request, product_id): try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) serializer = PlatformSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save(product=product) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, pk): try: platform = Platform.objects.get(pk=pk) except Platform.DoesNotExist: return Response({"error": "Platform not found."}, status=404) serializer = PlatformSerializerPOST(platform, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, pk): try: platform = Platform.objects.get(pk=pk) except Platform.DoesNotExist: return Response({"error": "Platform not found."}, status=404) platform.is_active = False platform.save() return Response(status=204) #--------------------------------------------Product Image API View--------------------------------------------# class ProductImageAPIView(APIView): module_name = "product" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, product_id): images = ProductImage.objects.filter(product__id=product_id, is_active=True) serializer = ProductImageSerializer(images, many=True) return Response(serializer.data) def post(self, request, product_id): try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) serializer = ProductImageSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save(product=product) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, image_id): try: image = ProductImage.objects.get(id=image_id) except ProductImage.DoesNotExist: return Response({"error": "Image not found."}, status=404) serializer = ProductImageSerializerPOST(image, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, image_id): try: image = ProductImage.objects.get(id=image_id) except ProductImage.DoesNotExist: return Response({"error": "Image not found."}, status=404) image.is_active = False image.save() return Response(status=204) #--------------------------------------------Product Click API View--------------------------------------------# class ProductClickAPIView(APIView): def get(self, request, product_id=None): if product_id: try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) clicks = ProductClick.objects.filter(product=product) serializer = ProductClickSerializer(clicks, many=True) return Response(serializer.data) else: clicks = ProductClick.objects.all() serializer = ProductClickSerializer(clicks, many=True) return Response(serializer.data) def post(self, request, product_id): try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) click = ProductClick.objects.create(product=product) return Response({ "message": "Product click recorded successfully.", "click_id": click.id, "product": product.product_name, "clicked_on": click.clicked_on }, status=200) #--------------------------------------------Marketplace Click API View--------------------------------------------# class MarketplaceClickAPIView(APIView): module_name = "dashboard" def get_permissions(self): return [AllowAny()] def get(self, request): start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') clicks_qs = MarketplaceClick.objects.all() if start_date: clicks_qs = clicks_qs.filter(clicked_on__gte=start_date) if end_date: clicks_qs = clicks_qs.filter(clicked_on__lte=end_date) data = ( clicks_qs.values('platform__name') .annotate(total_clicks=Count('id')) .order_by('-total_clicks')[:5] ) colors = ["#F297A0", "#B6BB79", "#2D3748", "#3182CE", "#ED8936"] formatted_data = [ {"name": item["platform__name"], "value": item["total_clicks"], "fill": colors[i % len(colors)]} for i, item in enumerate(data) ] return Response(formatted_data) def post(self, request, platform_id): try: platform = Platform.objects.get(id=platform_id) except Platform.DoesNotExist: return Response({"error": "Platform not found."}, status=404) serializer = MarketplaceClickSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save(platform=platform) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) #--------------------------------------------Spider Chart API View--------------------------------------------# from django.db.models import Avg, Q from django.db.models.functions import Coalesce from django.db.models import Avg from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import AllowAny from admin_panel.models import ProductCategory, ProductClick, Testimonial, Platform, Product class SpiderchartAPI(APIView): module_name = "dashboard" permission_classes = [IsAuthenticated, RBACPermission] # public access def get(self, request): start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') categories = ProductCategory.objects.filter(is_active=True) data = [] for cat in categories: # 1. Clicks (already correct) clicks_qs = ProductClick.objects.filter(product__category=cat) if start_date: clicks_qs = clicks_qs.filter(clicked_on__gte=start_date) if end_date: clicks_qs = clicks_qs.filter(clicked_on__lte=end_date) clicks = clicks_qs.count() # 2. Customer Ratings: get all product names in this category product_names = Product.objects.filter(category=cat, is_active=True).values_list('product_name', flat=True) ratings_qs = Testimonial.objects.filter(product_name__in=product_names, active=True) if start_date: ratings_qs = ratings_qs.filter(posted_on__gte=start_date) if end_date: ratings_qs = ratings_qs.filter(posted_on__lte=end_date) avg_customer_rating = ratings_qs.aggregate(avg=Avg('rating_value'))['avg'] or 0 # 3. Marketplace Ratings (already correct) mkt_ratings_qs = Platform.objects.filter(product__category=cat, is_active=True) avg_marketplace_rating = mkt_ratings_qs.aggregate(avg=Avg('rating'))['avg'] or 0 data.append({ "category": cat.name, "clicks": clicks, "customer_ratings": round(avg_customer_rating, 2), "marketplace_ratings": round(avg_marketplace_rating, 2), }) return Response(data) #---------------------------------------------Line Chart----------------------------------- class TrafficAnalyticsView(APIView): module_name = "dashboard" permission_classes = [IsAuthenticated, RBACPermission] def get(self, request): start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') queryset = Traffic.objects.all() if start_date: queryset = queryset.filter(date__gte=start_date) if end_date: queryset = queryset.filter(date__lte=end_date) data = ( queryset.values('date') .annotate(count=Count('id')) .order_by('date') ) labels = [str(item['date']) for item in data] values = [item['count'] for item in data] return Response({"labels": labels, "data": values}) #--------------------------------------------Top Performing Products API View--------------------------------------------# from django.db.models import Count from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status class topproductsAPI(APIView): module_name = "dashboard" permission_classes = [IsAuthenticated, RBACPermission] def get(self, request): start_date = request.query_params.get("start_date") end_date = request.query_params.get("end_date") products = Product.objects.filter( is_active=True ).prefetch_related("platforms") result = [] for product in products: # ------------------------------------------------- # PRODUCT DIRECT CLICKS # ------------------------------------------------- product_clicks_qs = ProductClick.objects.filter( product=product ) if start_date: product_clicks_qs = product_clicks_qs.filter( clicked_on__gte=start_date ) if end_date: product_clicks_qs = product_clicks_qs.filter( clicked_on__lte=end_date ) product_click_count = product_clicks_qs.count() # ------------------------------------------------- # MARKETPLACE CLICKS # ------------------------------------------------- marketplace_clicks_qs = MarketplaceClick.objects.filter( platform__product=product ) if start_date: marketplace_clicks_qs = marketplace_clicks_qs.filter( clicked_on__gte=start_date ) if end_date: marketplace_clicks_qs = marketplace_clicks_qs.filter( clicked_on__lte=end_date ) total_marketplace_clicks = marketplace_clicks_qs.count() # ------------------------------------------------- # PLATFORM-WISE CLICKS # ------------------------------------------------- platform_data = [] for platform in product.platforms.filter(is_active=True): platform_clicks_qs = MarketplaceClick.objects.filter( platform=platform ) if start_date: platform_clicks_qs = platform_clicks_qs.filter( clicked_on__gte=start_date ) if end_date: platform_clicks_qs = platform_clicks_qs.filter( clicked_on__lte=end_date ) platform_click_count = platform_clicks_qs.count() # Calculate share if total_marketplace_clicks > 0: share = round( ( platform_click_count / total_marketplace_clicks ) * 100, 1 ) else: share = 0 platform_data.append({ "id": platform.id, "name": platform.name, "url": platform.url, "clicks": platform_click_count, "share": share }) # Highest marketplace clicks first platform_data.sort( key=lambda x: x["clicks"], reverse=True ) result.append({ "id": product.id, "name": product.product_name, # Direct product clicks "product_clicks": product_click_count, # All marketplace clicks "marketplace_clicks": total_marketplace_clicks, "platforms": platform_data }) # ------------------------------------------------- # TOP PRODUCTS # SORT BY MARKETPLACE CLICKS # ------------------------------------------------- result.sort( key=lambda x: x["marketplace_clicks"], reverse=True ) return Response(result) #--------------------------------------------KPI API View--------------------------------------------# from django.utils import timezone from datetime import timedelta from rest_framework.views import APIView from rest_framework.response import Response class KPIAPI(APIView): module_name = "dashboard" permission_classes = [IsAuthenticated, RBACPermission] def get(self, request): start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') # Helper to filter by date range (inclusive) def filter_by_date(qs, date_field): if start_date: qs = qs.filter(**{f"{date_field}__gte": start_date}) if end_date: qs = qs.filter(**{f"{date_field}__lte": end_date}) return qs # Total counts (all‑time) total_products = Product.objects.filter(is_active=True).count() total_clicks = ProductClick.objects.count() total_subscribers = NewsletterSubscriber.objects.filter(is_active=True).count() # Default trends product_trend = 0 click_trend = 0 subscriber_trend = 0 # Only compute trends if both dates are provided and valid if start_date and end_date: try: # Convert strings to date objects from datetime import datetime start = datetime.strptime(start_date, '%Y-%m-%d').date() end = datetime.strptime(end_date, '%Y-%m-%d').date() if start <= end: delta = (end - start).days + 1 # Previous period of same length prev_start = start - timedelta(days=delta) prev_end = end - timedelta(days=delta) # Products created in current period vs previous curr_products = Product.objects.filter( is_active=True, created_at__date__gte=start, created_at__date__lte=end ).count() prev_products = Product.objects.filter( is_active=True, created_at__date__gte=prev_start, created_at__date__lt=start ).count() product_trend = ((curr_products - prev_products) / prev_products * 100) if prev_products > 0 else (100 if curr_products > 0 else 0) # Clicks in current period vs previous curr_clicks = ProductClick.objects.filter( clicked_on__gte=start, clicked_on__lte=end ).count() prev_clicks = ProductClick.objects.filter( clicked_on__gte=prev_start, clicked_on__lt=start ).count() click_trend = ((curr_clicks - prev_clicks) / prev_clicks * 100) if prev_clicks > 0 else (100 if curr_clicks > 0 else 0) # Subscribers in current period vs previous curr_subs = NewsletterSubscriber.objects.filter( is_active=True, subscribed_on__date__gte=start, subscribed_on__date__lte=end ).count() prev_subs = NewsletterSubscriber.objects.filter( is_active=True, subscribed_on__date__gte=prev_start, subscribed_on__date__lt=start ).count() subscriber_trend = ((curr_subs - prev_subs) / prev_subs * 100) if prev_subs > 0 else (100 if curr_subs > 0 else 0) except Exception as e: # If any date conversion fails, keep trends as 0 pass return Response({ "total_products": total_products, "product_trend_percentage": round(product_trend, 2), "total_clicks": total_clicks, "click_trend_percentage": round(click_trend, 2), "total_subscribers": total_subscribers, "subscriber_trend_percentage": round(subscriber_trend, 2) }) #--------------------------------------------Recent Activity API View--------------------------------------------# #-------------------------------------Pritam-------------------------------------# from rest_framework import generics, status from rest_framework.response import Response from rest_framework.decorators import api_view from django.http import JsonResponse from .models import Blog, Tag, Testimonial from .serializers import BlogSerializer, TagSerializer, TestimonialSerializer import os from django.conf import settings # TAG class TagListCreateView(APIView): module_name = "tag" def get_permissions(self): if self.request.method == "POST": return [IsAuthenticated(), RBACPermission()] return [AllowAny()] def get(self, request): tags = Tag.objects.filter(is_active=True) serializer = TagSerializer(tags, many=True) return Response(serializer.data) def post(self, request): serializer = TagSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def patch(self, request, pk): try: tag = Tag.objects.get(pk=pk) except Tag.DoesNotExist: return Response({"error": "Tag not found."}, status=status.HTTP_404_NOT_FOUND) serializer = TagSerializer(tag, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def delete(self, request, pk): try: tag = Tag.objects.get(pk=pk) except Tag.DoesNotExist: return Response({"error": "Tag not found."}, status=status.HTTP_404_NOT_FOUND) tag.is_active = False tag.save() return Response(status=status.HTTP_204_NO_CONTENT) # BLOG class BlogListCreateView(generics.ListCreateAPIView): queryset = Blog.objects.all() serializer_class = BlogSerializer module_name = "blog" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] class BlogDetailView(generics.RetrieveUpdateDestroyAPIView): queryset = Blog.objects.all() serializer_class = BlogSerializer module_name = "blog" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] class BlogLikeAPIView(APIView): def get(self, request, blog_id): try: blog = Blog.objects.get(id=blog_id) except Blog.DoesNotExist: return Response({"error": "Blog not found"}, status=status.HTTP_404_NOT_FOUND) return Response({ "blog_id": blog.id, "total_likes": blog.bloglikes.count() }) def post(self, request, blog_id): try: blog = Blog.objects.get(id=blog_id) except Blog.DoesNotExist: return Response({"error": "Blog not found"}, status=status.HTTP_404_NOT_FOUND) BlogLikes.objects.create(blog=blog) return Response({ "message": "Blog liked successfully", "total_likes": blog.bloglikes.count() }, status=status.HTTP_201_CREATED) class BlogSerializer(serializers.ModelSerializer): class Meta: model = Blog fields = [ 'id', 'tag', 'title', 'author', 'description', 'posted_on', 'likes', 'publish', 'active', 'image', # SEO fields 'meta_title', 'meta_description', 'meta_keywords', 'og_title', 'og_description', 'og_image', 'twitter_card', 'twitter_title', 'twitter_description', 'twitter_image', 'canonical_url', 'robots_tag', 'structured_data', 'focus_keyword' ] # TESTIMONIAL class TestimonialListCreateView(generics.ListCreateAPIView): queryset = Testimonial.objects.all().order_by('order', '-posted_on') serializer_class = TestimonialSerializer module_name = "testimonial" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def create(self, request, *args, **kwargs): serializer = self.get_serializer(data={ 'name': request.data.get('name'), 'avatar_url': request.data.get('avatar_url', ''), 'product_name': request.data.get('product_name'), 'rating_value': request.data.get('rating_value', 5.0), 'rating_stars': request.data.get('rating_stars', 5), 'description': request.data.get('description'), 'order': request.data.get('order', 0), }) serializer.is_valid(raise_exception=True) testimonial = serializer.save(active=True) photo_paths = [] video_paths = [] # PHOTOS for photo in request.FILES.getlist('photos'): folder = os.path.join(settings.MEDIA_ROOT, 'testimonials', str(testimonial.id), 'images') os.makedirs(folder, exist_ok=True) file_path = os.path.join(folder, photo.name) with open(file_path, 'wb+') as f: for chunk in photo.chunks(): f.write(chunk) photo_paths.append(f'testimonials/{testimonial.id}/images/{photo.name}') # VIDEOS for video in request.FILES.getlist('videos'): folder = os.path.join(settings.MEDIA_ROOT, 'testimonials', str(testimonial.id), 'videos') os.makedirs(folder, exist_ok=True) file_path = os.path.join(folder, video.name) with open(file_path, 'wb+') as f: for chunk in video.chunks(): f.write(chunk) video_paths.append(f'testimonials/{testimonial.id}/videos/{video.name}') # Update fields testimonial.photos = photo_paths testimonial.videos = video_paths testimonial.save() return Response(self.get_serializer(testimonial).data, status=status.HTTP_201_CREATED) class TestimonialDetailView(generics.RetrieveUpdateDestroyAPIView): queryset = Testimonial.objects.all().order_by('order', '-posted_on') serializer_class = TestimonialSerializer module_name = "testimonial" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] #--------------------------------------------Newsletter Subscriber API View--------------------------------------------# class NewsletterSubscriberAPIView(APIView): module_name = "newsletter_subscriber" def get_permissions(self): if self.request.method == "POST": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): subscribers = NewsletterSubscriber.objects.filter(is_active=True).order_by('-subscribed_on') serializer = NewsletterSubscriberSerializer(subscribers, many=True) return Response(serializer.data) def post(self, request): data = request.data.copy() # Check whether email OTP is verified verified_otp = NewsletterOTP.objects.filter( email=data.get("email"), verified=True ).first() if not verified_otp: return Response( { "message": "Please verify your email with OTP before subscribing." }, status=status.HTTP_400_BAD_REQUEST ) ip = get_client_ip(request) city, country = get_location(ip) user_agent = request.META.get('HTTP_USER_AGENT', '') data['ip_address'] = ip data['city'] = city data['country'] = country data['user_agent'] = user_agent if not data.get('source'): data['source'] = 'NEWSLETTER SUBSCRIBER' # Mark email as verified data['email_verified'] = True serializer = NewsletterSubscriberSerializer(data=data) if serializer.is_valid(): subscriber = serializer.save() # Delete OTP after successful registration (recommended) verified_otp.delete() return Response( serializer.data, status=status.HTTP_201_CREATED ) return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) def patch(self, request, subscriber_id): subscriber = get_object_or_404( NewsletterSubscriber, id=subscriber_id ) status_value = request.data.get("status") if status_value == "unsubscribed": subscriber.status = "unsubscribed" subscriber.unsubscribe_on = timezone.now() elif status_value == "subscribed": subscriber.status = "subscribed" subscriber.unsubscribe_on = None subscriber.save() return Response({ "message": "Status updated", "status": subscriber.status, "unsubscribe_on": subscriber.unsubscribe_on }) import secrets from datetime import timedelta from django.utils import timezone from django.core.mail import send_mail from django.conf import settings class NewsletterSendOTPAPIView(APIView): permission_classes = [AllowAny] def post(self, request): email = request.data.get("email") if not email: return Response( {"message": "Email is required."}, status=status.HTTP_400_BAD_REQUEST ) # Check if already subscribed subscriber = NewsletterSubscriber.objects.filter( email=email, status="subscribed", is_active=True ).first() if subscriber: return Response( {"message": "This email is already subscribed."}, status=status.HTTP_400_BAD_REQUEST ) # Generate secure 6-digit OTP otp = str(secrets.randbelow(900000) + 100000) NewsletterOTP.objects.update_or_create( email=email, defaults={ "otp": otp, "verified": False, "expires_at": timezone.now() + timedelta(minutes=10) } ) send_mail( subject="Verify Your Email", message=f"Your OTP for joining the Indizene Club is: {otp}\n\nThis OTP is valid for 10 minutes.", from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[email], fail_silently=False, ) return Response( {"message": "OTP sent successfully."}, status=status.HTTP_200_OK ) from django.utils import timezone class NewsletterVerifyOTPAPIView(APIView): permission_classes = [AllowAny] def post(self, request): email = request.data.get("email") otp = request.data.get("otp") if not email or not otp: return Response( {"message": "Email and OTP are required."}, status=status.HTTP_400_BAD_REQUEST ) try: otp_obj = NewsletterOTP.objects.get(email=email) except NewsletterOTP.DoesNotExist: return Response( {"message": "OTP not found. Please request a new OTP."}, status=status.HTTP_404_NOT_FOUND ) # Check expiry if timezone.now() > otp_obj.expires_at: otp_obj.delete() return Response( {"message": "OTP has expired. Please request a new OTP."}, status=status.HTTP_400_BAD_REQUEST ) # Check OTP if otp_obj.otp != otp: return Response( {"message": "Invalid OTP."}, status=status.HTTP_400_BAD_REQUEST ) # Mark verified otp_obj.verified = True otp_obj.save() return Response( { "message": "Email verified successfully." }, status=status.HTTP_200_OK ) class UnsubscribedUsersAPIView(APIView): permission_classes = [AllowAny] def get(self, request): users = NewsletterSubscriber.objects.filter( status="unsubscribed" ).order_by("-unsubscribe_on") data = [] for user in users: data.append({ "id": user.id, "email": user.email, "source": user.source, "subscribed_on": user.subscribed_on, "unsubscribe_on": user.unsubscribe_on, "ip_address": user.ip_address, "city": user.city, "country": user.country, }) return Response({ "total_unsubscribed": users.count(), "results": data }) class ContactMessageAPIView(APIView): module_name = "contact_message" def get_permissions(self): if self.request.method == "POST": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): messages = ContactMessage.objects.filter(is_active=True).order_by('-submitted_on') serializer = ContactMessageSerializer(messages, many=True) return Response(serializer.data) def post(self, request): serializer = ContactMessageSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response( serializer.data, status=status.HTTP_201_CREATED ) return Response(serializer.errors, status=400) def delete(self, request, message_id): message = get_object_or_404(ContactMessage, id=message_id) message.is_active = False message.save() return Response({"message": "Message deleted successfully."}, status=204) import json class contactmessagedeletebulk(APIView): module_name = "contact_message" def get_permissions(self): return [IsAuthenticated(), RBACPermission()] def delete(self, request): ids = request.data.get('ids') if not ids: try: body = json.loads(request.body) ids = body.get('ids', []) except: ids = [] if not ids: return Response( {"error": "No IDs provided."}, status=status.HTTP_400_BAD_REQUEST ) messages = ContactMessage.objects.filter(id__in=ids, is_active=True) updated_count = messages.update(is_active=False) return Response( {"message": f"{updated_count} messages deleted successfully."}, status=status.HTTP_200_OK ) #--------------------------------------------Message to Subscribers API View--------------------------------------------# class MessagesToSubscribersAPIView(APIView): module_name = "message_to_subscribers" def get_permissions(self): return [IsAuthenticated(), RBACPermission()] def get(self, request): messages = MessagesToSubscribers.objects.all().order_by('-sent_on') serializer = MessagesToSubscribersSerializer(messages, many=True) return Response(serializer.data) def post(self, request): serializer = MessagesToSubscribersSerializer(data=request.data) if serializer.is_valid(): message_obj = serializer.save() subscribers = NewsletterSubscriber.objects.filter( is_active=True, status="subscribed" ).values_list('email', flat=True) email_list = list(subscribers) if email_list: email = EmailMessage( subject=message_obj.subject, body=message_obj.content, bcc=email_list ) # Attach file if exists if message_obj.attachment: email.attach_file(message_obj.attachment.path) email.send(fail_silently=False) return Response({ "message": "Message created and sent successfully", "data": serializer.data, "total_sent": len(email_list) }, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=400) #-------------------------------------------Page Manager API View-------------------------------------------# #-----------------------------Arrival Products API View----------------------------- class ArrivalProductsAPI(APIView): module_name = "website_new_arrivals" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): products = Product.objects.filter(is_active=True, is_draft=False).order_by('-updated_at') serializer = ProductSerializer(products, many=True) return Response(serializer.data) def patch(self, request, product_id): try: product = Product.objects.get(id=product_id, is_active=True, is_draft=False) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) is_arrival = request.data.get('is_arrival') if is_arrival is not None: product.is_arrival = is_arrival product.save() return Response({"message": "Product Arrival status updated successfully."}) return Response({"error": "is_arrival field is required."}, status=400) class SpecialOfferProductAPI(APIView): module_name = "website_special_offers" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): products = Product.objects.filter( is_active=True, is_draft=False ).order_by('-updated_at') serializer = SpecialOfferProductSerializer(products, many=True) return Response(serializer.data, status=status.HTTP_200_OK) def patch(self, request, product_id): try: product = Product.objects.get(id=product_id, is_active=True, is_draft=False) except Product.DoesNotExist: return Response( {"error": "Product not found"}, status=status.HTTP_404_NOT_FOUND ) serializer = SpecialOfferUpdateSerializer( product, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response({ "message": "Special offer updated successfully", "data": serializer.data }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) #-----------------------------Featured Products API View----------------------------- class featuredProductsAPI(APIView): module_name = "website_featured_products" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): products = Product.objects.filter(is_active=True, is_draft=False).order_by('-updated_at') serializer = ProductSerializer(products, many=True) return Response(serializer.data) def patch(self, request, product_id): try: product = Product.objects.get(id=product_id, is_active=True, is_draft=False) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) is_featured = request.data.get('is_featured') if is_featured is not None: product.is_featured = is_featured product.save() return Response({"message": "Product Featured successfully."}) return Response({"error": "is_featured field is required."}, status=400) #----------------Collection List API View---------------- class CollectionListAPIView(APIView): def get(self, request, collection_id=None): if collection_id: collection = CollectionList.objects.filter(id=collection_id, is_active=True).first() if not collection: return Response({"error": "Collection not found"}, status=404) serializer = CollectionListSerializer(collection) return Response(serializer.data) collections = CollectionList.objects.filter(is_active=True) serializer = CollectionListSerializer(collections, many=True) return Response(serializer.data) def post(self, request): serializer = CollectionListSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, collection_id): collection = CollectionList.objects.filter(id=collection_id).first() if not collection: return Response({"error": "Collection not found"}, status=404) serializer = CollectionListSerializerPOST(collection, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, collection_id): collection = CollectionList.objects.filter(id=collection_id).first() if not collection: return Response({"error": "Collection not found"}, status=404) collection.is_active = False collection.save() return Response(status=204) #----------------Band Story API View---------------- class BandStoryAPIView(APIView): def get(self, request, story_id=None): if story_id: story = BandStory.objects.filter(id=story_id, is_active=True).first() if not story: return Response({"error": "Story not found"}, status=404) serializer = BandStorySerializer(story) return Response(serializer.data) stories = BandStory.objects.filter(is_active=True) serializer = BandStorySerializer(stories, many=True) return Response(serializer.data) def post(self, request): serializer = BandStorySerializerPOST(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, story_id): story = BandStory.objects.filter(id=story_id).first() if not story: return Response({"error": "Story not found"}, status=404) serializer = BandStorySerializerPOST(story, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, story_id): story = BandStory.objects.filter(id=story_id).first() if not story: return Response({"error": "Story not found"}, status=404) story.is_active = False story.save() return Response(status=204) #----------------Marketplace Link API View---------------- class MarketplaceLinkAPIView(APIView): module_name = "website_marketplace_links" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, link_id=None): if link_id: link = MarketplaceLink.objects.filter(id=link_id, is_active=True, place="").first() if not link: return Response({"error": "Link not found"}, status=404) serializer = MarketplaceLinkSerializer(link) return Response(serializer.data) links = MarketplaceLink.objects.filter(is_active=True, place="") serializer = MarketplaceLinkSerializer(links, many=True) return Response(serializer.data) def post(self, request): serializer = MarketplaceLinkSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, link_id): link = MarketplaceLink.objects.filter(id=link_id).first() if not link: return Response({"error": "Link not found"}, status=404) serializer = MarketplaceLinkSerializerPOST(link, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, link_id): link = MarketplaceLink.objects.filter(id=link_id).first() if not link: return Response({"error": "Link not found"}, status=404) link.is_active = False link.save() return Response(status=204) #----------------Featured Marketplace------------- class FeaturedMarketplaceAPIView(APIView): def get(self, request): obj = MarketplaceLink.objects.filter(is_active=True) serializer = MarketplaceLinkSerializer(obj, many=True) return Response(serializer.data) def patch(self, request, link_id): obj = MarketplaceLink.objects.filter(id=link_id, is_active=True).first() if not obj: return Response({"message":"No Such Record exists"}) is_featured = request.data.get('is_featured') if is_featured is not None: obj.is_featured = is_featured obj.save() return Response({"message": "Product Featured successfully."}) return Response({"error": "is_featured field is required."}, status=400) #----------------Instagram Gallery ---------------- class InstagramGalleryAPIView(APIView): module_name = "website_instagram_gallery" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, gallery_id=None): if gallery_id: item = Instagramgallery.objects.filter(id=gallery_id, is_active=True, place="").first() if not item: return Response({"error": "Item not found"}, status=404) serializer = InstagramGallerySerializer(item) return Response(serializer.data) items = Instagramgallery.objects.filter(is_active=True, place="") serializer = InstagramGallerySerializer(items, many=True) return Response(serializer.data) def post(self, request): serializer = InstagramGallerySerializerPOST(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, gallery_id): item = Instagramgallery.objects.filter(id=gallery_id).first() if not item: return Response({"error": "Item not found"}, status=404) serializer = InstagramGallerySerializerPOST(item, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, gallery_id): item = Instagramgallery.objects.filter(id=gallery_id).first() if not item: return Response({"error": "Item not found"}, status=404) item.is_active = False item.save() return Response(status=204) #-------------------------------------------- Hero Banner API View--------------------------------------------# class HeroBannerPageAPIView(APIView): module_name = "website_hero_banners" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): banners = Herobanner.objects.filter(is_active=True, place="page").order_by('-updated_at') serializer = HeroBannerPageSerializer(banners, many=True) return Response(serializer.data) def post(self, request): serializer = HeroBannerPageSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save(is_active=True, place="page") return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, banner_id): banner = Herobanner.objects.filter(id=banner_id, place="page").first() if not banner: return Response({"error": "Banner not found"}, status=404) serializer = HeroBannerPageSerializerPOST(banner, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, banner_id): banner = Herobanner.objects.filter(id=banner_id, place="page").first() if not banner: return Response({"error": "Banner not found"}, status=404) banner.is_active = False banner.save() return Response(status=204) #-----------------------------------------------------Store Manager------------------------------------------------------- #-----------------------------Arrival Products API View----------------------------- class ArrivalProductsStoreAPI(APIView): module_name = "store_new_arrivals" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): products = Product.objects.filter(is_active=True, is_draft=False).order_by('-updated_at') serializer = ProductSerializer(products, many=True) return Response(serializer.data) def patch(self, request, product_id): try: product = Product.objects.get(id=product_id, is_active=True, is_draft=False) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) is_arrival_store = request.data.get('is_arrival_store') if is_arrival_store is not None: product.is_arrival_store = is_arrival_store product.save() return Response({"message": "Product Arrival status updated successfully."}) return Response({"error": "is_arrival_store field is required."}, status=400) class SpecialOfferProductStoreAPI(APIView): module_name = "store_special_offers" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): products = Product.objects.filter( is_active=True, is_draft=False ).order_by('-updated_at') serializer = SpecialOfferProductSerializer(products, many=True) return Response(serializer.data, status=status.HTTP_200_OK) def patch(self, request, product_id): try: product = Product.objects.get(id=product_id, is_active=True, is_draft=False) except Product.DoesNotExist: return Response( {"error": "Product not found"}, status=status.HTTP_404_NOT_FOUND ) serializer = SpecialOfferUpdateSerializerStore( product, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response({ "message": "Special offer updated successfully", "data": serializer.data }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) #-----------------------------Featured Products API View----------------------------- class featuredProductsStoreAPI(APIView): module_name = "store_featured_products" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): products = Product.objects.filter(is_active=True, is_draft=False).order_by('-updated_at') serializer = ProductSerializer(products, many=True) return Response(serializer.data) def patch(self, request, product_id): try: product = Product.objects.get(id=product_id, is_active=True, is_draft=False) except Product.DoesNotExist: return Response({"error": "Product not found."}, status=404) is_featured_store = request.data.get('is_featured_store') if is_featured_store is not None: product.is_featured_store = is_featured_store product.save() return Response({"message": "Product Featured successfully."}) return Response({"error": "is_featured_store field is required."}, status=400) #----------------Marketplace Link API View---------------- class MarketplaceLinkStoreAPIView(APIView): module_name = "store_marketplace_links" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, link_id=None): if link_id: link = MarketplaceLink.objects.filter(id=link_id, is_active=True, place="store").first() if not link: return Response({"error": "Link not found"}, status=404) serializer = MarketplaceLinkSerializer(link) return Response(serializer.data) links = MarketplaceLink.objects.filter(is_active=True, place="store") serializer = MarketplaceLinkSerializer(links, many=True) return Response(serializer.data) def post(self, request): serializer = MarketplaceLinkSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save(place="store", is_active=True) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, link_id): link = MarketplaceLink.objects.filter(id=link_id, place="store").first() if not link: return Response({"error": "Link not found"}, status=404) serializer = MarketplaceLinkSerializerPOST(link, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, link_id): link = MarketplaceLink.objects.filter(id=link_id).first() if not link: return Response({"error": "Link not found"}, status=404) link.is_active = False link.save() return Response(status=204) #----------------Instagram Gallery ---------------- class InstagramGalleryStoreAPIView(APIView): module_name = "store_instagram_gallery" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, gallery_id=None): if gallery_id: item = Instagramgallery.objects.filter(id=gallery_id, is_active=True, place="store").first() if not item: return Response({"error": "Item not found"}, status=404) serializer = InstagramGallerySerializer(item) return Response(serializer.data) items = Instagramgallery.objects.filter(is_active=True, place="store") serializer = InstagramGallerySerializer(items, many=True) return Response(serializer.data) def post(self, request): serializer = InstagramGallerySerializerPOST(data=request.data) if serializer.is_valid(): serializer.save(is_active=True, place="store") return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, gallery_id): item = Instagramgallery.objects.filter(id=gallery_id, place="store").first() if not item: return Response({"error": "Item not found"}, status=404) serializer = InstagramGallerySerializerPOST(item, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, gallery_id): item = Instagramgallery.objects.filter(id=gallery_id).first() if not item: return Response({"error": "Item not found"}, status=404) item.is_active = False item.save() return Response(status=204) class HeroBannerStoreAPIView(APIView): module_name = "store_hero_banners" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): banners = Herobanner.objects.filter(is_active=True, place="store").order_by('-updated_at') serializer = HeroBannerPageSerializer(banners, many=True) return Response(serializer.data) def post(self, request): serializer = HeroBannerPageSerializerPOST(data=request.data) if serializer.is_valid(): serializer.save(is_active=True, place="store") return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, banner_id): banner = Herobanner.objects.filter(id=banner_id, place="store").first() if not banner: return Response({"error": "Banner not found"}, status=404) serializer = HeroBannerPageSerializerPOST(banner, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, banner_id): banner = Herobanner.objects.filter(id=banner_id, place="store").first() if not banner: return Response({"error": "Banner not found"}, status=404) banner.is_active = False banner.save() return Response(status=204) class PromiseAPIView(APIView): module_name = "promises" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): promises = Promises.objects.filter(is_active=True).order_by('-updated_at') serializer = PromisesSerializer(promises, many=True) return Response(serializer.data) def post(self, request): serializer = PromisesSerializer(data=request.data) if serializer.is_valid(): serializer.save(is_active=True) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, promise_id): promise = Promises.objects.filter(id=promise_id).first() if not promise: return Response({"error": "Promise not found"}, status=404) serializer = PromisesSerializer(promise, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, promise_id): promise = Promises.objects.filter(id=promise_id).first() if not promise: return Response({"error": "Promise not found"}, status=404) promise.is_active = False promise.save() return Response(status=204) from django.conf import settings from django.core.mail import send_mail from django.utils import timezone from datetime import timedelta from .models import LeadOTP class LeadCreateAPI(APIView): module_name = "lead" def get_permissions(self): if self.request.method == "POST": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): data = Lead.objects.all() serializer = LeadSerializer(data, many=True) return Response(serializer.data) def post(self, request): data = request.data.copy() # Get Client Information ip = get_client_ip(request) city, country = get_location(ip) user_agent = request.META.get("HTTP_USER_AGENT") data["ip_address"] = ip data["city"] = city data["country"] = country data["user_agent"] = user_agent email = data.get("email") if not email: return Response( { "status": False, "message": "Email is required for OTP verification." }, status=status.HTTP_400_BAD_REQUEST ) # Optional: Prevent OTP spam (1 OTP per minute) existing = LeadOTP.objects.filter(email=email).order_by("-created_at").first() if existing and timezone.now() < existing.created_at + timedelta(seconds=5): return Response( { "status": False, "message": "Please wait 60 seconds before requesting another OTP." }, status=status.HTTP_429_TOO_MANY_REQUESTS ) # Remove previous unverified OTPs LeadOTP.objects.filter(email=email, is_verified=False).delete() # Create OTP record (OTP generated automatically in model) otp_record = LeadOTP.objects.create( email=email, lead_data=data ) # Send OTP Email send_mail( subject="Email Verification OTP", message=f""" Hello, Your OTP is: {otp_record.otp} This OTP is valid for 5 minutes. If you did not request this OTP, please ignore this email. Regards, Your Team """, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[email], fail_silently=False, ) return Response( { "status": True, "message": "OTP sent successfully.", "email": email }, status=status.HTTP_200_OK ) from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from rest_framework.permissions import AllowAny from .models import LeadOTP, NewsletterSubscriber from .serializers import LeadSerializer from django.core.mail import EmailMultiAlternatives import logging logger = logging.getLogger(__name__) from django.db import IntegrityError from django.db import transaction class VerifyLeadOTPAPIView(APIView): permission_classes = [AllowAny] @transaction.atomic def post(self, request): try: email = request.data.get("email") otp = request.data.get("otp") if not email or not otp: return Response( { "status": False, "message": "Email and OTP are required." }, status=status.HTTP_400_BAD_REQUEST ) # Find OTP otp_record = LeadOTP.objects.filter( email=email, otp=otp, is_verified=False ).first() if not otp_record: return Response( { "status": False, "message": "Invalid OTP." }, status=status.HTTP_400_BAD_REQUEST ) # Check expiry if otp_record.is_expired: otp_record.delete() return Response( { "status": False, "message": "OTP has expired. Please request a new OTP." }, status=status.HTTP_400_BAD_REQUEST ) # Get stored form data data = otp_record.lead_data # Save Lead serializer = LeadSerializer(data=data) if not serializer.is_valid(): return Response( { "status": False, "errors": serializer.errors }, status=status.HTTP_400_BAD_REQUEST ) lead = serializer.save() # --------------------------- # Your existing coupon email code # --------------------------- email = data.get("email") whatsapp = data.get("whatsapp_number") source = data.get("source", "lead_capture") modal_setting = ModalSetting.objects.filter( strategy=source.removeprefix("promotion_modal_") ).select_related("coupon").first() coupon_name = None coupon_code = None coupon_offer_value = None if modal_setting and modal_setting.coupon: coupon_name = modal_setting.coupon.name coupon_code = modal_setting.coupon.coupon_code coupon_offer_value = modal_setting.coupon.offer_value logger.error("Email: %s, Coupon: %s Source: %s", email, coupon_name, source) if email and coupon_code: subject = f"🎉 {coupon_name}" text_content = f""" Congratulations! Thank you for registering. Your exclusive coupon is ready. Coupon Name : {coupon_name} Coupon Code : {coupon_code} Offer : {coupon_offer_value} Thank you, Indizene """ html_content = f""" {subject}

Indizene

Your Exclusive Coupon is Ready 🎉

Congratulations!

Thank you for registering with Indizene.

We've prepared an exclusive coupon especially for you.

Coupon {coupon_name}
Coupon Code {coupon_code}
Offer {coupon_offer_value}
{coupon_code}

Use this coupon during checkout to enjoy your exclusive discount.

Thank you for choosing Indizene.

This coupon is exclusively issued for your account. Please keep it safe and use it before it expires.

© Indizene. All rights reserved.

""" email_message = EmailMultiAlternatives( subject=subject, body=text_content, from_email=settings.DEFAULT_FROM_EMAIL, to=[email], ) email_message.attach_alternative(html_content, "text/html") email_message.send(fail_silently=False) # --------------------------- # Newsletter Subscriber # --------------------------- if whatsapp and email: NewsletterSubscriber.objects.update_or_create( email=email, defaults={ "mobile_number": whatsapp, "source": source, "subscriber_type": "whatsapp & Email", "status": "subscribed", "is_active": True, "ip_address": data.get("ip_address"), "city": data.get("city"), "country": data.get("country"), "user_agent": data.get("user_agent"), } ) elif email: NewsletterSubscriber.objects.update_or_create( email=email, defaults={ "source": source, "subscriber_type": "email", "status": "subscribed", "is_active": True, "ip_address": data.get("ip_address"), "city": data.get("city"), "country": data.get("country"), "user_agent": data.get("user_agent"), } ) elif whatsapp: NewsletterSubscriber.objects.update_or_create( mobile_number=whatsapp, defaults={ "source": source, "subscriber_type": "whatsapp", "status": "subscribed", "is_active": True, "ip_address": data.get("ip_address"), "city": data.get("city"), "country": data.get("country"), "user_agent": data.get("user_agent"), } ) otp_record.is_verified = True otp_record.save() return Response( { "status": True, "message": "OTP verified successfully.", "lead_id": lead.id if hasattr(lead, "id") else None }, status=status.HTTP_201_CREATED ) except IntegrityError: return Response( { "status": False, "message": "This email is already subscribed." }, status=status.HTTP_400_BAD_REQUEST ) except Exception as e: return Response( { "status": False, "message": "Something went wrong.", "error": str(e) # Remove this line in production }, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) #-----------------------------------About Us------------------------------------- class DescriptionAboutUsAPIView(APIView): def get(self, request): description = DescriptionAboutUs.objects.filter(is_active=True).first() serializer = DescriptionAboutUsSerializer(description) return Response(serializer.data) def post(self, request): serializer = DescriptionAboutUsSerializer(data=request.data) if serializer.is_valid(): DescriptionAboutUs.objects.filter(is_active=True).update(is_active=False) serializer.save(is_active=True) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, description_id): description = DescriptionAboutUs.objects.filter(id=description_id).first() if not description: return Response({"error": "Description not found"}, status=404) serializer = DescriptionAboutUsSerializer(description, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, description_id): description = DescriptionAboutUs.objects.filter(id=description_id).first() if not description: return Response({"error": "Description not found"}, status=404) description.is_active = False description.save() return Response(status=204) class TeamMemberAPIView(APIView): module_name = "about_us" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): members = TeamMember.objects.filter(is_active=True).order_by('-updated_at') serializer = TeamMemberSerializer(members, many=True) return Response(serializer.data) def post(self, request): serializer = TeamMemberSerializer(data=request.data) if serializer.is_valid(): serializer.save(is_active=True) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, member_id): member = TeamMember.objects.filter(id=member_id).first() if not member: return Response({"error": "Team member not found"}, status=404) serializer = TeamMemberSerializer(member, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, member_id): member = TeamMember.objects.filter(id=member_id).first() if not member: return Response({"error": "Team member not found"}, status=404) member.is_active = False member.save() return Response(status=204) class FactorsAPIView(APIView): def get(self, request): factors = Factors.objects.filter(is_active=True).order_by('-updated_at') serializer = FactorsSerializer(factors, many=True) return Response(serializer.data) def post(self, request): serializer = FactorsSerializer(data=request.data) if serializer.is_valid(): Factors.objects.filter(is_active=True).update(is_active=False) serializer.save(is_active=True) return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, factor_id): factor = Factors.objects.filter(id=factor_id).first() if not factor: return Response({"error": "Factor not found"}, status=404) serializer = FactorsSerializer(factor, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, factor_id): factor = Factors.objects.filter(id=factor_id).first() if not factor: return Response({"error": "Factor not found"}, status=404) factor.is_active = False factor.save() return Response(status=204) #-----------------------------------Recent Activity----------------------------------- class RecentActivityView(APIView): module_name = "recent_activity" def get_permissions(self): return [IsAuthenticated(), RBACPermission()] def get(self, request): activities = [] # Product Category for item in ProductCategory.objects.order_by('-id')[:3]: activities.append({ "type": "category", "title": "Category Added", "sub": item.name, "time": None # no timestamp field }) # Product for item in Product.objects.order_by('-created_at')[:5]: activities.append({ "type": "product", "title": "Product Added", "sub": item.product_name, "time": item.created_at }) # Tag for item in Tag.objects.order_by('-created_on')[:3]: activities.append({ "type": "tag", "title": "Tag Created", "sub": item.tag, "time": item.created_on }) # Blog for item in Blog.objects.order_by('-posted_on')[:5]: activities.append({ "type": "blog", "title": "Blog Published", "sub": item.title, "time": item.posted_on }) # Testimonial for item in Testimonial.objects.order_by('-posted_on')[:5]: activities.append({ "type": "testimonial", "title": "New Testimonial", "sub": item.name, "time": item.posted_on }) # Subscriber for item in NewsletterSubscriber.objects.order_by('-subscribed_on')[:5]: activities.append({ "type": "subscriber", "title": "New Subscriber", "sub": item.email, "time": item.subscribed_on }) # Contact Message for item in ContactMessage.objects.order_by('-submitted_on')[:5]: activities.append({ "type": "contact", "title": "New Contact Message", "sub": item.customer_name, "time": item.submitted_on }) # Marketplace Link for item in MarketplaceLink.objects.order_by('-id')[:3]: activities.append({ "type": "marketplace_link", "title": "Marketplace Link Added", "sub": item.name, "time": None }) # Messages to Subscribers for item in MessagesToSubscribers.objects.order_by('-sent_on')[:5]: activities.append({ "type": "email", "title": "Message Sent to Subscribers", "sub": item.subject, "time": item.sent_on }) # Promises for item in Promises.objects.order_by('-updated_at')[:3]: activities.append({ "type": "promise", "title": "Promise Updated", "sub": item.title, "time": item.updated_at }) # Hero Banner for item in Herobanner.objects.order_by('-created_at')[:3]: activities.append({ "type": "banner", "title": "Hero Banner Added", "sub": item.place, "time": item.created_at }) # Instagram Gallery for item in Instagramgallery.objects.order_by('-created_at')[:5]: activities.append({ "type": "instagram", "title": "Instagram Post Added", "sub": item.title, "time": item.created_at }) for item in DescriptionAboutUs.objects.order_by('-created_at')[:3]: activities.append({ "type": "about", "title": "About Section Updated", "sub": item.title, "time": item.created_at }) activities = [a for a in activities if a["time"] is not None] activities = sorted(activities, key=lambda x: x["time"], reverse=True) return Response(activities[:15]) #-----------------------------------Client Modification------------------------------------------------- class LogoTextAPI(APIView): module_name = "branding" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): obj = LogoText.objects.first() serializer = LogoTextSerializer(obj) return Response(serializer.data) def post(self, request): serializer = LogoTextSerializer(data=request.data) if serializer.is_valid(): LogoText.objects.all().delete() serializer.save() return Response(serializer.data) return Response(serializer.errors) def patch(self, request, pk): obj = LogoText.objects.get(pk=pk) if not obj: return Response({"message":"No data Found"}) serializer = LogoTextSerializer(obj, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.data) def delete(Self, request, pk): obj = LogoText.objects.get(pk=pk) if not obj: return Response({"message":"No data Found"}) obj.delete() return Response(status=201) #Satisfied Customers class Satisfied_CustomerAPIView(APIView): module_name = "website_settings" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): obj = Satisfied_Customer.objects.first() serializer = Satisfied_CustomerSerializer(obj) return Response(serializer.data) def post(self, request): serializer = Satisfied_CustomerSerializer(data=request.data) if serializer.is_valid(): Satisfied_Customer.objects.all().delete() serializer.save() return Response({"message":"facing problem to update"}) class ContactInfoAPI(APIView): module_name = "website_settings" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): obj = ContactInfo.objects.first() serializer = ContactInfoSerializer(obj) return Response(serializer.data) def post(Self, request): serializer = ContactInfoSerializer(data=request.data) if serializer.is_valid(): ContactInfo.objects.all().delete() serializer.save() return Response(serializer.data) return Response(serializer.errors) class SocialLinksAPIView(APIView): module_name = "website_settings" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): objs = SocialLinks.objects.all() serializer = SocialLinksSerializer(objs, many=True) return Response(serializer.data) def post(self, request): serializer = SocialLinksSerializer(data=request.data) if serializer.is_valid(): platform = request.data.get('platform') # Delete any existing link with the same platform to keep only one per platform SocialLinks.objects.filter(platform=platform).delete() serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def delete(self, request, pk): try: obj = SocialLinks.objects.get(pk=pk) obj.delete() return Response(status=204) except SocialLinks.DoesNotExist: return Response({"error": "Not found"}, status=404) from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from .models import Widget from .serializers import WidgetSerializer class WidgetAPIView(APIView): module_name = "website_settings" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get_object(self, pk): try: return Widget.objects.get(pk=pk) except Widget.DoesNotExist: return None # 🔵 GET (All OR Single) def get(self, request): pk = request.query_params.get('id') # 👉 GET SINGLE if pk: widget = self.get_object(pk) if not widget: return Response({"error": "Not found"}, status=404) serializer = WidgetSerializer(widget) return Response(serializer.data) # 👉 GET ALL widgets = Widget.objects.all() serializer = WidgetSerializer(widgets, many=True) return Response(serializer.data) # 🟢 POST (Create OR Append) def post(self, request): name = request.data.get("name") new_data = request.data.get("data", []) if not name: return Response({"error": "Name is required"}, status=400) if not isinstance(new_data, list): return Response({"error": "Data must be a list"}, status=400) widget = Widget.objects.filter(name=name).first() # 👉 If exists → append if widget: existing_links = widget.data or [] existing_set = {(i["label"], i["url"]) for i in existing_links} for item in new_data: if (item["label"], item["url"]) not in existing_set: existing_links.append(item) widget.data = existing_links widget.save() return Response({ "message": "Data appended", "data": WidgetSerializer(widget).data }) # 👉 Else → create serializer = WidgetSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response({ "message": "Created", "data": serializer.data }, status=201) return Response(serializer.errors, status=400) # 🟡 PATCH (Update) def patch(self, request): pk = request.query_params.get('id') if not pk: return Response({"error": "ID is required"}, status=400) widget = self.get_object(pk) if not widget: return Response({"error": "Not found"}, status=404) serializer = WidgetSerializer(widget, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response({ "message": "Updated", "data": serializer.data }) return Response(serializer.errors, status=400) # 🔴 DELETE def delete(self, request): pk = request.query_params.get('id') if not pk: return Response({"error": "ID is required"}, status=400) widget = self.get_object(pk) if not widget: return Response({"error": "Not found"}, status=404) widget.delete() return Response({"message": "Deleted"}, status=204) #-----------------------------------Admin Product Management API View-----------------------------------# class CreateRoleView(APIView): #Only Super Admin module_name = "role_management" def get_permissions(self): return [IsAuthenticated(), RBACPermission()] def post(self, request): data = request.data role = Role.objects.create( name=data.get('name'), description=data.get('description') ) permissions = data.get('permissions', []) role.permissions.set(permissions) return Response({"message": "Role created","id":role.id}) #EDIT =========================Soumya============================ class CreateAdminView(APIView): #Only Super Admin module_name = "role_management" def get_permissions(self): return [IsAuthenticated(), RBACPermission()] def post(self, request): # Support both JSON and FormData (multipart) if request.content_type and 'multipart' in request.content_type: data = request.POST avatar = request.FILES.get('avatar') else: data = request.data avatar = None username = data.get('username') email = data.get('email') password = data.get('password') role_id = data.get('role_id') if not username or not email or not password: return Response( {"error": "Username, email and password are required."}, status=status.HTTP_400_BAD_REQUEST ) try: role = Role.objects.get(id=role_id) except Role.DoesNotExist: return Response( {"error": f"Role with id {role_id} does not exist."}, status=status.HTTP_400_BAD_REQUEST ) try: user = AdminUser.objects.create_user( username=username, email=email, password=password, ) user.role = role if avatar: user.avatar = avatar user.save() except IntegrityError as e: return Response( {"error": str(e)}, status=status.HTTP_400_BAD_REQUEST ) except Exception as e: return Response( {"error": f"User creation failed: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST ) return Response( {"message": "Admin created successfully", "id": user.id}, status=status.HTTP_201_CREATED ) from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from admin_panel.models import Permission from admin_panel.serializers import PermissionSerializer from rest_framework.permissions import IsAuthenticated from admin_panel.rbac import RBACPermission class PermissionListView(APIView): #only super admin module_name = "role_management" def get_permissions(self): return [IsAuthenticated(), RBACPermission()] # ✅ GET → List permissions (grouped) def get(self, request): permissions = Permission.objects.all() data = {} for p in permissions: if p.module not in data: data[p.module] = [] data[p.module].append({ "id": p.id, "action": p.action }) return Response(data) # ✅ POST → Create new permission def post(self, request): serializer = PermissionSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) # ✅ PATCH → Update permission def patch(self, request, pk): try: permission = Permission.objects.get(pk=pk) except Permission.DoesNotExist: return Response({"error": "Permission not found"}, status=404) serializer = PermissionSerializer(permission, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) # ✅ DELETE → Delete permission def delete(self, request): try: permission = Permission.objects.all() except Permission.DoesNotExist: return Response({"error": "Permission not found"}, status=404) permission.delete() return Response({"message": "Deleted successfully"}, status=204) # ==================== LIST ROLES AND ADMINS SOUMYA ==================== class RoleListView(APIView): """List all roles (for the admin panel)""" permission_classes = [IsAuthenticated] # Only authenticated admins def get(self, request): roles = Role.objects.all() data = [] for role in roles: data.append({ 'id': role.id, 'name': role.name, 'description': role.description, 'permissions': [ {'id': p.id, 'key': f"{p.module}.{p.action}"} for p in role.permissions.all() ], 'user_count': role.adminuser_set.count(), # adjust if your related name is different }) return Response(data) class AdminUserListView(APIView): permission_classes = [IsAuthenticated] def get(self, request): admins = AdminUser.objects.all() data = [] for admin in admins: avatar_url = None if admin.avatar: # Build absolute URL – update domain if needed avatar_url = request.build_absolute_uri(admin.avatar.url) data.append({ 'id': admin.id, 'username': admin.username, 'email': admin.email, 'role': admin.role.id if admin.role else None, 'last_login': admin.last_login, 'last_active': admin.last_active, 'is_active': admin.is_active, 'avatar': avatar_url, # ✅ add this line }) return Response(data) class ChangePasswordView(APIView): permission_classes = [IsAuthenticated] def post(self, request): user = request.user current_password = request.data.get('current_password') new_password = request.data.get('new_password') if not current_password or not new_password: return Response( {"error": "Both current and new password are required."}, status=status.HTTP_400_BAD_REQUEST ) # Verify current password if not user.check_password(current_password): return Response( {"error": "Current password is incorrect."}, status=status.HTTP_400_BAD_REQUEST ) # Validate new password length if len(new_password) < 6: return Response( {"error": "New password must be at least 6 characters long."}, status=status.HTTP_400_BAD_REQUEST ) user.set_password(new_password) user.save() return Response( {"message": "Password changed successfully."}, status=status.HTTP_200_OK ) # ==================== UPDATE / DELETE ROLE ==================== class RoleDetailView(APIView): permission_classes = [IsAuthenticated, RBACPermission] module_name = "role_management" def patch(self, request, pk): try: role = Role.objects.get(pk=pk) except Role.DoesNotExist: return Response({"error": "Role not found"}, status=404) data = request.data role.name = data.get('name', role.name) role.description = data.get('description', role.description) if 'permissions' in data: role.permissions.set(data['permissions']) role.save() return Response({"message": "Role updated"}) def delete(self, request, pk): try: role = Role.objects.get(pk=pk) except Role.DoesNotExist: return Response({"error": "Role not found"}, status=404) # Prevent deleting super admin role if needed role.delete() return Response({"message": "Role deleted"}, status=204) # ==================== UPDATE / DELETE ADMIN ==================== class AdminUserDetailView(APIView): permission_classes = [IsAuthenticated, RBACPermission] module_name = "admin" def patch(self, request, pk): try: admin = AdminUser.objects.get(pk=pk) except AdminUser.DoesNotExist: return Response({"error": "Admin not found"}, status=404) # Handle multipart (file) and JSON if request.content_type and 'multipart' in request.content_type: data = request.POST avatar = request.FILES.get('avatar') else: data = request.data avatar = None if 'role' in data: try: role = Role.objects.get(id=data['role']) admin.role = role except Role.DoesNotExist: return Response({"error": "Role not found"}, status=400) if 'is_active' in data: admin.is_active = data['is_active'] in ['true', True, 'True'] if avatar: admin.avatar = avatar admin.save() return Response({"message": "Admin updated"}) def delete(self, request, pk): try: admin = AdminUser.objects.get(pk=pk) except AdminUser.DoesNotExist: return Response({"error": "Admin not found"}, status=404) if admin == request.user: return Response({"error": "Cannot delete your own account"}, status=400) admin.delete() return Response({"message": "Admin deleted"}, status=204) # ==================== CURRENT USER INFO ==================== class CurrentUserView(APIView): permission_classes = [IsAuthenticated] def get(self, request): user = request.user permissions = [] if user.is_super_admin: permissions = ["*"] else: if user.role: permissions = [f"{p.module}.{p.action}" for p in user.role.permissions.all()] avatar_url = None if user.avatar: avatar_url = request.build_absolute_uri(user.avatar.url) return Response({ "id": user.id, "username": user.username, "email": user.email, "is_super_admin": user.is_super_admin, "role": user.role.id if user.role else None, "permissions": permissions, "avatar": avatar_url, # ✅ add this line }) #==================Puskar====================== from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from .models import Coupon, Chat from .serializers import CouponSerializer, ChatSerializer # ------------------------- # Coupon API # ------------------------- class CouponAPI(APIView): module_name = "popup_settings" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): coupons = Coupon.objects.all() serializer = CouponSerializer(coupons, many=True) return Response(serializer.data) def post(self, request): serializer = CouponSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors) def patch(self, request, pk): coupon = Coupon.objects.get(pk=pk) serializer = CouponSerializer(coupon, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors) def delete(self, request, pk): coupon = Coupon.objects.get(pk=pk) coupon.delete() return Response({"message": "Deleted"}) # ------------------------- # Chat API # ------------------------- class ChatAPI(APIView): def get(self, request): chats = Chat.objects.all() serializer = ChatSerializer(chats, many=True) return Response(serializer.data) def post(self, request): # Only Super Admin can send if not request.user.is_superuser: return Response({"error": "Only super admin can send messages"}, status=403) serializer = ChatSerializer(data=request.data) if serializer.is_valid(): serializer.save(sender=request.user) return Response(serializer.data) return Response(serializer.errors) def patch(self, request, pk): chat = Chat.objects.get(pk=pk) serializer = ChatSerializer(chat, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors) def delete(self, request, pk): chat = Chat.objects.get(pk=pk) chat.delete() return Response({"message": "Chat deleted"}) # ========== CHAT MESSENGER VIEWS (add after existing ChatAPI) ========== from django.contrib.auth import get_user_model from django.db.models import Q AdminUser = get_user_model() class ConversationListView(APIView): authentication_classes = [JWTAuthentication] permission_classes = [IsAuthenticated] def get(self, request): user = request.user # Get all distinct chat partners chat_partners = AdminUser.objects.filter( Q(sent_messages__receiver=user) | Q(received_messages__sender=user) ).distinct() # Annotate unread count for each partner (messages sent by partner, not read by current user) unread_counts = Chat.objects.filter( sender=OuterRef('id'), receiver=user, is_read=False ).values('sender').annotate(cnt=Count('id')).values('cnt') chat_partners = chat_partners.annotate( unread_count=Subquery(unread_counts, output_field=models.IntegerField()) ).values('id', 'username', 'email', 'unread_count') return Response(list(chat_partners)) class MessageListView(APIView): """Get all messages between current user and another admin.""" permission_classes = [IsAuthenticated] def get(self, request, other_user_id): user = request.user try: other_user = AdminUser.objects.get(pk=other_user_id) except AdminUser.DoesNotExist: return Response({"error": "User not found"}, status=404) messages = Chat.objects.filter( (Q(sender=user) & Q(receiver=other_user)) | (Q(sender=other_user) & Q(receiver=user)) ).order_by('created_at') from .serializers import ChatSerializer serializer = ChatSerializer(messages, many=True) return Response(serializer.data) class MessageCreateView(APIView): """Send a new message to another admin.""" permission_classes = [IsAuthenticated] def post(self, request): receiver_id = request.data.get('receiver') message_text = request.data.get('message') if not receiver_id or not message_text: return Response({"error": "receiver and message are required"}, status=400) try: receiver = AdminUser.objects.get(pk=receiver_id) except AdminUser.DoesNotExist: return Response({"error": "Receiver not found"}, status=404) # Prevent self‑messaging if receiver == request.user: return Response({"error": "Cannot send message to yourself"}, status=400) chat = Chat.objects.create( sender=request.user, receiver=receiver, message=message_text ) from .serializers import ChatSerializer serializer = ChatSerializer(chat) return Response(serializer.data, status=201) class MarkMessagesReadView(APIView): permission_classes = [IsAuthenticated] def post(self, request, other_user_id): Chat.objects.filter(sender_id=other_user_id, receiver=request.user, is_read=False).update(is_read=True) return Response({"message": "Messages marked as read"}) class UnreadCountView(APIView): permission_classes = [IsAuthenticated] def get(self, request): count = Chat.objects.filter(receiver=request.user, is_read=False).count() return Response({"count": count}) class AllAdminsListView(APIView): permission_classes = [IsAuthenticated] def get(self, request): # Exclude current user admins = AdminUser.objects.exclude(id=request.user.id) serializer = AdminUserSimpleSerializer(admins, many=True) return Response(serializer.data) # Add at the end of views.py from .models import ModalSetting from .serializers import ModalSettingSerializer class ModalSettingAPIView(APIView): permission_classes = [IsAuthenticated, RBACPermission] module_name = "popup_settings" def get(self, request): # Get the first (and only) modal setting row, or create default if none setting, created = ModalSetting.objects.get_or_create(id=1) serializer = ModalSettingSerializer(setting) return Response(serializer.data) def patch(self, request): setting, _ = ModalSetting.objects.get_or_create(id=1) serializer = ModalSettingSerializer(setting, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) # views.py class PublicModalConfigView(APIView): permission_classes = [AllowAny] def get(self, request): settings = ModalSetting.objects.filter(is_enabled=True) data = [] for s in settings: data.append({ 'strategy': s.strategy, 'is_enabled': s.is_enabled, 'trigger_type': s.trigger_type, 'delay_seconds': s.delay_seconds, 'show_once_per_session': s.show_once_per_session, 'coupon_code': s.coupon.coupon_code if s.coupon else None, 'offer_value': s.coupon.offer_value if s.coupon else None, # ✅ ADD THIS }) return Response(data) # views.py class ModalSettingListAPIView(APIView): """ GET /modal-settings/ → list all modal settings GET /modal-settings/WELCOME/ → retrieve (creates if missing) PATCH /modal-settings/WELCOME/ → update (creates if missing) """ module_name = "popup_settings" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, strategy=None): if strategy: # Get or create default record for this strategy setting, created = ModalSetting.objects.get_or_create( strategy=strategy, defaults={ 'is_enabled': False, 'trigger_type': 'page_load', 'delay_seconds': 5, 'show_once_per_session': True, 'coupon': None, } ) serializer = ModalSettingSerializer(setting) return Response(serializer.data) else: settings = ModalSetting.objects.all() serializer = ModalSettingSerializer(settings, many=True) return Response(serializer.data) def patch(self, request, strategy): # Get or create – ensures PATCH never fails with 404 setting, created = ModalSetting.objects.get_or_create( strategy=strategy, defaults={ 'is_enabled': False, 'trigger_type': 'page_load', 'delay_seconds': 5, 'show_once_per_session': True, 'coupon': None, } ) serializer = ModalSettingSerializer(setting, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) from django.utils import timezone from .models import GroupChatMessage, GroupChatReadStatus from .serializers import GroupChatMessageSerializer from rest_framework_simplejwt.authentication import JWTAuthentication class GroupMessageListView(APIView): authentication_classes = [JWTAuthentication] permission_classes = [IsAuthenticated] def get(self, request): # Return last 50 messages, newest last messages = GroupChatMessage.objects.all()[:50] serializer = GroupChatMessageSerializer(messages, many=True) return Response(serializer.data) def post(self, request): message_text = request.data.get('message') if not message_text: return Response({"error": "message is required"}, status=400) group_msg = GroupChatMessage.objects.create( sender=request.user, message=message_text ) serializer = GroupChatMessageSerializer(group_msg) return Response(serializer.data, status=201) class GroupUnreadCountView(APIView): authentication_classes = [JWTAuthentication] permission_classes = [IsAuthenticated] def get(self, request): status_obj, _ = GroupChatReadStatus.objects.get_or_create(user=request.user) unread_count = GroupChatMessage.objects.filter(created_at__gt=status_obj.last_read_time).count() return Response({'count': unread_count}) class GroupMarkReadView(APIView): authentication_classes = [JWTAuthentication] permission_classes = [IsAuthenticated] def post(self, request): status_obj, _ = GroupChatReadStatus.objects.get_or_create(user=request.user) status_obj.last_read_time = timezone.now() status_obj.save() return Response({'status': 'marked read'}) from rest_framework.parsers import MultiPartParser, FormParser from .models import HeaderLogo from .serializers import HeaderLogoSerializer class HeaderLogoListCreateView(APIView): parser_classes = [MultiPartParser, FormParser] module_name = "website_settings" def get_permissions(self): if self.request.method == 'GET': return [AllowAny()] return [IsAuthenticated()] def get(self, request): logos = HeaderLogo.objects.filter(is_active=True) serializer = HeaderLogoSerializer(logos, many=True) return Response(serializer.data) def post(self, request): if HeaderLogo.objects.filter(is_active=True).count() >= 3: return Response({"error": "Maximum 3 active logos allowed"}, status=400) serializer = HeaderLogoSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) class HeaderLogoDetailView(APIView): parser_classes = [MultiPartParser, FormParser] module_name = "website_settings" def get_permissions(self): if self.request.method == 'GET': return [AllowAny()] return [IsAuthenticated()] def patch(self, request, pk): try: logo = HeaderLogo.objects.get(pk=pk) except HeaderLogo.DoesNotExist: return Response({"error": "Not found"}, status=404) serializer = HeaderLogoSerializer(logo, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, pk): try: logo = HeaderLogo.objects.get(pk=pk) except HeaderLogo.DoesNotExist: return Response({"error": "Not found"}, status=404) logo.delete() return Response(status=204) class ProductMarketplaceClicksView(APIView): permission_classes = [AllowAny] # or [IsAuthenticated] if needed def get(self, request, product_id): try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return Response({"error": "Product not found"}, status=404) # Get all platforms for this product platforms = Platform.objects.filter(product=product, is_active=True) # Count clicks per platform result = [] for platform in platforms: click_count = MarketplaceClick.objects.filter(platform=platform).count() result.append({ "platform_id": platform.id, "platform_name": platform.name, "clicks": click_count }) return Response(result) class ProductSEOAPIView(APIView): # ✅ GET → All or by product_id def get(self, request): product_id = request.query_params.get('product_id') if product_id: seo = product_seo.objects.filter(product_id=product_id) else: seo = product_seo.objects.all() serializer = ProductSEOSerializer(seo, many=True) return Response(serializer.data) # ✅ POST → Create SEO def post(self, request): serializer = ProductSEOSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response({ "message": "SEO created successfully", "data": serializer.data }, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class PageSEOAPIView(APIView): def get(self, request, page_name): seo = PageSEO.objects.filter(page=page_name).first() if not seo: return Response( {"error": "SEO page not found"}, status=404 ) serializer = PageSEOSerializer(seo) return Response(serializer.data) def patch(self, request, page_name): seo = PageSEO.objects.filter(page=page_name).first() if not seo: return Response( {"error": "SEO page not found"}, status=404 ) serializer = PageSEOSerializer( seo, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) import random from django.core.mail import send_mail from rest_framework.views import APIView from rest_framework.response import Response class forget_password_email(APIView): def post(self, request): email = request.data.get('email') if not email: return Response({ "error": "Email is required" }, status=400) user = AdminUser.objects.filter( email=email ).first() if not user: return Response({ "error": "User not found" }, status=404) # Generate OTP otp = str(secrets.randbelow(900000) + 100000) # Save OTP user.forgot_otp = otp user.otp_verified = False user.save() # Send email send_mail( subject="Password Reset OTP", message=f"Your OTP is: {otp}", from_email=None, # Uses DEFAULT_FROM_EMAIL recipient_list=[email], fail_silently=False ) return Response({ "message": "OTP sent successfully" }, status=200) class VerifyOTP(APIView): def post(self, request): email = request.data.get('email') otp = request.data.get('otp') user = AdminUser.objects.filter( email=email ).first() if not user: return Response({ "error": "User not found" }, status=404) if user.forgot_otp != otp: return Response({ "error": "Invalid OTP" }, status=400) user.otp_verified = True user.save() return Response({ "message": "OTP verified" }, status=200) class ResetPassword(APIView): def post(self, request): email = request.data.get('email') password = request.data.get('password') user = AdminUser.objects.filter( email=email ).first() if not user: return Response({ "error": "User not found" }, status=404) if not user.otp_verified: return Response({ "error": "OTP not verified" }, status=400) # Change password user.set_password(password) # if inheriting AbstractBaseUser # user.password = make_password(password) # if custom model user.forgot_otp = None user.otp_verified = False user.save() return Response({ "message": "Password reset successfully" }, status=200) class change_password(APIView): permission_classes = [IsAuthenticated] def post(self, request): user = request.user current_password = request.data.get('current_password') new_password = request.data.get('new_password') if not user.check_password(current_password): return Response({ "error": "Current password is incorrect" }, status=400) user.set_password(new_password) user.save() return Response({ "message": "Password changed successfully" }, status=200) class BlogHeroAPIView(APIView): module_name = "blog_banner" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, hero_id=None): if hero_id: hero = BlogHero.objects.filter( id=hero_id, is_active=True ).first() if not hero: return Response( {"error": "Hero not found"}, status=404 ) serializer = BlogHeroSerializer(hero) return Response(serializer.data) heroes = BlogHero.objects.filter( is_active=True ).order_by('-updated_at') serializer = BlogHeroSerializer( heroes, many=True ) return Response(serializer.data) def post(self, request): serializer = BlogHeroSerializer( data=request.data ) if serializer.is_valid(): serializer.save() return Response( serializer.data, status=201 ) return Response(serializer.errors, status=400) def patch(self, request, hero_id): hero = BlogHero.objects.filter( id=hero_id ).first() if not hero: return Response( {"error": "Hero not found"}, status=404 ) serializer = BlogHeroSerializer( hero, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, hero_id): hero = BlogHero.objects.filter( id=hero_id ).first() if not hero: return Response( {"error": "Hero not found"}, status=404 ) hero.is_active = False hero.save() return Response(status=204) class AboutHeroAPIView(APIView): module_name = "about_us_banner" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, hero_id=None): if hero_id: hero = AboutHero.objects.filter( id=hero_id, is_active=True ).first() if not hero: return Response({"error": "Hero not found"}, status=404) serializer = AboutHeroSerializer(hero) return Response(serializer.data) heroes = AboutHero.objects.filter( is_active=True ).order_by('-updated_at') serializer = AboutHeroSerializer( heroes, many=True ) return Response(serializer.data) def post(self, request): serializer = AboutHeroSerializer( data=request.data ) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, hero_id): hero = AboutHero.objects.filter( id=hero_id ).first() if not hero: return Response({"error": "Hero not found"}, status=404) serializer = AboutHeroSerializer( hero, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, hero_id): hero = AboutHero.objects.filter( id=hero_id ).first() if not hero: return Response({"error": "Hero not found"}, status=404) hero.is_active = False hero.save() return Response(status=204) class TestimonialHeroAPIView(APIView): module_name = "testimonial_banner" def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request, hero_id=None): if hero_id: hero = TestimonialHero.objects.filter( id=hero_id, is_active=True ).first() if not hero: return Response({"error": "Hero not found"}, status=404) serializer = TestimonialHeroSerializer(hero) return Response(serializer.data) heroes = TestimonialHero.objects.filter( is_active=True ).order_by('-updated_at') serializer = TestimonialHeroSerializer( heroes, many=True ) return Response(serializer.data) def post(self, request): serializer = TestimonialHeroSerializer( data=request.data ) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, hero_id): hero = TestimonialHero.objects.filter( id=hero_id ).first() if not hero: return Response({"error": "Hero not found"}, status=404) serializer = TestimonialHeroSerializer( hero, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, hero_id): hero = TestimonialHero.objects.filter( id=hero_id ).first() if not hero: return Response({"error": "Hero not found"}, status=404) hero.is_active = False hero.save() return Response(status=204) class AboutCoreValueAPIView(APIView): def get(self, request, value_id=None): if value_id: value = AboutCoreValue.objects.filter( id=value_id, is_active=True ).first() if not value: return Response( {"error": "Core value not found"}, status=404 ) serializer = AboutCoreValueSerializer(value) return Response(serializer.data) values = AboutCoreValue.objects.filter( is_active=True ).order_by('order') serializer = AboutCoreValueSerializer( values, many=True ) return Response(serializer.data) def post(self, request): serializer = AboutCoreValueSerializer( data=request.data ) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=201) return Response(serializer.errors, status=400) def patch(self, request, value_id): value = AboutCoreValue.objects.filter( id=value_id ).first() if not value: return Response( {"error": "Core value not found"}, status=404 ) serializer = AboutCoreValueSerializer( value, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) def delete(self, request, value_id): value = AboutCoreValue.objects.filter( id=value_id ).first() if not value: return Response( {"error": "Core value not found"}, status=404 ) value.is_active = False value.save() return Response(status=204) # ================= New IMplementation ================= # ============================================================ # TRAFFIC SOURCE DETECTION # ============================================================ from urllib.parse import urlparse from urllib.parse import urlparse def detect_traffic_source(request, data): # ======================================================== # 1. UTM SOURCE # ======================================================== source = request.GET.get( "utm_source", "" ).strip().lower() # ======================================================== # 2. SOURCE SENT BY FRONTEND # ======================================================== if not source: source = data.get("source") if source: source = str(source).strip().lower() # ======================================================== # 3. FBCLID # ======================================================== if not source and request.GET.get("fbclid"): source = "facebook" # ======================================================== # 4. HTTP REFERER # ======================================================== if not source: referer = request.META.get( "HTTP_REFERER", "" ).strip().lower() if referer: try: hostname = urlparse(referer).hostname or "" hostname = hostname.lower() except Exception: hostname = referer if ( hostname == "google.com" or hostname.endswith(".google.com") or ".google." in hostname ): source = "google" elif ( hostname == "instagram.com" or hostname.endswith(".instagram.com") ): source = "instagram" elif ( hostname == "facebook.com" or hostname.endswith(".facebook.com") or hostname == "fb.com" or hostname.endswith(".fb.com") ): source = "facebook" elif ( hostname == "whatsapp.com" or hostname.endswith(".whatsapp.com") ): source = "whatsapp" elif ( hostname == "youtube.com" or hostname.endswith(".youtube.com") or hostname == "youtu.be" ): source = "youtube" elif ( hostname == "linkedin.com" or hostname.endswith(".linkedin.com") ): source = "linkedin" elif ( hostname == "twitter.com" or hostname.endswith(".twitter.com") or hostname == "x.com" or hostname.endswith(".x.com") ): source = "twitter" elif ( hostname == "telegram.org" or hostname.endswith(".telegram.org") or hostname == "t.me" ): source = "telegram" else: source = "referral" # ======================================================== # 5. DIRECT # ======================================================== if not source: source = "direct" # ======================================================== # NORMALIZE # ======================================================== source_mapping = { "ig": "instagram", "insta": "instagram", "fb": "facebook", "meta": "facebook", "wa": "whatsapp", "yt": "youtube", "x": "twitter", "ref": "referral", } source = source_mapping.get( source, source ) # ======================================================== # VALID SOURCE # ======================================================== allowed_sources = { choice[0] for choice in BlogTrafficSource.SOURCE_CHOICES } if source not in allowed_sources: source = "referral" return source # ================= BLOG ANALYTICS VIEWS ================= from django.db.models import Sum, Count, Q from datetime import timedelta from django.utils.timezone import now from .models import * class BlogAnalyticsTrackingView(APIView): permission_classes = [AllowAny] def post(self, request, blog_id): data = request.data event_type = data.get( "event_type", "page_view" ) today = timezone.now().date() blog = get_object_or_404( Blog, id=blog_id ) # ==================================================== # PAGE VIEW # ==================================================== # ==================================================== # PAGE VIEW # ==================================================== if event_type == "page_view": # ==================================================== # IP ADDRESS # ==================================================== x_forwarded_for = request.META.get( "HTTP_X_FORWARDED_FOR" ) if x_forwarded_for: ip = ( x_forwarded_for .split(",")[0] .strip() ) else: ip = request.META.get( "REMOTE_ADDR", "" ) # ==================================================== # USER AGENT # ==================================================== user_agent = request.META.get( "HTTP_USER_AGENT", "" ) # ==================================================== # TRAFFIC SOURCE # ==================================================== source = detect_traffic_source( request, data ) # ==================================================== # DUPLICATE PAGE VIEW CHECK # ==================================================== duplicate_after = ( timezone.now() - timedelta(seconds=60) ) already_tracked = ( BlogPageViewEvent.objects.filter( blog=blog, ip_address=ip, user_agent=user_agent, source=source, viewed_at__gte=duplicate_after ) .exists() ) # ==================================================== # IGNORE DUPLICATE # ==================================================== if already_tracked: return Response( { "message": "duplicate page view ignored", "event_type": "page_view", "source": source, }, status=200 ) # ==================================================== # SAVE THIS PAGE VIEW # ==================================================== BlogPageViewEvent.objects.create( blog=blog, ip_address=ip, user_agent=user_agent, source=source, ) # ==================================================== # BLOG ANALYTICS # ==================================================== analytics, created = ( BlogAnalytics.objects.get_or_create( blog=blog, date=today ) ) # ==================================================== # TOTAL VIEWS # ==================================================== analytics.total_views += 1 # ==================================================== # UNIQUE VISITOR # ==================================================== visitor, visitor_created = ( BlogUniqueVisitor.objects.get_or_create( blog=blog, date=today, ip_address=ip, ) ) if visitor_created: analytics.unique_visitors += 1 # ==================================================== # TIME ON PAGE # ==================================================== time_on_page = int( data.get( "time_on_page", 0 ) or 0 ) if time_on_page > 0: analytics.total_read_time_seconds += ( time_on_page ) # ==================================================== # SCROLL # ==================================================== scroll = int( data.get( "scroll_percent", 0 ) or 0 ) scroll = max( 0, min(scroll, 100) ) BlogScrollSession.objects.create( blog=blog, date=today, ip_address=ip, max_scroll=scroll, read_time_seconds=time_on_page, ) # ==================================================== # SCROLL ANALYTICS # ==================================================== sessions = BlogScrollSession.objects.filter( blog=blog, date=today ) analytics.scroll_depth_25 = sessions.filter( max_scroll__gte=25, max_scroll__lt=50 ).count() analytics.scroll_depth_50 = sessions.filter( max_scroll__gte=50, max_scroll__lt=75 ).count() analytics.scroll_depth_75 = sessions.filter( max_scroll__gte=75, max_scroll__lt=100 ).count() analytics.scroll_depth_100 = sessions.filter( max_scroll=100 ).count() # ==================================================== # BOUNCE # ==================================================== if ( scroll == 0 and time_on_page < 5 ): analytics.bounce_count += 1 # ==================================================== # SAVE ANALYTICS # ==================================================== analytics.save() # ==================================================== # TRAFFIC SOURCE COUNT # ==================================================== traffic, _ = ( BlogTrafficSource.objects.get_or_create( blog=blog, source=source, date=today ) ) traffic.count += 1 traffic.save() # ==================================================== # PRODUCT CLICK # ==================================================== elif event_type == "product_click": product_id = data.get( "product_id" ) if product_id: product = get_object_or_404( Product, id=product_id ) prod_click, _ = ( BlogProductClick.objects.get_or_create( blog=blog, product=product, date=today ) ) prod_click.click_count += 1 prod_click.save() # ==================================================== # MARKETPLACE CLICK # ==================================================== elif event_type == "marketplace_click": platform_id = data.get( "platform_id" ) if platform_id: platform = get_object_or_404( Platform, id=platform_id ) market_click, _ = ( BlogMarketplaceClick.objects.get_or_create( blog=blog, platform=platform, date=today ) ) market_click.click_count += 1 market_click.save() # ==================================================== # RESPONSE # ==================================================== return Response( { "message": "tracked", "event_type": event_type, }, status=200 ) from io import StringIO class BlogAnalyticsSummaryView(APIView): permission_classes = [IsAuthenticated, RBACPermission] module_name = "blog_analytics" def get(self, request, blog_id, period): blog = get_object_or_404(Blog, id=blog_id) days = 30 if period == '30d' else 7 if period == '7d' else 99999 # lifetime start_date = timezone.now().date() - timedelta(days=days) if period != 'lifetime' else None analytics_qs = BlogAnalytics.objects.filter(blog=blog) traffic_qs = BlogTrafficSource.objects.filter(blog=blog) product_qs = BlogProductClick.objects.filter(blog=blog) marketplace_qs = BlogMarketplaceClick.objects.filter(blog=blog) scroll_qs = BlogScrollSession.objects.filter(blog=blog) if start_date: analytics_qs = analytics_qs.filter(date__gte=start_date) traffic_qs = traffic_qs.filter(date__gte=start_date) product_qs = product_qs.filter(date__gte=start_date) marketplace_qs = marketplace_qs.filter(date__gte=start_date) scroll_qs = scroll_qs.filter(date__gte=start_date) # Aggregate metrics total_views = analytics_qs.aggregate(Sum('total_views'))['total_views__sum'] or 0 total_unique = analytics_qs.aggregate(Sum('unique_visitors'))['unique_visitors__sum'] or 0 total_read_time = analytics_qs.aggregate(Sum('total_read_time_seconds'))['total_read_time_seconds__sum'] or 0 avg_read_time = total_read_time / total_views if total_views else 0 total_bounces = analytics_qs.aggregate(Sum('bounce_count'))['bounce_count__sum'] or 0 bounce_rate = (total_bounces / total_views * 100) if total_views else 0 # Scroll depth aggregated (percentage of users reaching each depth) # Scroll depth (Exclusive Buckets) scroll_25 = scroll_qs.filter( max_scroll__gte=25, max_scroll__lt=50 ).count() scroll_50 = scroll_qs.filter( max_scroll__gte=50, max_scroll__lt=75 ).count() scroll_75 = scroll_qs.filter( max_scroll__gte=75, max_scroll__lt=100 ).count() scroll_100 = scroll_qs.filter( max_scroll=100 ).count() # Traffic sources sources = traffic_qs.values('source').annotate(total=Sum('count')).order_by('-total') # Product clicks products = product_qs.values('product', 'product__product_name').annotate(total=Sum('click_count')).order_by('-total')[:10] # Marketplace clicks marketplaces = marketplace_qs.values('platform', 'platform__name').annotate(total=Sum('click_count')).order_by('-total')[:10] # Time series data for chart (daily) chart_data = list(analytics_qs.order_by('date').values('date', 'total_views', 'unique_visitors')) return Response({ "total_views": total_views, "unique_visitors": total_unique, "avg_read_time": round(avg_read_time, 1), "bounce_rate": total_bounces, "scroll_depth": { "25%": scroll_25, "50%": scroll_50, "75%": scroll_75, "100%": scroll_100, }, "traffic_sources": list(sources), "top_products": list(products), "top_marketplaces": list(marketplaces), "chart_data": chart_data, }) # ================= SUBSCRIBER SEGMENTATION & TAGGING ================= class SubscriberTagViewSet(viewsets.ModelViewSet): permission_classes = [AllowAny] module_name = "subscriber_tags" queryset = SubscriberTag.objects.all() serializer_class = SubscriberTagSerializer class SubscriberSegmentViewSet(viewsets.ModelViewSet): permission_classes = [AllowAny] queryset = SubscriberSegment.objects.all() serializer_class = SubscriberSegmentSerializer class SubscriberTagAssignmentView(APIView): permission_classes = [IsAuthenticated, RBACPermission] module_name = "subscriber_tag" def get(self, request): subscriber_id = request.query_params.get('subscriber_id') if not subscriber_id: return Response({"error": "subscriber_id required"}, status=400) subscriber = get_object_or_404(NewsletterSubscriber, id=subscriber_id) assignments = subscriber.tags.all() data = [{"tag_id": a.tag.id, "tag_name": a.tag.name, "color": a.tag.color} for a in assignments] return Response(data) def post(self, request): subscriber_id = request.data.get('subscriber_id') tag_id = request.data.get('tag_id') if not subscriber_id or not tag_id: return Response({"error": "subscriber_id and tag_id required"}, status=400) subscriber = get_object_or_404(NewsletterSubscriber, id=subscriber_id) tag = get_object_or_404(SubscriberTag, id=tag_id) assignment, created = SubscriberTagAssignment.objects.get_or_create(subscriber=subscriber, tag=tag) return Response({"message": "Tag assigned"}) def delete(self, request): subscriber_id = request.data.get('subscriber_id') tag_id = request.data.get('tag_id') if not subscriber_id or not tag_id: return Response({"error": "subscriber_id and tag_id required"}, status=400) SubscriberTagAssignment.objects.filter(subscriber_id=subscriber_id, tag_id=tag_id).delete() return Response({"message": "Tag removed"}) from datetime import timedelta from django.db.models import Q from django.utils import timezone from rest_framework.views import APIView from rest_framework.permissions import AllowAny from rest_framework.response import Response from datetime import timedelta from django.db.models import Q from django.utils import timezone from rest_framework.permissions import AllowAny from rest_framework.response import Response from rest_framework.views import APIView from datetime import timedelta from django.db.models import Q from django.utils import timezone from rest_framework.permissions import AllowAny from rest_framework.response import Response from rest_framework.views import APIView class SubscriberFilterView(APIView): permission_classes = [AllowAny] # module_name = "subscriber" LEAD_SOURCES = { "welcome": ["promotion_modal_WELCOME"], "exit": ["promotion_modal_EXIT"], "whatsapp": ["promotion_modal_WHATSAPP"], "product_engagement": ["product_page_subscription"], } CATEGORY_SOURCES = { "leads": [ "promotion_modal_WELCOME", "promotion_modal_EXIT", "promotion_modal_WHATSAPP", "promotion_modal_PUSH", "product_page_subscription", "lead_capture", ], "subscribers": [ "blog_subscription", "newsletter_footer", "contact_form", ], } def post(self, request): filters = request.data condition = filters.get("condition", "AND").upper() queryset = NewsletterSubscriber.objects.filter(is_active=True) conditions = [] # -------------------------------------------------- # Category # -------------------------------------------------- category = filters.get("category") if category in self.CATEGORY_SOURCES: conditions.append( Q(source__in=self.CATEGORY_SOURCES[category]) ) # -------------------------------------------------- # Lead Source # -------------------------------------------------- lead_source = filters.get("lead_source") if lead_source in self.LEAD_SOURCES: conditions.append( Q(source__in=self.LEAD_SOURCES[lead_source]) ) # -------------------------------------------------- # Source (optional) # -------------------------------------------------- source = filters.get("source") if source: conditions.append(Q(source=source)) # -------------------------------------------------- # Subscriber Type # -------------------------------------------------- subscriber_type = filters.get("subscriber_type") if subscriber_type == "email": conditions.append(Q(subscriber_type="email")) elif subscriber_type == "whatsapp": conditions.append(Q(subscriber_type="whatsapp")) elif subscriber_type == "both": # Email & WhatsApp -> no filter required pass # -------------------------------------------------- # Date From # -------------------------------------------------- if filters.get("date_from"): conditions.append( Q(subscribed_on__date__gte=filters["date_from"]) ) # -------------------------------------------------- # Date To # -------------------------------------------------- if filters.get("date_to"): conditions.append( Q(subscribed_on__date__lte=filters["date_to"]) ) # -------------------------------------------------- # Search # -------------------------------------------------- search = filters.get("search", "").strip() if search: conditions.append( Q(email__icontains=search) | Q(mobile_number__icontains=search) | Q(ip_address__icontains=search) | Q(city__icontains=search) | Q(country__icontains=search) ) # -------------------------------------------------- # Tags # -------------------------------------------------- tag_ids = filters.get("tags", []) if tag_ids: if condition == "AND": for tag_id in tag_ids: queryset = queryset.filter(tags__tag_id=tag_id) else: conditions.append( Q(tags__tag_id__in=tag_ids) ) # -------------------------------------------------- # Last Engagement # -------------------------------------------------- if filters.get("last_engagement_days"): try: days = int(filters["last_engagement_days"]) cutoff = timezone.now() - timedelta(days=days) engaged_ids = EmailTrackingEvent.objects.filter( created_at__gte=cutoff ).values_list("subscriber_id", flat=True) if condition == "AND": queryset = queryset.filter( id__in=engaged_ids ) else: conditions.append( Q(id__in=engaged_ids) ) except (ValueError, TypeError): pass # -------------------------------------------------- # Apply AND / OR # -------------------------------------------------- if conditions: if condition == "AND": for q in conditions: queryset = queryset.filter(q) else: final_query = conditions[0] for q in conditions[1:]: final_query |= q queryset = queryset.filter(final_query) queryset = queryset.distinct() serializer = SubscriberExtendedSerializer(queryset, many=True) return Response(serializer.data) from django.db.models import Q from datetime import datetime def evaluate_segment_rules(segment): """Evaluate segment rules and return list of subscriber IDs.""" if not segment.rules: return list(NewsletterSubscriber.objects.filter(is_active=True).values_list('id', flat=True)) condition = segment.rules.get('condition', 'AND') filters = segment.rules.get('filters', []) if not filters: return list(NewsletterSubscriber.objects.filter(is_active=True).values_list('id', flat=True)) q_objects = [] for f in filters: field = f.get('field') operator = f.get('operator') value = f.get('value') if not field or not operator or value is None: continue if field == 'tags': if operator in ('contains', 'equals'): q_objects.append(Q(tags__tag__name=value)) continue lookup = None if operator == 'equals': lookup = 'exact' elif operator == 'contains': lookup = 'icontains' elif operator == 'greater_than': lookup = 'gt' elif operator == 'less_than': lookup = 'lt' elif operator == 'greater_or_equal': lookup = 'gte' elif operator == 'less_or_equal': lookup = 'lte' elif operator == 'starts_with': lookup = 'istartswith' else: continue if field == 'subscribed_on' and isinstance(value, str): try: value = datetime.strptime(value, '%Y-%m-%d').date() except ValueError: continue filter_kwarg = {f"{field}__{lookup}": value} q_objects.append(Q(**filter_kwarg)) if not q_objects: return list(NewsletterSubscriber.objects.filter(is_active=True).values_list('id', flat=True)) if condition == 'AND': combined_q = Q() for q in q_objects: combined_q &= q else: combined_q = Q() for q in q_objects: combined_q |= q queryset = NewsletterSubscriber.objects.filter(combined_q, is_active=True).distinct() return list(queryset.values_list('id', flat=True)) # ================= EMAIL CAMPAIGN (FIXED) ================= import logging import re from urllib.parse import quote from django.db import transaction from django.core.paginator import Paginator from django.utils import timezone from django.core.mail import EmailMessage, get_connection from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.response import Response from .models import EmailCampaign, NewsletterSubscriber from .serializers import EmailCampaignSerializer # from .segment_utils import evaluate_segment_rules # assuming this exists from django.db.models import Q logger = logging.getLogger(__name__) class EmailCampaignViewSet(viewsets.ModelViewSet): # permission_classes = [IsAuthenticated, RBACPermission] module_name = "email_campaign" def get_permissions(self): if self.request.method == "POST": return [IsAuthenticated(), RBACPermission()] return [AllowAny()] queryset = EmailCampaign.objects.all() serializer_class = EmailCampaignSerializer def perform_create(self, serializer): campaign = serializer.save(draft='draft') return campaign @action(detail=False, methods=["get"]) def drafts(self, request): drafts = EmailCampaign.objects.filter( draft="draft" ).order_by("-created_at") serializer = self.get_serializer(drafts, many=True) return Response(serializer.data) @action(detail=True, methods=['post']) def send(self, request, pk=None): campaign = self.get_object() # ------------------------------------------------------------------ # Base queryset: active email subscribers # ------------------------------------------------------------------ subscribers = NewsletterSubscriber.objects.filter( is_active=True ).exclude( Q(email__isnull=True) | Q(email__exact="") ) # ------------------------------------------------------------------ # Filter by recipient type (all / segment / selected) # ------------------------------------------------------------------ if campaign.recipient_type == "all": pass elif campaign.recipient_type == "segment": if not campaign.recipient_sources: return Response( {"error": "No recipient sources selected."}, status=status.HTTP_400_BAD_REQUEST ) subscribers = subscribers.filter( source__in=campaign.recipient_sources ) elif campaign.recipient_type == "tags": tag_ids = list( campaign.recipient_tags.values_list( "id", flat=True ) ) if not tag_ids: return Response( {"error": "No tags selected for this campaign."}, status=status.HTTP_400_BAD_REQUEST ) subscribers = subscribers.filter( tags__tag_id__in=tag_ids ).distinct() elif campaign.recipient_type == "selected": if not campaign.selected_subscribers.exists(): return Response( {"error": "No subscribers selected."}, status=status.HTTP_400_BAD_REQUEST ) subscribers = subscribers.filter( id__in=campaign.selected_subscribers.values_list("id", flat=True) ) else: return Response( {"error": "Invalid recipient type."}, status=status.HTTP_400_BAD_REQUEST ) logger.error("================================") logger.error("Recipient Type: %s", campaign.recipient_type) logger.error("Subscribers Count: %s", subscribers.count()) logger.error( "Subscribers: %s", list(subscribers.values_list("id", "email")) ) logger.error("================================") # ------------------------------------------------------------------ # NEW: Exclude subscribers who already have a "sent" event for this campaign # ------------------------------------------------------------------ already_sent = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="sent" ).values_list("subscriber_id", flat=True) subscribers = subscribers.exclude(id__in=already_sent) # ------------------------------------------------------------------ # NEW: If no new subscribers remain, respond gracefully # ------------------------------------------------------------------ if not subscribers.exists(): return Response( {"message": "This campaign has already been sent to all selected subscribers."}, status=status.HTTP_200_OK ) # ------------------------------------------------------------------ # Prepare tracking URLs # ------------------------------------------------------------------ base_url = request.build_absolute_uri('/').rstrip('/') tracking_pixel_url = f"{base_url}/api/admin-panel/track/open/{{campaign_id}}/{{subscriber_id}}/" click_redirect_url = ( f"{base_url}/api/admin-panel/track/click/" f"{{campaign_id}}/{{subscriber_id}}/?url={{url}}" ) unsubscribe_url_template = ( f"{base_url}/api/admin-panel/unsubscribe/" f"{{campaign_id}}/{{subscriber_id}}/" ) # ------------------------------------------------------------------ # Email template (unchanged from your original) # ------------------------------------------------------------------ email_template = """ {{subject}}

Indizene

Discover premium quality
{{content}}

You received this email because you subscribed to our newsletter.
If you'd like to stop receiving updates, you can unsubscribe anytime.

Unsubscribe

© Indizene. All rights reserved.

""" # ------------------------------------------------------------------ # Send emails and record "sent" events # ------------------------------------------------------------------ connection = get_connection() connection.open() sent_count = 0 logger.info("Starting campaign %s", campaign.id) try: # Process in batches of 500 to avoid memory issues paginator = Paginator(subscribers, 500) for page in paginator.page_range: page_subscribers = paginator.page(page).object_list for subscriber in page_subscribers: # ------------------------------------------ # Prepare HTML content with tracking # ------------------------------------------ html_content = campaign.content def replace_link(match): original_url = match.group(1).strip() # Skip already tracked links if "/api/admin-panel/track/click/" in original_url: return match.group(0) # Skip mailto/tel/# links if ( original_url.startswith("mailto:") or original_url.startswith("tel:") or original_url.startswith("#") ): return match.group(0) # Only track absolute http/https URLs if not ( original_url.startswith("http://") or original_url.startswith("https://") ): return match.group(0) encoded_url = quote(original_url, safe="") tracking_url = click_redirect_url.format( campaign_id=campaign.id, subscriber_id=subscriber.id, url=encoded_url, ) return f'href="{tracking_url}"' # Replace only HTTP/HTTPS links html_content = re.sub( r'href=["\'](https?://[^"\']+)["\']', replace_link, html_content, flags=re.IGNORECASE, ) # ------------------------------------------ # Build final HTML # ------------------------------------------ final_html = email_template.replace( "{{subject}}", campaign.subject ) final_html = final_html.replace( "{{content}}", html_content ) final_html = final_html.replace( "{{unsubscribe_url}}", unsubscribe_url_template.format( campaign_id=campaign.id, subscriber_id=subscriber.id ) ) # ------------------------------------------ # Tracking Pixel # ------------------------------------------ pixel = ( f'' ) if "" in final_html: final_html = final_html.replace("", pixel + "") else: final_html += pixel # ------------------------------------------ # Send Email # ------------------------------------------ email = EmailMessage( subject=campaign.subject, body=final_html, from_email=None, to=[subscriber.email], connection=connection, ) email.content_subtype = "html" if campaign.attachment and hasattr(campaign.attachment, "path"): email.attach_file(campaign.attachment.path) email.send(fail_silently=False) EmailTrackingEvent.objects.create( campaign=campaign, subscriber=subscriber, event_type="sent", user_agent="", ip_address="", device_type="" ) sent_count += 1 except Exception as e: logger.exception("Email campaign sending failed", exc_info=True) return Response( {"error": f"Failed to send campaign: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) finally: connection.close() # ------------------------------------------------------------------ # Update campaign statistics atomically # ------------------------------------------------------------------ with transaction.atomic(): campaign.recipient_count += sent_count # You may still set status to "sent" to mark overall completion campaign.status = "sent" campaign.sent_at = timezone.now() campaign.save() logger.info("Campaign %s completed successfully", campaign.id) return Response( { "message": "Campaign sent successfully.", "recipient_count": campaign.recipient_count, "campaign_id": campaign.id, "status": campaign.status, "sent_at": campaign.sent_at, }, status=status.HTTP_200_OK, ) from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from django.shortcuts import get_object_or_404 class SaveDraftCampaignAPIView(APIView): permission_classes = [IsAuthenticated] def post(self, request): campaign_id = request.data.get("campaign_id") if campaign_id: campaign = get_object_or_404( EmailCampaign, id=campaign_id ) else: campaign = EmailCampaign() # ----------------------------- # Campaign Details # ----------------------------- campaign.subject = request.data.get("subject", "").strip() campaign.content = request.data.get("content", "").strip() campaign.recipient_type = request.data.get("recipient_type", "all") campaign.draft = "draft" # ----------------------------- # Recipient Segment # ----------------------------- recipient_sources = request.data.get("recipient_sources", []) if isinstance(recipient_sources, str): try: recipient_sources = json.loads(recipient_sources) except Exception: recipient_sources = [recipient_sources] campaign.recipient_sources = recipient_sources # ----------------------------- # Attachment # ----------------------------- if "attachment" in request.FILES: campaign.attachment = request.FILES["attachment"] campaign.save() # ----------------------------- # Selected Subscribers # ----------------------------- subscriber_ids = request.data.getlist("selected_subscribers") if campaign.recipient_type == "selected": subscribers = NewsletterSubscriber.objects.filter( id__in=subscriber_ids, is_active=True, subscriber_type__in=[ "email", "whatsapp & Email", ] ) campaign.selected_subscribers.set(subscribers) else: campaign.selected_subscribers.clear() # ----------------------------- # Recipient Tags # ----------------------------- tag_ids = request.data.getlist("recipient_tags") if campaign.recipient_type == "tags": campaign.recipient_tags.set(tag_ids) else: campaign.recipient_tags.clear() # ------------------------------------ # IMPORTANT: # Do NOT update recipient_count here. # recipient_count should only be updated # after the campaign is actually sent. # ------------------------------------ campaign.save() return Response( { "message": "Draft saved successfully.", "campaign_id": campaign.id, "recipient_type": campaign.recipient_type, "recipient_count": campaign.recipient_count, "selected_subscribers": list( campaign.selected_subscribers.values_list( "id", flat=True ) ), "recipient_tags": list( campaign.recipient_tags.values_list( "id", flat=True ) ), }, status=status.HTTP_200_OK, ) class DraftCampaignListAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request): drafts = EmailCampaign.objects.filter( draft="draft" ).order_by("-created_at") serializer = EmailCampaignSerializer( drafts, many=True ) return Response(serializer.data) class DraftCampaignDetailAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request, campaign_id): campaign = get_object_or_404( EmailCampaign, id=campaign_id, draft="draft" ) serializer = EmailCampaignSerializer(campaign) return Response(serializer.data) class DeleteDraftCampaignAPIView(APIView): permission_classes = [IsAuthenticated] def delete(self, request, campaign_id): campaign = get_object_or_404( EmailCampaign, id=campaign_id, draft="draft" ) campaign.delete() return Response( { "message": "Draft deleted successfully." }, status=status.HTTP_200_OK ) class UnsubscribeAPIView(APIView): permission_classes = [AllowAny] def get(self, request, campaign_id, subscriber_id): campaign = get_object_or_404( EmailCampaign, id=campaign_id ) subscriber = get_object_or_404( NewsletterSubscriber, id=subscriber_id ) subscriber.status = "unsubscribed" subscriber.is_active = False subscriber.save() user_agent = request.META.get("HTTP_USER_AGENT", "") device_type = "desktop" if "Mobile" in user_agent or "Android" in user_agent: device_type = "mobile" elif "iPad" in user_agent or "Tablet" in user_agent: device_type = "tablet" EmailTrackingEvent.objects.create( campaign=campaign, subscriber=subscriber, event_type="unsubscribe", user_agent=user_agent, ip_address=get_client_ip(request), device_type=device_type, ) return HttpResponse(""" Unsubscribed | Indizene

Successfully Unsubscribed

We've processed your request

You're all set

You have been removed from our newsletter list. You will no longer receive email campaigns from Indizene.

We'll miss you ❤️
""") # ================= TRACKING (ONLY THESE TWO) ================= from django.http import HttpResponse from django.shortcuts import get_object_or_404 from rest_framework.permissions import AllowAny from rest_framework.views import APIView from django.http import HttpResponse from django.shortcuts import get_object_or_404 from rest_framework.permissions import AllowAny from rest_framework.views import APIView from django.http import HttpResponse from django.shortcuts import get_object_or_404 from rest_framework.permissions import AllowAny from rest_framework.views import APIView import logging from django.http import HttpResponse from django.shortcuts import get_object_or_404 from django.utils import timezone logger = logging.getLogger(__name__) class TrackOpenAPIView(APIView): permission_classes = [AllowAny] PIXEL = ( b'GIF89a\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff\x00\x00\x00' b'!\xf9\x04\x01\x00\x00\x00\x00,\x00\x00\x00\x00\x01\x00\x01' b'\x00\x00\x02\x02D\x01\x00;' ) def get(self, request, campaign_id, subscriber_id): campaign = get_object_or_404( EmailCampaign, id=campaign_id ) subscriber = get_object_or_404( NewsletterSubscriber, id=subscriber_id ) user_agent = request.META.get("HTTP_USER_AGENT", "") ua = user_agent.lower() ip = get_client_ip(request) logger.error("=" * 80) logger.error("TRACK OPEN HIT") logger.error("Time: %s", timezone.now()) logger.error("Campaign: %s", campaign.id) logger.error("Subscriber: %s", subscriber.id) logger.error("Method: %s", request.method) logger.error("IP: %s", ip) logger.error("IP Type: %s", type(ip)) logger.error("User-Agent: %s", user_agent) logger.error("=" * 80) # Ignore HEAD request if request.method == "HEAD": logger.error("HEAD request ignored.") return HttpResponse(self.PIXEL, content_type="image/gif") # ----------------------------- # Device Detection # ----------------------------- device_type = "desktop" if any(x in ua for x in [ "iphone", "android", "mobile", "opera mini", "blackberry", ]): device_type = "mobile" elif any(x in ua for x in [ "ipad", "tablet", "kindle", "silk", ]): device_type = "tablet" # ----------------------------- # Detect Proxy # ----------------------------- proxy_request = False bot_keywords = [ "bot", "crawler", "spider", "crawl", "scanner", "preview", "fetch", "monitor", ] if any(x in ua for x in bot_keywords): proxy_request = True logger.error("Matched Bot Keyword") proxy_agents = [ "googleimageproxy", "gmailimageproxy", "google-proxy", "google proxy", "mailprivacy", "mail privacy", "apple", "outlook", "office365", "exchange", "microsoft", "proofpoint", "mimecast", "barracuda", "symantec", "trend micro", "mcafee", "kaspersky", "eset", "bitdefender", "forcepoint", "fortinet", "cloudmark", "sophos", ] if any(x in ua for x in proxy_agents): proxy_request = True logger.error("Matched Proxy User-Agent") google_ip_prefixes = ( "209.85.", "66.249.", "64.233.", "74.125.", ) if ip: ip_string = str(ip) if any(ip_string.startswith(prefix) for prefix in google_ip_prefixes): proxy_request = True logger.error("Matched Google Proxy IP") logger.error("Proxy Request: %s", proxy_request) logger.error("Device Type: %s", device_type) # ----------------------------- # Event Type # ----------------------------- if ( "googleimageproxy" in ua or ( ip and str(ip).startswith(( "66.249.", "64.233.", "74.125.", "209.85.", )) ) ): event_type = "gmail_proxy" elif "mailprivacy" in ua: event_type = "apple_proxy" elif "outlook" in ua: event_type = "outlook_proxy" elif proxy_request: event_type = "proxy_open" else: event_type = "open" logger.error("Final Event Type: %s", event_type) # ----------------------------- # Gmail Proxy Handling # ----------------------------- if event_type in ["gmail_proxy","apple_proxy","outlook_proxy","proxy_open"]: existing_proxy = EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber, event_type=event_type, ).first() if existing_proxy: logger.error("Duplicate proxy request ignored.") return HttpResponse( self.PIXEL, content_type="image/gif" ) # First request → create proxy_open exists = EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber, event_type=event_type, ).exists() if not exists: EmailTrackingEvent.objects.create( campaign=campaign, subscriber=subscriber, event_type=event_type, user_agent=user_agent, ip_address=ip, device_type=device_type, ) logger.error("Event Saved Successfully") # ----------------------------- # Update Analytics # ----------------------------- campaign.total_opens = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="open" ).count() campaign.unique_opens = ( EmailTrackingEvent.objects.filter( campaign=campaign, event_type="open" ) .values("subscriber") .distinct() .count() ) campaign.save(update_fields=[ "total_opens", "unique_opens", ]) logger.error( "Campaign Updated -> Total Opens=%s Unique Opens=%s", campaign.total_opens, campaign.unique_opens, ) logger.error("=" * 80) return HttpResponse( self.PIXEL, content_type="image/gif" ) from urllib.parse import unquote from django.shortcuts import get_object_or_404, redirect from rest_framework.permissions import AllowAny from rest_framework.views import APIView class TrackClickAPIView(APIView): permission_classes = [AllowAny] def get(self, request, campaign_id, subscriber_id): decoded_url = unquote(request.GET.get("url", "")) campaign = get_object_or_404( EmailCampaign, id=campaign_id ) subscriber = get_object_or_404( NewsletterSubscriber, id=subscriber_id ) user_agent = request.META.get("HTTP_USER_AGENT", "") ua = user_agent.lower() device_type = "desktop" if any(x in ua for x in [ "iphone", "android", "mobile", "opera mini", "blackberry", ]): device_type = "mobile" elif any(x in ua for x in [ "ipad", "tablet", "kindle", "silk", ]): device_type = "tablet" # --------------------------------------- # Save Click # --------------------------------------- EmailTrackingEvent.objects.create( campaign=campaign, subscriber=subscriber, event_type="click", clicked_url=decoded_url, ip_address=get_client_ip(request), # city, country = get_location(ip_address) user_agent=user_agent, device_type=device_type, ) # --------------------------------------- # Upgrade to VERIFIED OPEN # (Click proves the email was opened) # --------------------------------------- if not EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber, event_type="open", ).exists(): EmailTrackingEvent.objects.create( campaign=campaign, subscriber=subscriber, event_type="open", ip_address=get_client_ip(request), # city, country = get_location(ip_address), user_agent=user_agent, device_type=device_type, ) EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber, event_type__in=[ "gmail_proxy", "apple_proxy", "outlook_proxy", "proxy_open", ], ).delete() # --------------------------------------- # Update campaign statistics # --------------------------------------- campaign.total_clicks = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="click", ).count() campaign.unique_clicks = ( EmailTrackingEvent.objects.filter( campaign=campaign, event_type="click", ) .values("subscriber") .distinct() .count() ) campaign.total_opens = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="open", ).count() campaign.unique_opens = ( EmailTrackingEvent.objects.filter( campaign=campaign, event_type="open", ) .values("subscriber") .distinct() .count() ) campaign.save( update_fields=[ "total_clicks", "unique_clicks", "total_opens", "unique_opens", ] ) return redirect(decoded_url) # ================= ANALYTICS ================= class CampaignAnalyticsAPIView(APIView): def get(self, request): campaigns = EmailCampaign.objects.filter(removed=False).order_by("-id") data = [] for campaign in campaigns: gmail_proxy = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="gmail_proxy" ) apple_proxy = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="apple_proxy" ) outlook_proxy = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="outlook_proxy" ) verified_opens = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="open" ) clicks = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="click" ) unsubscribes = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="unsubscribe" ) bounces = EmailTrackingEvent.objects.filter( campaign=campaign, event_type="bounce" ) delivered = max( campaign.recipient_count - bounces.count(), 1 ) device_breakdown = { "desktop": verified_opens.filter( device_type__iexact="desktop" ).count(), "mobile": verified_opens.filter( device_type__iexact="mobile" ).count(), "tablet": verified_opens.filter( device_type__iexact="tablet" ).count(), } hourly_engagement = {} for event in EmailTrackingEvent.objects.filter( campaign=campaign ): hour = event.created_at.strftime("%H") hourly_engagement[hour] = ( hourly_engagement.get(hour, 0) + 1 ) estimated_unique = ( EmailTrackingEvent.objects.filter( campaign=campaign, event_type__in=[ "gmail_proxy", "apple_proxy", "outlook_proxy", "open", ] ) .values("subscriber") .distinct() .count() ) verified_unique = ( verified_opens.values("subscriber") .distinct() .count() ) unique_clicks = ( clicks.values("subscriber") .distinct() .count() ) data.append({ "id": campaign.id, "subject": campaign.subject, "sent_at": campaign.sent_at, "recipient_count": campaign.recipient_count, "estimated_opens": estimated_unique, "unique_opens": verified_unique, "unique_clicks": unique_clicks, "estimated_open_rate": round( estimated_unique * 100 / delivered, 2 ), "open_rate": round( verified_unique * 100 / delivered, 2 ), "click_rate": round( unique_clicks * 100 / delivered, 2 ), "ctor": round( unique_clicks * 100 / verified_unique, 2 ) if verified_unique else 0, "unsubscribe_rate": round( unsubscribes.count() * 100 / delivered, 2 ), "bounce_rate": campaign.recipient_count-unique_clicks, "device_breakdown": device_breakdown, "hourly_engagement": hourly_engagement, }) return Response(data) # data.append({ # "id": campaign.id, # "subject": campaign.subject, # "sent_at": campaign.sent_at, # "recipient_count": total, # "unique_opens": opens.values("subscriber").distinct().count(), # "unique_clicks": clicks.values("subscriber").distinct().count(), # "open_rate": round((opens.values("subscriber").distinct().count() / total) * 100, 2), # "click_rate": round((clicks.values("subscriber").distinct().count() / total) * 100, 2), # "unsubscribe_rate": round((unsubs.count() / total) * 100, 2), # "bounce_rate": 100-round((opens.values("subscriber").distinct().count() / total) * 100, 2), # "device_breakdown": device_breakdown, # "hourly_engagement": hourly_engagement, # }) class CampaignDelete(APIView): # permission_classes = [IsAuthenticated] module_name = "email_campaign" def get_permissions(self): if self.request.method == "POST": return [IsAuthenticated(), RBACPermission()] return [AllowAny()] def delete(self, request, pk): campaign = EmailCampaign.objects.filter(id=pk).first() # campaign.removed=True campaign.delete() return Response({'status':'Success'}) class CampaignEventsAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request, campaign_id): events = ( EmailTrackingEvent.objects .filter(campaign_id=campaign_id) .select_related("subscriber") .order_by("-created_at") ) data = [] for event in events: data.append({ "id": event.id, "subscriber_email": ( event.subscriber.email if event.subscriber else None ), "event_type": event.event_type, "device_type": event.device_type, "ip_address": event.ip_address, "clicked_url": event.clicked_url, "created_at": event.created_at, }) return Response(data) # ================= BOUNCE WEBHOOK STUB ================= class BounceWebhookView(APIView): permission_classes = [AllowAny] def post(self, request): email = request.data.get('email') campaign_id = request.data.get('campaign_id') if email: subscriber = NewsletterSubscriber.objects.filter(email=email).first() if subscriber: EmailTrackingEvent.objects.create( subscriber=subscriber, campaign_id=campaign_id, event_type='bounce', ip_address=get_client_ip(request) ) return Response({"status": "bounce recorded"}) # ================= SUBSCRIBER EXPORT ================= from datetime import datetime, timedelta from io import StringIO import csv from django.http import HttpResponse from django.utils import timezone from openpyxl import Workbook from openpyxl.styles import Font from rest_framework.permissions import AllowAny from rest_framework.views import APIView class SubscriberExportView(APIView): module_name = "subscriber_export" permission_classes = [IsAuthenticated, RBACPermission] def post(self, request): filters = request.data.get("filters", {}) fields = request.data.get( "fields", [ "email", "mobile_number", "source", "subscribed_on", "tags", "last_engagement", ], ) export_format = request.data.get("format", "csv").lower() queryset = NewsletterSubscriber.objects.filter(is_active=True) # ----------------------- # Filters # ----------------------- if filters.get("source"): queryset = queryset.filter( source__icontains=filters["source"] ) if filters.get("subscriber_type"): queryset = queryset.filter( subscriber_type=filters["subscriber_type"] ) if filters.get("date_from"): queryset = queryset.filter( subscribed_on__gte=filters["date_from"] ) if filters.get("date_to"): queryset = queryset.filter( subscribed_on__lte=filters["date_to"] ) if filters.get("tags"): queryset = queryset.filter( tags__tag_id__in=filters["tags"] ).distinct() if filters.get("last_engagement_days"): cutoff = timezone.now() - timedelta( days=int(filters["last_engagement_days"]) ) engaged_subscribers = EmailTrackingEvent.objects.filter( created_at__gte=cutoff ).values_list("subscriber_id", flat=True) queryset = queryset.filter( id__in=engaged_subscribers ) # ----------------------- # Build Export Data # ----------------------- data = [] for sub in queryset: row = {} for field in fields: # Tags if field == "tags": row["tags"] = ", ".join( tag.tag.name for tag in sub.tags.select_related("tag").all() ) # Last Engagement elif field == "last_engagement": last_event = ( EmailTrackingEvent.objects.filter( subscriber=sub ) .order_by("-created_at") .first() ) if last_event: row["last_engagement"] = timezone.localtime( last_event.created_at ).strftime("%d-%m-%Y %I:%M:%S %p") else: row["last_engagement"] = "" # Normal Fields else: value = getattr(sub, field, "") if value is None: value = "" elif isinstance(value, datetime): if timezone.is_aware(value): value = timezone.localtime(value) value = value.strftime( "%d-%m-%Y %I:%M:%S %p" ) row[field] = value data.append(row) # ----------------------- # CSV Export # ----------------------- if export_format == "csv": output = StringIO() writer = csv.DictWriter( output, fieldnames=fields ) writer.writeheader() writer.writerows(data) response = HttpResponse( output.getvalue(), content_type="text/csv", ) response[ "Content-Disposition" ] = 'attachment; filename="subscribers.csv"' return response # ----------------------- # Excel Export # ----------------------- wb = Workbook() ws = wb.active ws.title = "Subscribers" # Header for col, field in enumerate(fields, 1): cell = ws.cell( row=1, column=col, value=field.replace("_", " ").title(), ) cell.font = Font(bold=True) # Data for row_idx, row in enumerate(data, start=2): for col_idx, field in enumerate(fields, start=1): ws.cell( row=row_idx, column=col_idx, value=row.get(field, ""), ) # Auto Width for column_cells in ws.columns: length = max( len(str(cell.value or "")) for cell in column_cells ) ws.column_dimensions[ column_cells[0].column_letter ].width = min(length + 5, 50) response = HttpResponse( content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) response[ "Content-Disposition" ] = 'attachment; filename="subscribers.xlsx"' wb.save(response) return response # ================= VISITOR HEARTBEAT (OPTIONAL) ================= import uuid class VisitorHeartbeatView(APIView): permission_classes = [AllowAny] def post(self, request): session_id = request.data.get('session_id') email = request.data.get('email') if not session_id: session_id = ( request.META.get('HTTP_USER_AGENT', '') + request.META.get('REMOTE_ADDR', '') ) session_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, session_id)) ip = get_client_ip(request) city, country = get_location(ip) user_agent = request.META.get('HTTP_USER_AGENT', '') visitor, created = ActiveVisitor.objects.update_or_create( session_id=session_id, defaults={ 'ip_address': ip, 'email': email, 'city': city, 'country': country, 'user_agent': user_agent, 'last_active': timezone.now(), } ) return Response({ "status": "ok", "session_id": session_id, "city": city, "country": country }) class ActiveVisitorsView(APIView): permission_classes = [AllowAny] def get(self, request): cutoff = timezone.now() - timedelta(minutes=5) active = ActiveVisitor.objects.filter( last_active__gte=cutoff ).order_by('-last_active') data = [] for v in active: data.append({ 'ip': v.ip_address or 'unknown', 'email': v.email or '—', 'city': v.city or 'Unknown', 'country': v.country or 'Unknown', 'location': f"{v.city}, {v.country}" if v.city and v.country else "Unknown", 'last_active': v.last_active.isoformat(), 'is_online': True, }) return Response(data) from django.http import HttpResponse from django.shortcuts import get_object_or_404 from rest_framework.views import APIView from openpyxl import Workbook from openpyxl.styles import Font, PatternFill import csv from .models import ( EmailCampaign, EmailTrackingEvent, NewsletterSubscriber, ) from datetime import datetime from django.utils import timezone from django.shortcuts import get_object_or_404 from django.http import HttpResponse from rest_framework.views import APIView from openpyxl import Workbook from openpyxl.styles import PatternFill, Font import csv from .models import EmailCampaign, NewsletterSubscriber, EmailTrackingEvent from datetime import datetime from django.utils import timezone from django.shortcuts import get_object_or_404 from django.http import HttpResponse from rest_framework.views import APIView from openpyxl import Workbook from openpyxl.styles import PatternFill, Font import csv from .models import EmailCampaign, NewsletterSubscriber, EmailTrackingEvent class ExportCampaignReportAPIView(APIView): module_name = "campaign_report_export" permission_classes = [IsAuthenticated, RBACPermission] def excel_safe(self, value): """ Convert timezone-aware datetime to naive datetime (local time) for Excel. Any other value is returned unchanged. """ if isinstance(value, datetime): if timezone.is_aware(value): value = timezone.localtime(value) return value.replace(tzinfo=None) return value def get(self, request, campaign_id): campaign = get_object_or_404( EmailCampaign.objects.prefetch_related( "events", "selected_subscribers" ), id=campaign_id ) export_format = request.GET.get("format", "xlsx").lower() if export_format == "csv": return self.export_csv(campaign) return self.export_excel(campaign) # ========================================================== # EXCEL EXPORT # ========================================================== def export_excel(self, campaign): wb = Workbook() header_fill = PatternFill( start_color="1F4E78", end_color="1F4E78", fill_type="solid" ) header_font = Font( bold=True, color="FFFFFF" ) # ===================================================== # SUMMARY SHEET # ===================================================== summary = wb.active summary.title = "Campaign Summary" headers = [ "Campaign Subject", "Status", "Recipient Count", "Sent At", "Total Opens", "Unique Opens", "Open Rate", "Total Clicks", "Unique Clicks", "CTR", "Bounce Count", "Bounce Rate", "Unsubscribes", "Mobile Opens", "Desktop Opens", "Tablet Opens", ] summary.append(headers) for cell in summary[1]: cell.fill = header_fill cell.font = header_font bounce_count = campaign.events.filter( event_type="bounce" ).count() delivered = max( campaign.recipient_count - bounce_count, 0 ) mobile_opens = campaign.events.filter( event_type="open", device_type="mobile" ).count() desktop_opens = campaign.events.filter( event_type="open", device_type="desktop" ).count() tablet_opens = campaign.events.filter( event_type="open", device_type="tablet" ).count() open_rate = ( (campaign.unique_opens / delivered) * 100 if delivered else 0 ) ctr = ( (campaign.unique_clicks / delivered) * 100 if delivered else 0 ) bounce_rate = ( (bounce_count / campaign.recipient_count) * 100 if campaign.recipient_count else 0 ) summary.append([ campaign.subject, campaign.status, campaign.recipient_count, campaign.sent_at.strftime("%d-%m-%Y %H:%M") if campaign.sent_at else "", campaign.total_opens, campaign.unique_opens, f"{open_rate:.2f}%", campaign.total_clicks, campaign.unique_clicks, f"{ctr:.2f}%", campaign.recipient_count-campaign.unique_clicks, campaign.unsubscribe_count, mobile_opens, desktop_opens, tablet_opens, ]) # ===================================================== # SUBSCRIBER REPORT (removed First Click & Last Click) # ===================================================== subscriber_sheet = wb.create_sheet("Subscribers") subscriber_headers = [ "Email", "Subscriber Status", "Opened", "Clicked", "Bounced", "Open Count", "Click Count", "First Open", "Last Open", # "First Click", <-- removed # "Last Click", <-- removed "Clicked URLs", "Devices", "IP Addresses", ] subscriber_sheet.append(subscriber_headers) for cell in subscriber_sheet[1]: cell.fill = header_fill cell.font = header_font subscribers = NewsletterSubscriber.objects.filter( emailtrackingevent__campaign=campaign ).distinct() for subscriber in subscribers: events = EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber ).order_by("created_at") open_events = events.filter( event_type="open" ) click_events = events.filter( event_type="click" ) bounce_events = events.filter( event_type="bounce" ) first_open = ( open_events.first().created_at if open_events.exists() else "" ) last_open = ( open_events.last().created_at if open_events.exists() else "" ) # first_click and last_click are no longer used # but we keep them in the code if needed elsewhere urls = ", ".join( click_events.exclude( clicked_url__isnull=True ).exclude( clicked_url="" ).values_list( "clicked_url", flat=True ).distinct() ) devices = ", ".join( open_events.exclude( device_type__isnull=True ).exclude( device_type="" ).values_list( "device_type", flat=True ).distinct() ) ips = ", ".join( events.exclude( ip_address__isnull=True ).values_list( "ip_address", flat=True ).distinct() ) subscriber_sheet.append([ subscriber.email, subscriber.status, "Yes" if open_events.exists() else "No", "Yes" if click_events.exists() else "No", "Yes" if click_events.exists() else "No", open_events.count(), click_events.count(), self.excel_safe(first_open), self.excel_safe(last_open), # self.excel_safe(first_click), <-- removed # self.excel_safe(last_click), <-- removed urls, devices, ips, ]) # ===================================================== # RAW EVENTS SHEET (removed Clicked URL) # ===================================================== events_sheet = wb.create_sheet("Raw Events") event_headers = [ "Event ID", "Subscriber", "Event Type", "Device", "IP Address", # "Clicked URL", <-- removed "User Agent", "Created At", ] events_sheet.append(event_headers) for cell in events_sheet[1]: cell.fill = header_fill cell.font = header_font events = EmailTrackingEvent.objects.filter( campaign=campaign ).select_related("subscriber").order_by("-created_at") for event in events: events_sheet.append([ event.id, event.subscriber.email if event.subscriber else "", event.event_type, event.device_type or "", event.ip_address or "", # event.clicked_url or "", <-- removed event.user_agent or "", self.excel_safe(event.created_at), ]) # Auto-fit column widths for sheet in wb.worksheets: for column in sheet.columns: max_length = 0 column_letter = column[0].column_letter for cell in column: try: if cell.value: max_length = max( max_length, len(str(cell.value)) ) except Exception: pass sheet.column_dimensions[column_letter].width = min(max_length + 5, 60) response = HttpResponse( content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) response[ "Content-Disposition" ] = f'attachment; filename="campaign_{campaign.id}_report.xlsx"' wb.save(response) return response # ========================================================== # CSV EXPORT (unchanged, but kept for completeness) # ========================================================== def export_csv(self, campaign): response = HttpResponse(content_type="text/csv") response[ "Content-Disposition" ] = f'attachment; filename="campaign_{campaign.id}_report.csv"' writer = csv.writer(response) writer.writerow([ "Email", "Subscriber Status", "Opened", "Clicked", "Bounced", "Open Count", "Click Count", "First Open", "Last Open", "First Click", "Last Click", "Clicked URLs", "Devices", "IP Addresses", ]) subscribers = NewsletterSubscriber.objects.filter( emailtrackingevent__campaign=campaign ).distinct() for subscriber in subscribers: events = EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber ).order_by("created_at") open_events = events.filter(event_type="open") click_events = events.filter(event_type="click") bounce_events = events.filter(event_type="bounce") first_open_str = ( timezone.localtime(open_events.first().created_at).strftime("%d-%m-%Y %H:%M:%S") if open_events.exists() else "" ) last_open_str = ( timezone.localtime(open_events.last().created_at).strftime("%d-%m-%Y %H:%M:%S") if open_events.exists() else "" ) first_click_str = ( timezone.localtime(click_events.first().created_at).strftime("%d-%m-%Y %H:%M:%S") if click_events.exists() else "" ) last_click_str = ( timezone.localtime(click_events.last().created_at).strftime("%d-%m-%Y %H:%M:%S") if click_events.exists() else "" ) writer.writerow([ subscriber.email, subscriber.status, "Yes" if open_events.exists() else "No", "Yes" if click_events.exists() else "No", "Yes" if click_events.exists() else "No", open_events.count(), click_events.count(), first_open_str, last_open_str, first_click_str, last_click_str, ", ".join( click_events.exclude( clicked_url__isnull=True ).exclude( clicked_url="" ).values_list( "clicked_url", flat=True ).distinct() ), ", ".join( open_events.exclude( device_type__isnull=True ).exclude( device_type="" ).values_list( "device_type", flat=True ).distinct() ), ", ".join( events.exclude( ip_address__isnull=True ).values_list( "ip_address", flat=True ).distinct() ), ]) return response from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from django.shortcuts import get_object_or_404 class SubscriberHistoryAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request, subscriber_id): subscriber = get_object_or_404( NewsletterSubscriber, id=subscriber_id ) events = ( EmailTrackingEvent.objects .filter(subscriber=subscriber) .select_related("campaign") .order_by("created_at") ) timeline = [] # Subscribe Event timeline.append({ "type": "subscribed", "date": subscriber.subscribed_on, "description": "Subscriber joined the newsletter." }) opens = 0 clicks = 0 bounces = 0 unsubscribes = 0 for event in events: if event.event_type == "open": opens += 1 elif event.event_type == "click": clicks += 1 elif event.event_type == "bounce": bounces += 1 elif event.event_type == "unsubscribe": unsubscribes += 1 timeline.append({ "type": event.event_type, "campaign_id": event.campaign.id, "campaign_subject": event.campaign.subject, "date": event.created_at, "device": event.device_type, "ip_address": event.ip_address, "clicked_url": event.clicked_url }) # Final unsubscribe event if subscriber.unsubscribe_on: timeline.append({ "type": "unsubscribed", "date": subscriber.unsubscribe_on, "description": "Subscriber unsubscribed from newsletter." }) timeline = sorted( timeline, key=lambda x: x["date"] ) return Response({ "subscriber": { "id": subscriber.id, "email": subscriber.email, "source": subscriber.source, "status": subscriber.status, "subscribed_on": subscriber.subscribed_on, "unsubscribe_on": subscriber.unsubscribe_on, }, "summary": { "total_events": events.count(), "opens": opens, "clicks": clicks, "bounces": bounces, "unsubscribes": unsubscribes, }, "timeline": timeline }) from django.db.models import Count, Min, Q from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from django.db.models import Count, Q, Min from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from django.db.models import Count, Q, Min from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status class SubscriberCampaignEngagementAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request, subscriber_id): subscriber = get_object_or_404( NewsletterSubscriber, id=subscriber_id ) campaigns = ( EmailCampaign.objects.filter(status="sent") .order_by("-sent_at") ) campaign_data = [] for campaign in campaigns: open_event = ( EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber, event_type="open" ) .order_by("created_at") .first() ) click_event = ( EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber, event_type="click" ) .order_by("created_at") .first() ) proxy_event = ( EmailTrackingEvent.objects.filter( campaign=campaign, subscriber=subscriber, event_type="proxy_open" ) .order_by("created_at") .first() ) campaign_data.append({ "campaign_id": campaign.id, "campaign_subject": campaign.subject, "sent_at": campaign.sent_at, "opened": open_event is not None, "opened_at": open_event.created_at if open_event else None, "clicked": click_event is not None, "clicked_at": click_event.created_at if click_event else None, # Optional: useful for Gmail debugging "proxy_opened": proxy_event is not None, "proxy_opened_at": proxy_event.created_at if proxy_event else None, }) return Response( { "subscriber": { "id": subscriber.id, "email": subscriber.email, "status": subscriber.status, }, "campaigns": campaign_data, }, status=status.HTTP_200_OK, ) from django.db.models import Count from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from django.utils.dateparse import parse_date class CampaignDashboardAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request): start_date = request.query_params.get("start_date") end_date = request.query_params.get("end_date") campaigns = EmailCampaign.objects.filter(status="sent") if start_date: campaigns = campaigns.filter(sent_at__date__gte=parse_date(start_date)) if end_date: campaigns = campaigns.filter(sent_at__date__lte=parse_date(end_date)) # ------------------------- # Summary # ------------------------- total_campaigns = campaigns.count() total_sent = sum(c.recipient_count for c in campaigns) total_opens = sum(c.unique_opens for c in campaigns) total_clicks = sum(c.unique_clicks for c in campaigns) avg_open_rate = ( round((total_opens / total_sent) * 100, 2) if total_sent else 0 ) avg_click_rate = ( round((total_clicks / total_sent) * 100, 2) if total_sent else 0 ) # ------------------------- # Campaign Comparison # ------------------------- campaign_comparison = [] for campaign in campaigns: open_rate = ( round(campaign.unique_opens * 100 / campaign.recipient_count, 2) if campaign.recipient_count else 0 ) click_rate = ( round(campaign.unique_clicks * 100 / campaign.recipient_count, 2) if campaign.recipient_count else 0 ) campaign_comparison.append({ "date":campaign.created_at, "campaign_id": campaign.id, "campaign_name": campaign.subject, "sent": campaign.recipient_count, "opened": campaign.unique_opens, "clicked": campaign.unique_clicks, "open_rate": open_rate, "click_rate": click_rate, "bounce": campaign.recipient_count-campaign.unique_clicks, "unsubscribe": campaign.unsubscribe_count, }) # ------------------------- # Device Usage # ------------------------- device_queryset = EmailTrackingEvent.objects.filter( event_type="open" ) if start_date: device_queryset = device_queryset.filter( created_at__date__gte=parse_date(start_date) ) if end_date: device_queryset = device_queryset.filter( created_at__date__lte=parse_date(end_date) ) device_queryset = device_queryset.values( "device_type" ).annotate( count=Count("id") ) total_devices = sum(item["count"] for item in device_queryset) device_usage = [] for item in device_queryset: device_usage.append({ "device": item["device_type"], "count": item["count"], "percentage": round( item["count"] * 100 / total_devices, 2 ) if total_devices else 0 }) # ------------------------- # Top Performing Campaigns # ------------------------- top_campaigns = sorted( campaign_comparison, key=lambda x: x["click_rate"], reverse=True ) return Response({ "summary": { "total_campaigns": total_campaigns, "total_sent": total_sent, "total_opens": total_opens, "total_clicks": total_clicks, "average_open_rate": avg_open_rate, "average_click_rate": avg_click_rate, }, "campaign_comparison": campaign_comparison, "device_usage": device_usage, "top_campaigns": top_campaigns, }) import json from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from rest_framework.permissions import IsAuthenticated import json from rest_framework.views import APIView from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework import status class BulkDeleteSubscribersAPIView(APIView): module_name = "newsletter_subscriber" def get_permissions(self): if self.request.method == "POST": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def post(self, request): # DRF parses JSON automatically for POST requests ids = request.data.get("subscriber_ids", []) # Fallback (normally not needed for POST) if not ids: try: body = json.loads(request.body) ids = body.get("subscriber_ids", []) except Exception: ids = [] if not isinstance(ids, list): return Response( {"error": "subscriber_ids must be a list."}, status=status.HTTP_400_BAD_REQUEST, ) if not ids: return Response( {"error": "subscriber_ids is required."}, status=status.HTTP_400_BAD_REQUEST, ) subscribers = NewsletterSubscriber.objects.filter( id__in=ids, is_active=True, ) deleted_count = subscribers.delete() return Response( { "success": True, "message": f"{deleted_count} subscriber(s) deleted successfully.", "deleted_count": deleted_count, }, status=status.HTTP_200_OK, ) import json from rest_framework.views import APIView from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework import status class BulkDeleteLeadsAPIView(APIView): module_name = "lead" def get_permissions(self): if self.request.method == "POST": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def post(self, request): # DRF parses JSON automatically for POST requests ids = request.data.get("lead_ids", []) # Fallback (normally not needed for POST) if not ids: try: body = json.loads(request.body) ids = body.get("lead_ids", []) except Exception: ids = [] if not isinstance(ids, list): return Response( {"error": "lead_ids must be a list."}, status=status.HTTP_400_BAD_REQUEST, ) if not ids: return Response( {"error": "lead_ids is required."}, status=status.HTTP_400_BAD_REQUEST, ) leads = Lead.objects.filter(id__in=ids) deleted_count, _ = leads.delete() return Response( { "success": True, "message": f"{deleted_count} lead(s) deleted successfully.", "deleted_count": deleted_count, }, status=status.HTTP_200_OK, ) from .models import ModalSetting, ModalSettingsUI from .serializers import ModalSettingsUISerializer from rest_framework.parsers import MultiPartParser, FormParser, JSONParser class ModalSettingsUIView(APIView): module_name = "popup_design" parser_classes = (MultiPartParser, FormParser, JSONParser) def get_permissions(self): if self.request.method == "GET": return [AllowAny()] return [IsAuthenticated(), RBACPermission()] def get(self, request): """ Get all UI settings or a single one. ?id= ?modal= """ ui_id = request.query_params.get("id") modal_id = request.query_params.get("modal") if ui_id: try: ui = ModalSettingsUI.objects.select_related("modal").get(id=ui_id) except ModalSettingsUI.DoesNotExist: return Response( {"success": False, "message": "UI settings not found."}, status=status.HTTP_404_NOT_FOUND, ) serializer = ModalSettingsUISerializer( ui, context={"request": request}, ) return Response( { "success": True, "data": serializer.data, } ) if modal_id: try: ui = ModalSettingsUI.objects.select_related("modal").get(modal_id=modal_id) except ModalSettingsUI.DoesNotExist: return Response( {"success": False, "message": "UI settings not found."}, status=status.HTTP_404_NOT_FOUND, ) serializer = ModalSettingsUISerializer( ui, context={"request": request}, ) return Response( { "success": True, "data": serializer.data, } ) queryset = ModalSettingsUI.objects.select_related("modal").all().order_by("-updated_at") serializer = ModalSettingsUISerializer( queryset, many=True, context={"request": request}, ) return Response( { "success": True, "count": queryset.count(), "data": serializer.data, } ) def post(self, request): """ Create UI settings. """ modal_id = request.data.get("modal") # Validate modal id if modal_id in [None, "", "undefined", "null"]: return Response( { "success": False, "message": "Valid Modal ID is required." }, status=status.HTTP_400_BAD_REQUEST, ) # Convert to integer try: modal_id = int(modal_id) except (TypeError, ValueError): return Response( { "success": False, "message": "Invalid Modal ID." }, status=status.HTTP_400_BAD_REQUEST, ) # Check modal exists try: modal = ModalSetting.objects.get(id=modal_id) except ModalSetting.DoesNotExist: return Response( { "success": False, "message": "Modal not found." }, status=status.HTTP_404_NOT_FOUND, ) # Prevent duplicate UI settings if ModalSettingsUI.objects.filter(modal=modal).exists(): return Response( { "success": False, "message": "UI settings already exist for this modal." }, status=status.HTTP_400_BAD_REQUEST, ) # Make request data mutable data = request.data.copy() data["modal"] = modal.id serializer = ModalSettingsUISerializer( data=data, context={"request": request}, ) if serializer.is_valid(): serializer.save() return Response( { "success": True, "message": "UI settings created successfully.", "data": serializer.data, }, status=status.HTTP_201_CREATED, ) return Response( { "success": False, "errors": serializer.errors, }, status=status.HTTP_400_BAD_REQUEST, ) def patch(self, request): logger.warning("CONTENT TYPE: %s", request.content_type) logger.warning("FILES: %s", request.FILES) logger.warning("DATA: %s", request.data) ui_id = request.data.get("id") ui = get_object_or_404(ModalSettingsUI, id=ui_id) data = request.data.copy() # Explicitly include uploaded file if "background_image" in request.FILES: data["background_image"] = request.FILES["background_image"] logger.warning("Received image: %s", request.FILES["background_image"].name) serializer = ModalSettingsUISerializer( ui, data=data, partial=True, context={"request": request}, ) if serializer.is_valid(): instance = serializer.save() logger.warning("VALIDATED DATA: %s", serializer.validated_data) logger.warning("Saved image: %s", instance.background_image) return Response({ "success": True, "data": ModalSettingsUISerializer( instance, context={"request": request} ).data, }) logger.error(serializer.errors) return Response(serializer.errors, status=400) def delete(self, request): """ Delete UI settings. """ ui_id = request.data.get("id") if not ui_id: return Response( { "success": False, "message": "UI ID is required." }, status=status.HTTP_400_BAD_REQUEST, ) try: ui = ModalSettingsUI.objects.get(id=ui_id) except ModalSettingsUI.DoesNotExist: return Response( { "success": False, "message": "UI settings not found." }, status=status.HTTP_404_NOT_FOUND, ) ui.delete() return Response( { "success": True, "message": "UI settings deleted successfully." } ) class SubscriberSourcesAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request): sources = ( NewsletterSubscriber.objects .exclude(source__isnull=True) .exclude(source="") .values_list("source", flat=True) .distinct() .order_by("source") ) return Response(list(sources)) from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from .models import privecy, terms from .serializers import PrivecySerializer, termsSerializer class PrivecyUploadAPIView(APIView): module_name = "website_settings" def get_permissions(self): if self.request.method == 'GET': return [AllowAny()] return [IsAuthenticated, RBACPermission] def get(self, request): qs = privecy.objects.all() serializer = PrivecySerializer(qs, many=True) return Response(serializer.data) def post(self, request): try: # Keep only one Privacy Policy PDF privecy.objects.all().delete() serializer = PrivecySerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) except Exception as e: return Response( {"error": str(e)}, status=500 ) class TermsUploadAPIView(APIView): module_name = "website_settings" def get_permissions(self): if self.request.method == 'GET': return [AllowAny()] return [IsAuthenticated, RBACPermission] def get(self, request): qs = terms.objects.all() serializer = termsSerializer(qs, many=True) return Response(serializer.data) def post(self, request): try: # IMPORTANT: # Delete TERMS, not privacy records terms.objects.all().delete() serializer = termsSerializer(data=request.data) if serializer.is_valid(): serializer.save() return Response(serializer.data) return Response(serializer.errors, status=400) except Exception as e: return Response( {"error": str(e)}, status=500 ) from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from rest_framework.permissions import IsAuthenticated from .models import Watch_Buy from .serializers import WatchBuySerializer class WatchBuyAPIView(APIView): module_name = "watch_buy" def get_permissions(self): if self.request.method == 'GET': return [AllowAny()] return [IsAuthenticated()] def get(self, request): watch_buys = Watch_Buy.objects.select_related( "product" ).order_by("-updated_at") serializer = WatchBuySerializer( watch_buys, many=True, context={"request": request} ) return Response( serializer.data, status=status.HTTP_200_OK ) def post(self, request): serializer = WatchBuySerializer( data=request.data, context={"request": request} ) if serializer.is_valid(): watch_buy = serializer.save() return Response( WatchBuySerializer( watch_buy, context={"request": request} ).data, status=status.HTTP_201_CREATED ) return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) def delete(self, request): watch_buy_id = request.data.get("id") if not watch_buy_id: return Response( { "success": False, "message": "Watch Buy ID is required." }, status=status.HTTP_400_BAD_REQUEST ) try: watch_buy = Watch_Buy.objects.get( id=watch_buy_id ) except Watch_Buy.DoesNotExist: return Response( { "success": False, "message": "Watch Buy entry not found." }, status=status.HTTP_404_NOT_FOUND ) watch_buy.delete() return Response( { "success": True, "message": "Watch Buy entry deleted successfully." }, status=status.HTTP_200_OK ) class OurJourneyAPIView(APIView): module_name = "our_journey" def get_permissions(self): if self.request.method == 'GET': return [AllowAny()] return [IsAuthenticated()] # ========================= # GET # ========================= def get(self, request): journeys = Our_Journey.objects.all().order_by("-id") serializer = OurJourneySerializer( journeys, many=True, context={"request": request} ) return Response( serializer.data, status=status.HTTP_200_OK ) # ========================= # POST # ========================= def post(self, request): serializer = OurJourneySerializer( data=request.data, context={"request": request} ) if serializer.is_valid(): journey = serializer.save() return Response( OurJourneySerializer( journey, context={"request": request} ).data, status=status.HTTP_201_CREATED ) return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) # ========================= # PATCH # ========================= def patch(self, request): journey_id = request.data.get("id") if not journey_id: return Response( { "success": False, "message": "Journey ID is required." }, status=status.HTTP_400_BAD_REQUEST ) try: journey = Our_Journey.objects.get( id=journey_id ) except Our_Journey.DoesNotExist: return Response( { "success": False, "message": "Journey entry not found." }, status=status.HTTP_404_NOT_FOUND ) serializer = OurJourneySerializer( journey, data=request.data, partial=True, context={"request": request} ) if serializer.is_valid(): journey = serializer.save() return Response( OurJourneySerializer( journey, context={"request": request} ).data, status=status.HTTP_200_OK ) return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) # ========================= # DELETE # ========================= def delete(self, request): journey_id = request.data.get("id") if not journey_id: return Response( { "success": False, "message": "Journey ID is required." }, status=status.HTTP_400_BAD_REQUEST ) try: journey = Our_Journey.objects.get( id=journey_id ) except Our_Journey.DoesNotExist: return Response( { "success": False, "message": "Journey entry not found." }, status=status.HTTP_404_NOT_FOUND ) journey.delete() return Response( { "success": True, "message": "Journey entry deleted successfully." }, status=status.HTTP_200_OK ) from .serializers import EmailTrackingEventSerializer class CampaignSorting(APIView): permission_classes = [IsAuthenticated] def get(self, request, campaign_id): campaign = EmailCampaign.objects.filter( id=campaign_id ).first() if not campaign: return Response( {"error": "Campaign not found"}, status=404 ) impressions = ( EmailTrackingEvent.objects .filter(campaign_id=campaign_id, event_type="click") .select_related("campaign", "subscriber") .order_by("-created_at") ) serializer = EmailTrackingEventSerializer( impressions, many=True ) return Response(serializer.data) from django.http import HttpResponse from django.utils.timezone import localtime from rest_framework.views import APIView from openpyxl import Workbook from openpyxl.styles import ( Font, PatternFill, Alignment, Border, Side, ) from .models import EmailCampaign, EmailTrackingEvent class CampaignExcelExportAPIView(APIView): module_name = "campaign_report_export" permission_classes = [IsAuthenticated, RBACPermission] def get_permissions(self): return [IsAuthenticated(), RBACPermission()] def get(self, request): # ========================================================= # GET ALL CAMPAIGNS # ========================================================= campaigns = ( EmailCampaign.objects .filter(removed=False) .order_by("-id") ) # ========================================================= # CREATE WORKBOOK # ========================================================= workbook = Workbook() # Default sheet summary_sheet = workbook.active summary_sheet.title = "Campaign Summary" # ========================================================= # COLORS / STYLES # ========================================================= # Main title title_fill = PatternFill( fill_type="solid", fgColor="17365D" ) title_font = Font( bold=True, color="FFFFFF", size=14 ) # Header header_fill = PatternFill( fill_type="solid", fgColor="1F4E78" ) header_font = Font( bold=True, color="FFFFFF", size=11 ) # Alternate row alternate_fill = PatternFill( fill_type="solid", fgColor="EAF2F8" ) # Sent sent_fill = PatternFill( fill_type="solid", fgColor="D9EAD3" ) # Click click_fill = PatternFill( fill_type="solid", fgColor="D9EAF7" ) # Unsubscribe unsubscribe_fill = PatternFill( fill_type="solid", fgColor="F4CCCC" ) # Border thin_border = Border( left=Side( style="thin", color="D9D9D9" ), right=Side( style="thin", color="D9D9D9" ), top=Side( style="thin", color="D9D9D9" ), bottom=Side( style="thin", color="D9D9D9" ) ) # ========================================================= # CAMPAIGN SUMMARY SHEET # ========================================================= summary_sheet.merge_cells("A1:F1") summary_sheet["A1"] = "Email Campaign Report" summary_sheet["A1"].fill = title_fill summary_sheet["A1"].font = title_font summary_sheet["A1"].alignment = Alignment( horizontal="center", vertical="center" ) summary_sheet.row_dimensions[1].height = 28 # --------------------------------------------------------- # SUMMARY HEADERS # --------------------------------------------------------- summary_headers = [ "Campaign Name", "Date", "Time", "Sent", "Clicked", "Unsubscribe" ] for col_num, header in enumerate( summary_headers, 1 ): cell = summary_sheet.cell( row=3, column=col_num, value=header ) cell.fill = header_fill cell.font = header_font cell.alignment = Alignment( horizontal="center", vertical="center" ) cell.border = thin_border # ========================================================= # CAMPAIGN PROCESSING # ========================================================= summary_row = 4 used_sheet_names = set() for campaign in campaigns: # ===================================================== # ALL EVENTS # # Used ONLY for summary counts. # ===================================================== all_events = ( EmailTrackingEvent.objects .filter( campaign=campaign ) .select_related( "subscriber" ) .order_by("-created_at") ) # ===================================================== # ONLY CLICK EVENTS # # Used for individual campaign sheet. # ===================================================== click_events = ( EmailTrackingEvent.objects .filter( campaign=campaign, event_type="click" ) .select_related( "subscriber" ) .order_by("-created_at") ) # ===================================================== # SUMMARY COUNTS # ===================================================== sent_count = all_events.filter( event_type="sent" ).count() clicked_count = all_events.filter( event_type="click" ).count() unsubscribe_count = all_events.filter( event_type="unsubscribe" ).count() # ===================================================== # CAMPAIGN DATE / TIME # ===================================================== campaign_datetime = getattr( campaign, "sent_at", None ) if campaign_datetime: # Convert timezone-aware datetime # into local timezone campaign_datetime = localtime( campaign_datetime ) campaign_date = ( campaign_datetime.strftime( "%Y-%m-%d" ) ) campaign_time = ( campaign_datetime.strftime( "%H:%M:%S" ) ) else: campaign_date = "" campaign_time = "" # ===================================================== # SUMMARY ROW # ===================================================== summary_values = [ campaign.subject, campaign_date, campaign_time, sent_count, clicked_count, unsubscribe_count ] for col_num, value in enumerate( summary_values, 1 ): cell = summary_sheet.cell( row=summary_row, column=col_num, value=value ) cell.border = thin_border cell.alignment = Alignment( vertical="center" ) # Alternate row if summary_row % 2 == 0: cell.fill = alternate_fill # ----------------------------------------------------- # COLOR SUMMARY COUNTERS # ----------------------------------------------------- # Sent summary_sheet.cell( row=summary_row, column=4 ).fill = sent_fill # Clicked summary_sheet.cell( row=summary_row, column=5 ).fill = click_fill # Unsubscribe summary_sheet.cell( row=summary_row, column=6 ).fill = unsubscribe_fill summary_row += 1 # ===================================================== # CREATE SAFE SHEET NAME # ===================================================== campaign_name = ( campaign.subject or f"Campaign {campaign.id}" ) # Excel does not allow these characters invalid_chars = [ "\\", "/", "*", "?", ":", "[", "]" ] sheet_name = campaign_name for char in invalid_chars: sheet_name = sheet_name.replace( char, "-" ) # Excel sheet max length = 31 sheet_name = sheet_name[:31] if not sheet_name: sheet_name = ( f"Campaign-{campaign.id}" ) original_sheet_name = sheet_name counter = 1 # Make sure duplicate campaign names # don't create duplicate sheet names while sheet_name in used_sheet_names: suffix = f"-{counter}" sheet_name = ( original_sheet_name[ :31 - len(suffix) ] + suffix ) counter += 1 used_sheet_names.add( sheet_name ) # ===================================================== # CREATE CAMPAIGN SHEET # ===================================================== campaign_sheet = workbook.create_sheet( title=sheet_name ) # ===================================================== # CAMPAIGN SHEET TITLE # ===================================================== campaign_sheet.merge_cells( "A1:J1" ) campaign_sheet["A1"] = campaign_name campaign_sheet["A1"].fill = title_fill campaign_sheet["A1"].font = title_font campaign_sheet["A1"].alignment = Alignment( horizontal="center", vertical="center" ) campaign_sheet.row_dimensions[1].height = 28 # ===================================================== # CAMPAIGN SHEET HEADERS # ===================================================== headers = [ "Name / Subscriber", "Subscriber Source", "Email", "Event Type", "IP Address", "User Agent", "Device", "City", "Country", "Created At" ] for col_num, header in enumerate( headers, 1 ): cell = campaign_sheet.cell( row=3, column=col_num, value=header ) cell.fill = header_fill cell.font = header_font cell.alignment = Alignment( horizontal="center", vertical="center", wrap_text=True ) cell.border = thin_border # ===================================================== # CLICK EVENT DATA # ===================================================== row_num = 4 for event in click_events: # ------------------------------------------------- # SUBSCRIBER # ------------------------------------------------- subscriber = event.subscriber if subscriber: subscriber_name = ( subscriber.email or str(subscriber.id) ) subscriber_source = ( subscriber.source or "" ) subscriber_email = ( subscriber.email or "" ) # IMPORTANT: # City comes from NewsletterSubscriber subscriber_city = ( subscriber.city or "" ) # IMPORTANT: # Country comes from NewsletterSubscriber subscriber_country = ( subscriber.country or "" ) else: subscriber_name = "" subscriber_source = "" subscriber_email = "" subscriber_city = "" subscriber_country = "" # ------------------------------------------------- # IP ADDRESS # # IMPORTANT: # IP comes from EmailTrackingEvent # ------------------------------------------------- event_ip = ( event.ip_address or "" ) # ------------------------------------------------- # USER AGENT # ------------------------------------------------- event_user_agent = ( event.user_agent or "" ) # ------------------------------------------------- # DEVICE # ------------------------------------------------- device = ( event.device_type or "" ) # ------------------------------------------------- # EVENT TYPE # # Since this queryset contains only clicks, # this will always be "click". # ------------------------------------------------- event_type = ( event.event_type or "" ) # ------------------------------------------------- # CREATED AT # ------------------------------------------------- created_at = event.created_at if created_at: # Convert UTC/timezone-aware datetime # to local timezone first created_at = localtime( created_at ) # Excel does not support timezone-aware # datetime objects created_at = created_at.replace( tzinfo=None ) # ------------------------------------------------- # ROW DATA # ------------------------------------------------- row_values = [ subscriber_name, subscriber_source, subscriber_email, event_type, event_ip, event_user_agent, device, subscriber_city, subscriber_country, created_at ] # ------------------------------------------------- # WRITE ROW # ------------------------------------------------- for col_num, value in enumerate( row_values, 1 ): cell = campaign_sheet.cell( row=row_num, column=col_num, value=value ) cell.border = thin_border cell.alignment = Alignment( vertical="top", wrap_text=True ) # Alternate rows if row_num % 2 == 0: cell.fill = alternate_fill # ------------------------------------------------- # CLICK EVENT COLOR # ------------------------------------------------- event_cell = campaign_sheet.cell( row=row_num, column=4 ) event_cell.fill = click_fill event_cell.font = Font( bold=True ) # ------------------------------------------------- # DATE FORMAT # ------------------------------------------------- if created_at: created_cell = ( campaign_sheet.cell( row=row_num, column=10 ) ) created_cell.number_format = ( "yyyy-mm-dd hh:mm:ss" ) row_num += 1 # ===================================================== # FILTER # ===================================================== if row_num > 4: campaign_sheet.auto_filter.ref = ( f"A3:J{row_num - 1}" ) # ===================================================== # FREEZE HEADER # ===================================================== campaign_sheet.freeze_panes = "A4" # ===================================================== # COLUMN WIDTHS # ===================================================== column_widths = { "A": 28, # Name "B": 22, # Source "C": 32, # Email "D": 18, # Event "E": 20, # IP "F": 45, # User Agent "G": 15, # Device "H": 20, # City "I": 20, # Country "J": 22, # Created At } for column, width in column_widths.items(): campaign_sheet.column_dimensions[ column ].width = width # ========================================================= # SUMMARY SHEET FORMATTING # ========================================================= summary_sheet.freeze_panes = "A4" if summary_row > 4: summary_sheet.auto_filter.ref = ( f"A3:F{summary_row - 1}" ) # ========================================================= # SUMMARY COLUMN WIDTHS # ========================================================= summary_widths = { "A": 40, # Campaign Name "B": 15, # Date "C": 15, # Time "D": 12, # Sent "E": 12, # Clicked "F": 15, # Unsubscribe } for column, width in summary_widths.items(): summary_sheet.column_dimensions[ column ].width = width # ========================================================= # HTTP RESPONSE # ========================================================= response = HttpResponse( content_type=( "application/vnd.openxmlformats-officedocument." "spreadsheetml.sheet" ) ) response[ "Content-Disposition" ] = ( 'attachment; ' 'filename="campaign_report.xlsx"' ) # ========================================================= # SAVE WORKBOOK # ========================================================= workbook.save(response) return response