263 lines
9.9 KiB
Python
263 lines
9.9 KiB
Python
"""Admin invoice / subscription views and Stripe webhook."""
|
|
|
|
import json
|
|
import logging
|
|
|
|
from django.conf import settings
|
|
from django.contrib import messages
|
|
from django.http import HttpResponse, HttpResponseBadRequest
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.views.decorators.http import require_POST
|
|
|
|
from .forms import OneOffInvoiceForm, RecurringSubscriptionForm
|
|
from .models import BillingCustomer, Invoice, RecurringSubscription
|
|
from .permissions import financial_admin_required
|
|
from .stripe_billing import (
|
|
StripeAPIError,
|
|
StripeNotConfigured,
|
|
apply_checkout_session_completed,
|
|
apply_stripe_invoice_event,
|
|
apply_subscription_event,
|
|
construct_webhook_event,
|
|
create_one_off_invoice,
|
|
create_subscription_checkout,
|
|
send_pay_link_email,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@financial_admin_required
|
|
def invoice_list(request):
|
|
invoices = Invoice.objects.select_related("customer").all()[:100]
|
|
subscriptions = RecurringSubscription.objects.select_related("customer").all()[:100]
|
|
customers = BillingCustomer.objects.all()[:50]
|
|
return render(
|
|
request,
|
|
"financial/invoices.html",
|
|
{
|
|
"invoices": invoices,
|
|
"subscriptions": subscriptions,
|
|
"customers": customers,
|
|
"stripe_configured": bool(getattr(settings, "STRIPE_SECRET_KEY", "")),
|
|
},
|
|
)
|
|
|
|
|
|
@financial_admin_required
|
|
def invoice_new(request):
|
|
if request.method == "POST":
|
|
form = OneOffInvoiceForm(request.POST)
|
|
if form.is_valid():
|
|
invoice = form.build_invoice(request.user)
|
|
invoice.save()
|
|
try:
|
|
create_one_off_invoice(invoice)
|
|
if form.cleaned_data.get("send_email") and invoice.hosted_invoice_url:
|
|
send_pay_link_email(
|
|
to_email=invoice.customer.email,
|
|
customer_name=invoice.customer.name,
|
|
description=invoice.description,
|
|
amount_dollars=invoice.amount_dollars,
|
|
pay_url=invoice.hosted_invoice_url,
|
|
kind="invoice",
|
|
)
|
|
invoice.pay_link_emailed_at = timezone.now()
|
|
invoice.save(update_fields=["pay_link_emailed_at"])
|
|
messages.success(
|
|
request,
|
|
f"Invoice created and payment link emailed to {invoice.customer.email}.",
|
|
)
|
|
else:
|
|
messages.success(
|
|
request,
|
|
"Invoice created in Stripe. Payment link saved on the detail page.",
|
|
)
|
|
return redirect("invoice_detail", invoice_id=invoice.pk)
|
|
except StripeNotConfigured as exc:
|
|
messages.error(request, str(exc))
|
|
except StripeAPIError as exc:
|
|
messages.error(request, f"Stripe error: {exc.user_message}")
|
|
else:
|
|
form = OneOffInvoiceForm()
|
|
|
|
return render(request, "financial/invoice_new.html", {"form": form})
|
|
|
|
|
|
@financial_admin_required
|
|
def invoice_detail(request, invoice_id):
|
|
invoice = get_object_or_404(
|
|
Invoice.objects.select_related("customer"), pk=invoice_id
|
|
)
|
|
return render(request, "financial/invoice_detail.html", {"invoice": invoice})
|
|
|
|
|
|
@financial_admin_required
|
|
def invoice_resend_email(request, invoice_id):
|
|
invoice = get_object_or_404(
|
|
Invoice.objects.select_related("customer"), pk=invoice_id
|
|
)
|
|
if request.method != "POST":
|
|
return redirect("invoice_detail", invoice_id=invoice.pk)
|
|
if not invoice.hosted_invoice_url:
|
|
messages.error(request, "No payment link available for this invoice.")
|
|
return redirect("invoice_detail", invoice_id=invoice.pk)
|
|
try:
|
|
send_pay_link_email(
|
|
to_email=invoice.customer.email,
|
|
customer_name=invoice.customer.name,
|
|
description=invoice.description,
|
|
amount_dollars=invoice.amount_dollars,
|
|
pay_url=invoice.hosted_invoice_url,
|
|
kind="invoice",
|
|
)
|
|
invoice.pay_link_emailed_at = timezone.now()
|
|
invoice.save(update_fields=["pay_link_emailed_at"])
|
|
messages.success(request, f"Payment link resent to {invoice.customer.email}.")
|
|
except Exception as exc: # noqa: BLE001 — surface mail errors to admin UI
|
|
messages.error(request, f"Email failed: {exc}")
|
|
return redirect("invoice_detail", invoice_id=invoice.pk)
|
|
|
|
|
|
@financial_admin_required
|
|
def subscription_new(request):
|
|
if request.method == "POST":
|
|
form = RecurringSubscriptionForm(request.POST)
|
|
if form.is_valid():
|
|
subscription = form.build_subscription(request.user)
|
|
subscription.save()
|
|
success_url = (
|
|
request.build_absolute_uri(
|
|
reverse("subscription_detail", args=[subscription.pk])
|
|
)
|
|
+ "?checkout=success"
|
|
)
|
|
cancel_url = (
|
|
request.build_absolute_uri(
|
|
reverse("subscription_detail", args=[subscription.pk])
|
|
)
|
|
+ "?checkout=cancel"
|
|
)
|
|
try:
|
|
create_subscription_checkout(
|
|
subscription,
|
|
success_url=success_url,
|
|
cancel_url=cancel_url,
|
|
)
|
|
if form.cleaned_data.get("send_email") and subscription.checkout_url:
|
|
send_pay_link_email(
|
|
to_email=subscription.customer.email,
|
|
customer_name=subscription.customer.name,
|
|
description=subscription.description,
|
|
amount_dollars=subscription.amount_dollars,
|
|
pay_url=subscription.checkout_url,
|
|
kind="subscription",
|
|
)
|
|
subscription.pay_link_emailed_at = timezone.now()
|
|
subscription.save(update_fields=["pay_link_emailed_at"])
|
|
messages.success(
|
|
request,
|
|
f"Subscription checkout created and emailed to {subscription.customer.email}.",
|
|
)
|
|
else:
|
|
messages.success(
|
|
request,
|
|
"Subscription checkout created. Share the link from the detail page.",
|
|
)
|
|
return redirect("subscription_detail", subscription_id=subscription.pk)
|
|
except StripeNotConfigured as exc:
|
|
messages.error(request, str(exc))
|
|
except StripeAPIError as exc:
|
|
messages.error(request, f"Stripe error: {exc.user_message}")
|
|
else:
|
|
form = RecurringSubscriptionForm()
|
|
|
|
return render(request, "financial/subscription_new.html", {"form": form})
|
|
|
|
|
|
@financial_admin_required
|
|
def subscription_detail(request, subscription_id):
|
|
subscription = get_object_or_404(
|
|
RecurringSubscription.objects.select_related("customer"),
|
|
pk=subscription_id,
|
|
)
|
|
return render(
|
|
request,
|
|
"financial/subscription_detail.html",
|
|
{"subscription": subscription},
|
|
)
|
|
|
|
|
|
@financial_admin_required
|
|
def subscription_resend_email(request, subscription_id):
|
|
subscription = get_object_or_404(
|
|
RecurringSubscription.objects.select_related("customer"),
|
|
pk=subscription_id,
|
|
)
|
|
if request.method != "POST":
|
|
return redirect("subscription_detail", subscription_id=subscription.pk)
|
|
if not subscription.checkout_url:
|
|
messages.error(request, "No checkout link available for this subscription.")
|
|
return redirect("subscription_detail", subscription_id=subscription.pk)
|
|
try:
|
|
send_pay_link_email(
|
|
to_email=subscription.customer.email,
|
|
customer_name=subscription.customer.name,
|
|
description=subscription.description,
|
|
amount_dollars=subscription.amount_dollars,
|
|
pay_url=subscription.checkout_url,
|
|
kind="subscription",
|
|
)
|
|
subscription.pay_link_emailed_at = timezone.now()
|
|
subscription.save(update_fields=["pay_link_emailed_at"])
|
|
messages.success(
|
|
request, f"Checkout link resent to {subscription.customer.email}."
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
messages.error(request, f"Email failed: {exc}")
|
|
return redirect("subscription_detail", subscription_id=subscription.pk)
|
|
|
|
|
|
@csrf_exempt
|
|
@require_POST
|
|
def stripe_webhook(request):
|
|
"""Receive Stripe webhooks; verify signature when secret is configured."""
|
|
payload = request.body
|
|
sig_header = request.META.get("HTTP_STRIPE_SIGNATURE", "")
|
|
webhook_secret = getattr(settings, "STRIPE_WEBHOOK_SECRET", "") or ""
|
|
|
|
try:
|
|
if webhook_secret:
|
|
event = construct_webhook_event(payload, sig_header, webhook_secret)
|
|
else:
|
|
logger.warning(
|
|
"STRIPE_WEBHOOK_SECRET unset — accepting unsigned webhook (dev only)"
|
|
)
|
|
event = json.loads(payload.decode("utf-8"))
|
|
except (ValueError, json.JSONDecodeError) as exc:
|
|
return HttpResponseBadRequest(str(exc))
|
|
|
|
event_type = event.get("type")
|
|
data_object = event.get("data", {}).get("object", {})
|
|
|
|
if event_type in {
|
|
"invoice.paid",
|
|
"invoice.payment_failed",
|
|
"invoice.finalized",
|
|
"invoice.voided",
|
|
"invoice.marked_uncollectible",
|
|
}:
|
|
apply_stripe_invoice_event(data_object)
|
|
elif event_type in {
|
|
"customer.subscription.updated",
|
|
"customer.subscription.deleted",
|
|
}:
|
|
apply_subscription_event(data_object)
|
|
elif event_type == "checkout.session.completed":
|
|
apply_checkout_session_completed(data_object)
|
|
|
|
return HttpResponse(status=200)
|