web_django_template/site/events/services.py

"""Event capacity, Stripe ticket checkout, and confirmation email."""

from __future__ import annotations

import logging
import secrets
from datetime import date
from decimal import Decimal

from django.conf import settings
from django.db import transaction
from django.utils import timezone

from events.models import Event, Ticket, TicketOrder

logger = logging.getLogger(__name__)


class EventsError(RuntimeError):
    pass


def _stripe():
    secret = (settings.STRIPE_SECRET_KEY or "").strip()
    if not secret:
        raise EventsError("STRIPE_SECRET_KEY is not configured")
    try:
        import stripe
    except ImportError as exc:
        raise EventsError("stripe package is not installed") from exc
    stripe.api_key = secret
    return stripe


def next_ticket_order_number() -> str:
    today = date.today().strftime("%Y%m%d")
    prefix = f"TIX-{today}-"
    existing = TicketOrder.objects.filter(number__startswith=prefix).count()
    return f"{prefix}{existing + 1:03d}"


def new_ticket_code() -> str:
    while True:
        code = secrets.token_hex(4).upper()
        if not Ticket.objects.filter(code=code).exists():
            return code


def assert_capacity(
    event: Event, quantity: int, *, exclude_order_id=None
) -> None:
    if quantity < 1:
        raise EventsError("Quantity must be at least 1.")
    if not event.capacity:
        return
    held = event.tickets_held(exclude_order_id=exclude_order_id)
    remaining = max(event.capacity - held, 0)
    if quantity > remaining:
        raise EventsError(f"Only {remaining} seats left for {event.title}.")


def create_ticket_order(
    event: Event, *, email: str, customer_name: str = "", quantity: int = 1
) -> TicketOrder:
    email = (email or "").strip()
    if not email:
        raise EventsError("Email is required.")
    if not event.is_published:
        raise EventsError("This event is not on sale.")
    assert_capacity(event, quantity)
    amount = event.price * quantity
    return TicketOrder.objects.create(
        number=next_ticket_order_number(),
        event=event,
        email=email,
        customer_name=(customer_name or "").strip(),
        quantity=quantity,
        amount=amount,
        currency=(event.currency or settings.STRIPE_CURRENCY or "usd").lower(),
        status=TicketOrder.Status.DRAFT,
    )


def create_checkout_session(
    order: TicketOrder, *, success_url: str, cancel_url: str
) -> str:
    stripe = _stripe()
    session = stripe.checkout.Session.create(
        mode="payment",
        customer_email=order.email or None,
        line_items=[
            {
                "quantity": order.quantity,
                "price_data": {
                    "currency": (order.currency or "usd").lower(),
                    "unit_amount": int(
                        (order.event.price * Decimal("100")).quantize(Decimal("1"))
                    ),
                    "product_data": {"name": order.event.title},
                },
            }
        ],
        metadata={
            "ticket_order_id": str(order.pk),
            "ticket_order_number": order.number,
        },
        success_url=success_url,
        cancel_url=cancel_url,
    )
    order.stripe_checkout_session_id = session.id
    order.hosted_checkout_url = session.url or ""
    order.status = TicketOrder.Status.OPEN
    order.save(
        update_fields=[
            "stripe_checkout_session_id",
            "hosted_checkout_url",
            "status",
            "updated_at",
        ]
    )
    return session.url or ""


def _issue_tickets(order: TicketOrder) -> list[Ticket]:
    created = []
    for _ in range(order.quantity):
        created.append(
            Ticket.objects.create(
                order=order,
                event=order.event,
                code=new_ticket_code(),
                attendee_name=order.customer_name,
            )
        )
    return created


def mark_paid(order: TicketOrder, *, stripe_id: str = "") -> None:
    if order.status == TicketOrder.Status.PAID:
        return
    with transaction.atomic():
        locked = TicketOrder.objects.select_for_update().get(pk=order.pk)
        if locked.status == TicketOrder.Status.PAID:
            return
        assert_capacity(
            locked.event, locked.quantity, exclude_order_id=locked.pk
        )
        _issue_tickets(locked)
        locked.status = TicketOrder.Status.PAID
        locked.paid_at = timezone.now()
        locked.save(update_fields=["status", "paid_at", "updated_at"])
    order.refresh_from_db()
    try:
        send_ticket_email(order)
    except Exception:
        logger.exception("ticket email failed for %s", order.number)


def send_ticket_email(order: TicketOrder) -> bool:
    to_email = (order.email or "").strip()
    if not to_email:
        raise EventsError("Order has no email address")
    from django.core.mail import EmailMultiAlternatives

    codes = ", ".join(t.code for t in order.tickets.all())
    name = order.customer_name or "there"
    when = timezone.localtime(order.event.starts_at).strftime("%b %d, %Y %-I:%M %p")
    subject = f"Tickets for {order.event.title} — {settings.SITE_NAME}"
    text = (
        f"Hi {name},\n\n"
        f"Your tickets for {order.event.title} on {when}:\n"
        f"{codes}\n\nOrder {order.number}\n"
    )
    html = (
        f"<p>Hi {name},</p>"
        f"<p>Tickets for <strong>{order.event.title}</strong> on {when}:</p>"
        f"<p><strong>{codes}</strong></p>"
        f"<p>Order {order.number}</p>"
    )
    mail = EmailMultiAlternatives(
        subject=subject,
        body=text,
        from_email=settings.DEFAULT_FROM_EMAIL,
        to=[to_email],
    )
    mail.attach_alternative(html, "text/html")
    mail.send(fail_silently=False)
    return True