from uuid import UUID from django.utils.translation import gettext_lazy as _ from drf_spectacular.utils import extend_schema_view from rest_framework import status from rest_framework.permissions import AllowAny from rest_framework.response import Response from rest_framework.views import APIView from engine.vibes_auth.docs.drf.emailing import TRACKING_SCHEMA, UNSUBSCRIBE_SCHEMA from engine.vibes_auth.models import User @extend_schema_view(**UNSUBSCRIBE_SCHEMA) class UnsubscribeView(APIView): """ Public endpoint for one-click unsubscribe from email campaigns. Supports both GET (for email client compatibility) and POST (RFC 8058). """ permission_classes = [AllowAny] authentication_classes = [] def get(self, request): """Handle GET request for unsubscribe (email link click).""" return self._process_unsubscribe(request) def post(self, request): """Handle POST request for one-click unsubscribe (RFC 8058).""" return self._process_unsubscribe(request) def _process_unsubscribe(self, request) -> Response: """Process the unsubscribe request.""" token = request.query_params.get("token") or request.data.get("token") if not token: return Response( {"detail": _("Unsubscribe token is required.")}, status=status.HTTP_400_BAD_REQUEST, ) try: token_uuid = UUID(token) except (ValueError, TypeError): return Response( {"detail": _("Invalid unsubscribe token format.")}, status=status.HTTP_400_BAD_REQUEST, ) try: user = User.objects.get(unsubscribe_token=token_uuid) except User.DoesNotExist: return Response( {"detail": _("User not found.")}, status=status.HTTP_404_NOT_FOUND, ) if not user.is_subscribed: return Response( {"detail": _("You are already unsubscribed.")}, status=status.HTTP_200_OK, ) user.is_subscribed = False user.save(update_fields=["is_subscribed", "modified"]) return Response( {"detail": _("You have been successfully unsubscribed from our emails.")}, status=status.HTTP_200_OK, ) @extend_schema_view(**TRACKING_SCHEMA) class TrackingView(APIView): """ Endpoint for tracking email opens and clicks. This is optional - can be used to track engagement metrics. """ permission_classes = [AllowAny] authentication_classes = [] def get(self, request): """Track email open via tracking pixel.""" from django.utils import timezone from engine.vibes_auth.emailing.choices import RecipientStatus from engine.vibes_auth.emailing.models import CampaignRecipient tracking_id = request.query_params.get("tid") if not tracking_id: return Response(status=status.HTTP_404_NOT_FOUND) try: tracking_uuid = UUID(tracking_id) recipient = CampaignRecipient.objects.get(tracking_id=tracking_uuid) if not recipient.opened_at: recipient.opened_at = timezone.now() recipient.status = RecipientStatus.OPENED recipient.save(update_fields=["opened_at", "status", "modified"]) # Update campaign opened count campaign = recipient.campaign campaign.opened_count = campaign.recipients.filter( status__in=(RecipientStatus.OPENED, RecipientStatus.CLICKED) ).count() campaign.save(update_fields=["opened_count", "modified"]) except (ValueError, TypeError, CampaignRecipient.DoesNotExist): pass # Silently ignore invalid tracking IDs # Return a 1x1 transparent GIF gif_data = ( b"GIF89a\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff" b"\x00\x00\x00!\xf9\x04\x01\x00\x00\x00\x00,\x00" b"\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02D\x01\x00;" ) return Response( gif_data, status=status.HTTP_200_OK, content_type="image/gif", )