Egor fureunoir Gorbunov
20473818a9
Includes detailed OpenAPI schemas for unsubscribe (GET and POST) and tracking pixel (GET) endpoints, supporting email compatibility and event tracking. Added support for RFC 8058-compliant one-click unsubscribe functionality and transparent image-based email tracking.
126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
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",
|
|
)
|