Add shop, POS sync, event ticketing, and shipping catalog apps (#6)

## Summary
- Closes #5
- Optional `shop`, `pos_sync`, `events`, and `shipping` apps gated by `FEATURE_*` flags
- Deps match the catalog: shop/events need email + Stripe; POS/shipping need shop
- Portal inventory, POS webhooks, capacity tickets, EasyPost/Pirate Ship shipping

## Test plan
- [ ] `manage.py test` (152 passed locally)
- [ ] Shop cart + paid order decrements stocked inventory
- [ ] POS inbound webhook decrements SKU; paid order queues outbound reserve
- [ ] Event capacity blocks overbook; paid order emails ticket codes
- [ ] Shipping stub label + Pirate Ship CSV of unshipped paid orders
- [ ] `validate-env.sh` rejects shop without email/payments, POS/shipping without shop

Reviewed-on: #6
This commit was merged in pull request #6.
This commit is contained in:
2026-09-06 14:00:22 -07:00
parent 97b8607bf2
commit 9cdce7a897
82 changed files with 3908 additions and 1 deletions
+188
View File
@@ -0,0 +1,188 @@
"""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