79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from django.core import signing
|
|
from django.core.mail import EmailMultiAlternatives
|
|
from django.http import Http404, JsonResponse
|
|
from django.shortcuts import get_object_or_404, redirect
|
|
from django.template.loader import render_to_string
|
|
from django.urls import reverse
|
|
from django.views import View
|
|
|
|
from apps.newsletter.forms import SubscriptionForm
|
|
from apps.newsletter.models import NewsletterSubscription
|
|
from apps.newsletter.services import ProviderSyncError, get_provider_service
|
|
|
|
CONFIRMATION_TOKEN_MAX_AGE_SECONDS = 60 * 60 * 24 * 2
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def confirmation_token(email: str) -> str:
|
|
return signing.dumps(email, salt="newsletter-confirm")
|
|
|
|
|
|
def send_confirmation_email(request, subscription: NewsletterSubscription) -> None:
|
|
token = confirmation_token(subscription.email)
|
|
confirm_url = request.build_absolute_uri(reverse("newsletter_confirm", args=[token]))
|
|
context = {"confirmation_url": confirm_url, "subscription": subscription}
|
|
subject = render_to_string("newsletter/email/confirmation_subject.txt", context).strip()
|
|
text_body = render_to_string("newsletter/email/confirmation_body.txt", context)
|
|
html_body = render_to_string("newsletter/email/confirmation_body.html", context)
|
|
message = EmailMultiAlternatives(
|
|
subject=subject,
|
|
body=text_body,
|
|
to=[subscription.email],
|
|
)
|
|
message.attach_alternative(html_body, "text/html")
|
|
message.send()
|
|
|
|
|
|
class SubscribeView(View):
|
|
def post(self, request):
|
|
form = SubscriptionForm(request.POST)
|
|
if not form.is_valid():
|
|
return JsonResponse({"status": "error", "field": "email"}, status=400)
|
|
if form.cleaned_data.get("honeypot"):
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
email = form.cleaned_data["email"].lower().strip()
|
|
source = form.cleaned_data.get("source") or "unknown"
|
|
subscription, created = NewsletterSubscription.objects.get_or_create(
|
|
email=email,
|
|
defaults={"source": source},
|
|
)
|
|
if created and not subscription.confirmed:
|
|
send_confirmation_email(request, subscription)
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
class ConfirmView(View):
|
|
def get(self, request, token: str):
|
|
try:
|
|
email = signing.loads(
|
|
token,
|
|
salt="newsletter-confirm",
|
|
max_age=CONFIRMATION_TOKEN_MAX_AGE_SECONDS,
|
|
)
|
|
except signing.BadSignature as exc:
|
|
raise Http404 from exc
|
|
subscription = get_object_or_404(NewsletterSubscription, email=email)
|
|
subscription.confirmed = True
|
|
subscription.save(update_fields=["confirmed"])
|
|
service = get_provider_service()
|
|
try:
|
|
service.sync(subscription)
|
|
except ProviderSyncError as exc:
|
|
logger.exception("Newsletter provider sync failed: %s", exc)
|
|
return redirect("/")
|