westfarn
03bb01664f
Unit Tests / test (push) Successful in 12s
## Summary - Closes #26 - Shared dark branded email base (`emails/base_email.html`) matching site tokens: `#0a0a0a` / `#1a1a1a`, cyan `#00f3ff`, purple `#bc13fe`, logo, Inter-safe stack - Refactors contact, marketing (+ preview), and invoice pay-link HTML/text to extend that base; drops Materialize CDN - Adds `PUBLIC_SITE_URL` + `email_brand_context()` for absolute logo/footer links ## Email renders ### Contact  ### Marketing  ### Invoice pay link  Also see `docs/email-previews/`. ## Test plan - [ ] Preview a marketing `EmailMessage` in admin / `/preview_email/<pk>/` — dark brand, logo, no Materialize - [ ] Submit contact form (or render `emails/contact_email.html`) — subject/from/message fields + branded footer - [ ] Send (or dry-render) invoice pay-link email — cyan CTA, amount accent - [ ] Confirm logo loads from `PUBLIC_SITE_URL/static/public/img/logo.png` in a real clientReviewed-on: #27
386 lines
13 KiB
Python
386 lines
13 KiB
Python
"""Stripe REST helpers (stdlib HTTP) for invoices and subscriptions.
|
|
|
|
Customers pay us — one-off hosted invoices and Checkout subscription sessions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
|
|
from django.conf import settings
|
|
from django.core.mail import EmailMultiAlternatives
|
|
from django.template.loader import get_template
|
|
from django.utils import timezone
|
|
|
|
from .models import BillingCustomer, Invoice, RecurringSubscription
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
STRIPE_API_BASE = "https://api.stripe.com/v1"
|
|
|
|
|
|
class StripeNotConfigured(Exception):
|
|
"""Raised when Stripe secret key is missing."""
|
|
|
|
|
|
class StripeAPIError(Exception):
|
|
"""Raised when Stripe API returns an error."""
|
|
|
|
def __init__(self, message: str, status: int | None = None):
|
|
super().__init__(message)
|
|
self.status = status
|
|
self.user_message = message
|
|
|
|
|
|
def _secret_key() -> str:
|
|
key = getattr(settings, "STRIPE_SECRET_KEY", "") or ""
|
|
if not key:
|
|
raise StripeNotConfigured(
|
|
"STRIPE_SECRET_KEY is not set. Add it to the environment."
|
|
)
|
|
return key
|
|
|
|
|
|
def _stripe_request(method: str, path: str, data: dict | None = None) -> dict:
|
|
"""Call Stripe API with form-encoded body (Stripe default)."""
|
|
url = f"{STRIPE_API_BASE}{path}"
|
|
body = None
|
|
headers = {
|
|
"Authorization": f"Bearer {_secret_key()}",
|
|
}
|
|
if data is not None:
|
|
body = urllib.parse.urlencode(_flatten_params(data)).encode("utf-8")
|
|
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
|
|
|
request = urllib.request.Request(url, data=body, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as exc:
|
|
raw = exc.read().decode("utf-8", errors="replace")
|
|
try:
|
|
payload = json.loads(raw)
|
|
message = payload.get("error", {}).get("message") or raw
|
|
except json.JSONDecodeError:
|
|
message = raw or str(exc)
|
|
raise StripeAPIError(message, status=exc.code) from exc
|
|
except urllib.error.URLError as exc:
|
|
raise StripeAPIError(f"Could not reach Stripe: {exc.reason}") from exc
|
|
|
|
|
|
def _flatten_params(data: dict, prefix: str = "") -> list[tuple[str, str]]:
|
|
"""Flatten nested dicts/lists into Stripe form keys (a[b]=1, items[0][x]=y)."""
|
|
items: list[tuple[str, str]] = []
|
|
for key, value in data.items():
|
|
full_key = f"{prefix}[{key}]" if prefix else str(key)
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, dict):
|
|
items.extend(_flatten_params(value, full_key))
|
|
elif isinstance(value, (list, tuple)):
|
|
for index, entry in enumerate(value):
|
|
indexed = f"{full_key}[{index}]"
|
|
if isinstance(entry, dict):
|
|
items.extend(_flatten_params(entry, indexed))
|
|
else:
|
|
items.append((indexed, str(entry)))
|
|
elif isinstance(value, bool):
|
|
items.append((full_key, "true" if value else "false"))
|
|
else:
|
|
items.append((full_key, str(value)))
|
|
return items
|
|
|
|
|
|
def construct_webhook_event(payload: bytes, sig_header: str, secret: str) -> dict:
|
|
"""Verify Stripe-Signature and return event dict."""
|
|
if not sig_header:
|
|
raise ValueError("Missing Stripe-Signature header")
|
|
|
|
elements = {part.split("=", 1)[0]: part.split("=", 1)[1] for part in sig_header.split(",") if "=" in part}
|
|
timestamp = elements.get("t")
|
|
signatures = [part.split("=", 1)[1] for part in sig_header.split(",") if part.startswith("v1=")]
|
|
if not timestamp or not signatures:
|
|
raise ValueError("Invalid Stripe-Signature header")
|
|
|
|
signed_payload = f"{timestamp}.{payload.decode('utf-8')}".encode("utf-8")
|
|
expected = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
|
|
if not any(hmac.compare_digest(expected, candidate) for candidate in signatures):
|
|
raise ValueError("Webhook signature verification failed")
|
|
|
|
# Reject stale timestamps (>5 minutes)
|
|
if abs(time.time() - int(timestamp)) > 300:
|
|
raise ValueError("Webhook timestamp outside tolerance")
|
|
|
|
return json.loads(payload.decode("utf-8"))
|
|
|
|
|
|
def ensure_stripe_customer(customer: BillingCustomer) -> str:
|
|
"""Create Stripe Customer if needed; return stripe_customer_id."""
|
|
if customer.stripe_customer_id:
|
|
return customer.stripe_customer_id
|
|
|
|
stripe_customer = _stripe_request(
|
|
"POST",
|
|
"/customers",
|
|
{
|
|
"email": customer.email,
|
|
"name": customer.name,
|
|
"metadata": {
|
|
"local_customer_id": str(customer.pk),
|
|
"company": customer.company or "",
|
|
},
|
|
},
|
|
)
|
|
customer.stripe_customer_id = stripe_customer["id"]
|
|
customer.save(update_fields=["stripe_customer_id", "last_modified"])
|
|
return customer.stripe_customer_id
|
|
|
|
|
|
def create_one_off_invoice(invoice: Invoice) -> Invoice:
|
|
"""Create + finalize a Stripe Invoice; store hosted pay URL."""
|
|
stripe_customer_id = ensure_stripe_customer(invoice.customer)
|
|
|
|
_stripe_request(
|
|
"POST",
|
|
"/invoiceitems",
|
|
{
|
|
"customer": stripe_customer_id,
|
|
"amount": invoice.amount_cents,
|
|
"currency": invoice.currency,
|
|
"description": invoice.description,
|
|
},
|
|
)
|
|
|
|
create_data: dict = {
|
|
"customer": stripe_customer_id,
|
|
"collection_method": "send_invoice",
|
|
"auto_advance": True,
|
|
"metadata": {"local_invoice_id": str(invoice.pk)},
|
|
}
|
|
if invoice.due_date:
|
|
due_dt = datetime.combine(invoice.due_date, datetime.min.time())
|
|
if timezone.is_naive(due_dt):
|
|
due_dt = timezone.make_aware(due_dt, timezone.get_current_timezone())
|
|
create_data["due_date"] = int(due_dt.timestamp())
|
|
else:
|
|
create_data["days_until_due"] = 30
|
|
|
|
stripe_invoice = _stripe_request("POST", "/invoices", create_data)
|
|
stripe_invoice = _stripe_request(
|
|
"POST", f"/invoices/{stripe_invoice['id']}/finalize", {}
|
|
)
|
|
|
|
invoice.stripe_invoice_id = stripe_invoice["id"]
|
|
invoice.hosted_invoice_url = stripe_invoice.get("hosted_invoice_url") or ""
|
|
invoice.invoice_pdf_url = stripe_invoice.get("invoice_pdf") or ""
|
|
status = stripe_invoice.get("status") or Invoice.Status.OPEN
|
|
if status in dict(Invoice.Status.choices):
|
|
invoice.status = status
|
|
invoice.last_modified = timezone.now()
|
|
invoice.save(
|
|
update_fields=[
|
|
"stripe_invoice_id",
|
|
"hosted_invoice_url",
|
|
"invoice_pdf_url",
|
|
"status",
|
|
"last_modified",
|
|
]
|
|
)
|
|
return invoice
|
|
|
|
|
|
def create_subscription_checkout(
|
|
subscription: RecurringSubscription,
|
|
*,
|
|
success_url: str,
|
|
cancel_url: str,
|
|
) -> RecurringSubscription:
|
|
"""Create Stripe Product/Price + Checkout Session (subscription mode)."""
|
|
stripe_customer_id = ensure_stripe_customer(subscription.customer)
|
|
|
|
product = _stripe_request(
|
|
"POST",
|
|
"/products",
|
|
{
|
|
"name": subscription.description,
|
|
"metadata": {"local_subscription_id": str(subscription.pk)},
|
|
},
|
|
)
|
|
price = _stripe_request(
|
|
"POST",
|
|
"/prices",
|
|
{
|
|
"product": product["id"],
|
|
"unit_amount": subscription.amount_cents,
|
|
"currency": subscription.currency,
|
|
"recurring": {"interval": subscription.interval},
|
|
},
|
|
)
|
|
session = _stripe_request(
|
|
"POST",
|
|
"/checkout/sessions",
|
|
{
|
|
"mode": "subscription",
|
|
"customer": stripe_customer_id,
|
|
"line_items": [{"price": price["id"], "quantity": 1}],
|
|
"success_url": success_url,
|
|
"cancel_url": cancel_url,
|
|
"metadata": {"local_subscription_id": str(subscription.pk)},
|
|
"subscription_data": {
|
|
"metadata": {"local_subscription_id": str(subscription.pk)},
|
|
},
|
|
},
|
|
)
|
|
|
|
subscription.stripe_product_id = product["id"]
|
|
subscription.stripe_price_id = price["id"]
|
|
subscription.checkout_session_id = session["id"]
|
|
subscription.checkout_url = session.get("url") or ""
|
|
subscription.status = RecurringSubscription.Status.INCOMPLETE
|
|
subscription.last_modified = timezone.now()
|
|
subscription.save(
|
|
update_fields=[
|
|
"stripe_product_id",
|
|
"stripe_price_id",
|
|
"checkout_session_id",
|
|
"checkout_url",
|
|
"status",
|
|
"last_modified",
|
|
]
|
|
)
|
|
return subscription
|
|
|
|
|
|
def dollars_to_cents(amount: Decimal | float | str) -> int:
|
|
return int(round(Decimal(str(amount)) * 100))
|
|
|
|
|
|
def send_pay_link_email(
|
|
*,
|
|
to_email: str,
|
|
customer_name: str,
|
|
description: str,
|
|
amount_dollars: float,
|
|
pay_url: str,
|
|
kind: str = "invoice",
|
|
) -> None:
|
|
"""Email customer a Stripe payment / checkout link."""
|
|
subject = (
|
|
f"Payment request from AI ML Operations — {description}"
|
|
if kind == "invoice"
|
|
else f"Subscription signup — {description}"
|
|
)
|
|
from_email = getattr(
|
|
settings,
|
|
"DEFAULT_FROM_EMAIL",
|
|
"AI ML Operations, LLC <info@aimloperations.com>",
|
|
)
|
|
from public.email_branding import email_brand_context
|
|
|
|
context = email_brand_context(
|
|
customer_name=customer_name,
|
|
description=description,
|
|
amount_dollars=f"{amount_dollars:.2f}",
|
|
pay_url=pay_url,
|
|
kind=kind,
|
|
subject=subject,
|
|
)
|
|
html_content = get_template("emails/invoice_pay_link.html").render(context)
|
|
text_content = get_template("emails/invoice_pay_link.txt").render(context)
|
|
msg = EmailMultiAlternatives(subject, text_content, from_email, [to_email])
|
|
msg.attach_alternative(html_content, "text/html")
|
|
msg.send(fail_silently=False)
|
|
|
|
|
|
def apply_stripe_invoice_event(stripe_invoice: dict) -> Invoice | None:
|
|
"""Sync local Invoice from Stripe invoice object dict."""
|
|
local_id = (stripe_invoice.get("metadata") or {}).get("local_invoice_id")
|
|
stripe_id = stripe_invoice.get("id")
|
|
invoice = None
|
|
if local_id:
|
|
invoice = Invoice.objects.filter(pk=local_id).first()
|
|
if invoice is None and stripe_id:
|
|
invoice = Invoice.objects.filter(stripe_invoice_id=stripe_id).first()
|
|
if invoice is None:
|
|
return None
|
|
|
|
status = stripe_invoice.get("status") or invoice.status
|
|
if status in dict(Invoice.Status.choices):
|
|
invoice.status = status
|
|
invoice.hosted_invoice_url = (
|
|
stripe_invoice.get("hosted_invoice_url") or invoice.hosted_invoice_url
|
|
)
|
|
invoice.invoice_pdf_url = (
|
|
stripe_invoice.get("invoice_pdf") or invoice.invoice_pdf_url
|
|
)
|
|
if stripe_id:
|
|
invoice.stripe_invoice_id = stripe_id
|
|
invoice.last_modified = timezone.now()
|
|
invoice.save(
|
|
update_fields=[
|
|
"status",
|
|
"hosted_invoice_url",
|
|
"invoice_pdf_url",
|
|
"stripe_invoice_id",
|
|
"last_modified",
|
|
]
|
|
)
|
|
return invoice
|
|
|
|
|
|
def apply_subscription_event(stripe_sub: dict) -> RecurringSubscription | None:
|
|
"""Sync local RecurringSubscription from Stripe subscription object."""
|
|
local_id = (stripe_sub.get("metadata") or {}).get("local_subscription_id")
|
|
stripe_id = stripe_sub.get("id")
|
|
sub = None
|
|
if local_id:
|
|
sub = RecurringSubscription.objects.filter(pk=local_id).first()
|
|
if sub is None and stripe_id:
|
|
sub = RecurringSubscription.objects.filter(
|
|
stripe_subscription_id=stripe_id
|
|
).first()
|
|
if sub is None:
|
|
return None
|
|
|
|
status = stripe_sub.get("status") or sub.status
|
|
if status in dict(RecurringSubscription.Status.choices):
|
|
sub.status = status
|
|
if stripe_id:
|
|
sub.stripe_subscription_id = stripe_id
|
|
sub.last_modified = timezone.now()
|
|
sub.save(update_fields=["status", "stripe_subscription_id", "last_modified"])
|
|
return sub
|
|
|
|
|
|
def apply_checkout_session_completed(session: dict) -> RecurringSubscription | None:
|
|
"""Attach subscription id after Checkout completes."""
|
|
if session.get("mode") != "subscription":
|
|
return None
|
|
local_id = (session.get("metadata") or {}).get("local_subscription_id")
|
|
sub = None
|
|
if local_id:
|
|
sub = RecurringSubscription.objects.filter(pk=local_id).first()
|
|
if sub is None and session.get("id"):
|
|
sub = RecurringSubscription.objects.filter(
|
|
checkout_session_id=session["id"]
|
|
).first()
|
|
if sub is None:
|
|
return None
|
|
|
|
stripe_sub_id = session.get("subscription")
|
|
if stripe_sub_id:
|
|
sub.stripe_subscription_id = stripe_sub_id
|
|
sub.status = RecurringSubscription.Status.ACTIVE
|
|
sub.last_modified = timezone.now()
|
|
sub.save(update_fields=["stripe_subscription_id", "status", "last_modified"])
|
|
return sub
|