Template
westfarn
5b11cc18c7
Closes #9. Shop-gated buyer accounts, purchase reviews, Stripe customer ids, shipment tracking, slim public contact form, and a template-neutral seed_demo command.
385 lines
12 KiB
Python
385 lines
12 KiB
Python
"""Catalog, cart, inventory, and Stripe checkout for FEATURE_SHOP."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
from django.conf import settings
|
||
from django.db import transaction
|
||
from django.db.models import F, Q, Sum
|
||
from django.utils import timezone
|
||
|
||
from shop.models import Order, OrderItem, Product, ProductReview
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CART_SESSION_KEY = "shop_cart"
|
||
|
||
|
||
class ShopError(RuntimeError):
|
||
pass
|
||
|
||
|
||
def _stripe():
|
||
secret = (settings.STRIPE_SECRET_KEY or "").strip()
|
||
if not secret:
|
||
raise ShopError("STRIPE_SECRET_KEY is not configured")
|
||
try:
|
||
import stripe
|
||
except ImportError as exc:
|
||
raise ShopError("stripe package is not installed") from exc
|
||
stripe.api_key = secret
|
||
return stripe
|
||
|
||
|
||
def next_order_number() -> str:
|
||
today = date.today().strftime("%Y%m%d")
|
||
prefix = f"ORD-{today}-"
|
||
existing = Order.objects.filter(number__startswith=prefix).count()
|
||
return f"{prefix}{existing + 1:03d}"
|
||
|
||
|
||
def get_cart(session) -> dict[str, int]:
|
||
raw = session.get(CART_SESSION_KEY) or {}
|
||
cart: dict[str, int] = {}
|
||
for key, qty in raw.items():
|
||
try:
|
||
n = int(qty)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if n > 0:
|
||
cart[str(key)] = n
|
||
return cart
|
||
|
||
|
||
def save_cart(session, cart: dict[str, int]) -> None:
|
||
session[CART_SESSION_KEY] = cart
|
||
session.modified = True
|
||
|
||
|
||
def add_to_cart(session, product: Product, quantity: int = 1) -> dict[str, int]:
|
||
if quantity < 1:
|
||
raise ShopError("Quantity must be at least 1.")
|
||
cart = get_cart(session)
|
||
pid = str(product.pk)
|
||
cart[pid] = cart.get(pid, 0) + quantity
|
||
save_cart(session, cart)
|
||
return cart
|
||
|
||
|
||
def set_cart_qty(session, product: Product, quantity: int) -> dict[str, int]:
|
||
cart = get_cart(session)
|
||
pid = str(product.pk)
|
||
if quantity < 1:
|
||
cart.pop(pid, None)
|
||
else:
|
||
cart[pid] = quantity
|
||
save_cart(session, cart)
|
||
return cart
|
||
|
||
|
||
def cart_lines(session) -> list[dict]:
|
||
cart = get_cart(session)
|
||
products = {
|
||
str(p.pk): p
|
||
for p in Product.objects.filter(pk__in=cart.keys(), is_published=True)
|
||
}
|
||
lines = []
|
||
for pid, qty in cart.items():
|
||
product = products.get(pid)
|
||
if not product:
|
||
continue
|
||
lines.append(
|
||
{
|
||
"product": product,
|
||
"quantity": qty,
|
||
"unit_price": product.price,
|
||
"line_total": product.price * qty,
|
||
}
|
||
)
|
||
return lines
|
||
|
||
|
||
def cart_total(lines: list[dict]) -> Decimal:
|
||
return sum((line["line_total"] for line in lines), Decimal("0"))
|
||
|
||
|
||
def queued_print_minutes() -> int:
|
||
total = (
|
||
OrderItem.objects.filter(
|
||
order__status__in=[Order.Status.OPEN, Order.Status.PAID],
|
||
product__fulfillment=Product.Fulfillment.MADE_TO_ORDER,
|
||
).aggregate(total=Sum(F("print_minutes") * F("quantity")))["total"]
|
||
or 0
|
||
)
|
||
return int(total)
|
||
|
||
|
||
def available_qty(product: Product) -> int | None:
|
||
"""Units that can be sold. None means unlimited (untracked made-to-order)."""
|
||
if not product.track_inventory:
|
||
return None
|
||
if product.fulfillment == Product.Fulfillment.STOCKED:
|
||
return max(product.stock_qty, 0)
|
||
limit = int(getattr(settings, "SHOP_PRINT_QUEUE_LIMIT_MINUTES", 0) or 0)
|
||
if not limit or not product.print_minutes:
|
||
return None
|
||
remaining_minutes = max(limit - queued_print_minutes(), 0)
|
||
return remaining_minutes // product.print_minutes
|
||
|
||
|
||
def assert_can_sell(product: Product, quantity: int) -> None:
|
||
if quantity < 1:
|
||
raise ShopError("Quantity must be at least 1.")
|
||
if not product.is_published:
|
||
raise ShopError("This product is not available.")
|
||
avail = available_qty(product)
|
||
if avail is not None and quantity > avail:
|
||
raise ShopError(f"Only {avail} of {product.name} available.")
|
||
|
||
|
||
def adjust_stock(product: Product, delta: int) -> Product:
|
||
"""Increment (positive) or decrement (negative) on-hand stock."""
|
||
product.stock_qty = product.stock_qty + delta
|
||
if product.stock_qty < 0:
|
||
raise ShopError(f"Insufficient stock for {product.sku}.")
|
||
product.save(update_fields=["stock_qty", "updated_at"])
|
||
return product
|
||
|
||
|
||
def create_order_from_cart(
|
||
session,
|
||
*,
|
||
email: str,
|
||
customer_name: str = "",
|
||
shipping_address: dict | None = None,
|
||
notes: str = "",
|
||
user=None,
|
||
) -> Order:
|
||
lines = cart_lines(session)
|
||
if not lines:
|
||
raise ShopError("Cart is empty.")
|
||
email = (email or "").strip()
|
||
if not email:
|
||
raise ShopError("Email is required.")
|
||
for line in lines:
|
||
assert_can_sell(line["product"], line["quantity"])
|
||
currency = (settings.STRIPE_CURRENCY or "usd").lower()
|
||
with transaction.atomic():
|
||
order = Order.objects.create(
|
||
number=next_order_number(),
|
||
user=user if getattr(user, "is_authenticated", False) else None,
|
||
email=email,
|
||
customer_name=(customer_name or "").strip(),
|
||
status=Order.Status.DRAFT,
|
||
amount=cart_total(lines),
|
||
currency=currency,
|
||
shipping_address=shipping_address or {},
|
||
notes=notes,
|
||
)
|
||
for line in lines:
|
||
product = line["product"]
|
||
OrderItem.objects.create(
|
||
order=order,
|
||
product=product,
|
||
name=product.name,
|
||
sku=product.sku,
|
||
quantity=line["quantity"],
|
||
unit_price=product.price,
|
||
print_minutes=product.print_minutes,
|
||
)
|
||
return order
|
||
|
||
|
||
def ensure_stripe_customer(user) -> str:
|
||
"""Create or reuse a Stripe Customer. Card data never leaves Stripe."""
|
||
if not user or not getattr(user, "is_authenticated", False):
|
||
return ""
|
||
from accounts.services import get_customer_profile
|
||
|
||
profile = get_customer_profile(user)
|
||
if profile.stripe_customer_id:
|
||
return profile.stripe_customer_id
|
||
stripe = _stripe()
|
||
customer = stripe.Customer.create(
|
||
email=(user.email or user.username or "") or None,
|
||
name=(user.get_full_name() or "") or None,
|
||
metadata={"user_id": str(user.pk)},
|
||
)
|
||
customer_id = getattr(customer, "id", None) or customer.get("id") or ""
|
||
if not customer_id:
|
||
raise ShopError("Stripe did not return a customer id.")
|
||
profile.stripe_customer_id = customer_id
|
||
profile.save(update_fields=["stripe_customer_id", "updated_at"])
|
||
return customer_id
|
||
|
||
|
||
def remember_stripe_customer(order: Order, customer_id: str) -> None:
|
||
customer_id = (customer_id or "").strip()
|
||
if not customer_id or not order.user_id:
|
||
return
|
||
from accounts.services import get_customer_profile
|
||
|
||
profile = get_customer_profile(order.user)
|
||
if profile.stripe_customer_id:
|
||
return
|
||
profile.stripe_customer_id = customer_id
|
||
profile.save(update_fields=["stripe_customer_id", "updated_at"])
|
||
|
||
|
||
def create_checkout_session(order: Order, *, success_url: str, cancel_url: str) -> str:
|
||
stripe = _stripe()
|
||
line_items = [
|
||
{
|
||
"quantity": item.quantity,
|
||
"price_data": {
|
||
"currency": (order.currency or "usd").lower(),
|
||
"unit_amount": int(
|
||
(item.unit_price * Decimal("100")).quantize(Decimal("1"))
|
||
),
|
||
"product_data": {"name": item.name},
|
||
},
|
||
}
|
||
for item in order.items.all()
|
||
]
|
||
if not line_items:
|
||
raise ShopError("Order has no items.")
|
||
params = {
|
||
"mode": "payment",
|
||
"line_items": line_items,
|
||
"metadata": {"shop_order_id": str(order.pk), "order_number": order.number},
|
||
"success_url": success_url,
|
||
"cancel_url": cancel_url,
|
||
}
|
||
customer_id = ""
|
||
if order.user_id:
|
||
try:
|
||
customer_id = ensure_stripe_customer(order.user)
|
||
except ShopError:
|
||
raise
|
||
except Exception:
|
||
logger.exception("stripe customer create failed for order %s", order.number)
|
||
if customer_id:
|
||
params["customer"] = customer_id
|
||
params["payment_intent_data"] = {"setup_future_usage": "on_session"}
|
||
else:
|
||
params["customer_email"] = order.email or None
|
||
session = stripe.checkout.Session.create(**params)
|
||
order.stripe_checkout_session_id = session.id
|
||
order.hosted_checkout_url = session.url or ""
|
||
order.status = Order.Status.OPEN
|
||
order.save(
|
||
update_fields=[
|
||
"stripe_checkout_session_id",
|
||
"hosted_checkout_url",
|
||
"status",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return session.url or ""
|
||
|
||
|
||
def _reserve_inventory(order: Order) -> None:
|
||
for item in order.items.select_related("product"):
|
||
product = item.product
|
||
if product is None or not product.track_inventory:
|
||
continue
|
||
if product.fulfillment == Product.Fulfillment.STOCKED:
|
||
adjust_stock(product, -item.quantity)
|
||
|
||
|
||
def _notify_pos(order: Order) -> None:
|
||
from django.apps import apps
|
||
|
||
if not apps.is_installed("pos_sync"):
|
||
return
|
||
from pos_sync.services import enqueue_online_sale
|
||
|
||
enqueue_online_sale(order)
|
||
|
||
|
||
def mark_paid(
|
||
order: Order, *, stripe_id: str = "", stripe_customer_id: str = ""
|
||
) -> None:
|
||
if order.status == Order.Status.PAID:
|
||
return
|
||
with transaction.atomic():
|
||
locked = Order.objects.select_for_update().get(pk=order.pk)
|
||
if locked.status == Order.Status.PAID:
|
||
return
|
||
_reserve_inventory(locked)
|
||
locked.status = Order.Status.PAID
|
||
locked.paid_at = timezone.now()
|
||
locked.save(update_fields=["status", "paid_at", "updated_at"])
|
||
order.refresh_from_db()
|
||
remember_stripe_customer(order, stripe_customer_id)
|
||
try:
|
||
send_order_email(order)
|
||
except Exception:
|
||
logger.exception("order confirmation email failed for %s", order.number)
|
||
try:
|
||
_notify_pos(order)
|
||
except Exception:
|
||
logger.exception("POS notify failed for %s", order.number)
|
||
|
||
|
||
def send_order_email(order: Order) -> bool:
|
||
to_email = (order.email or "").strip()
|
||
if not to_email:
|
||
raise ShopError("Order has no email address")
|
||
from django.core.mail import EmailMultiAlternatives
|
||
|
||
name = order.customer_name or "there"
|
||
amount = f"{order.amount} {order.currency.upper()}"
|
||
lines = "\n".join(
|
||
f"- {item.quantity}× {item.name} ({item.sku})" for item in order.items.all()
|
||
)
|
||
subject = f"Order {order.number} from {settings.SITE_NAME}"
|
||
text = (
|
||
f"Hi {name},\n\n"
|
||
f"We received order {order.number} for {amount}.\n\n"
|
||
f"{lines}\n\nThank you.\n"
|
||
)
|
||
html = (
|
||
f"<p>Hi {name},</p>"
|
||
f"<p>We received order <strong>{order.number}</strong> for "
|
||
f"<strong>{amount}</strong>.</p>"
|
||
f"<pre>{lines}</pre>"
|
||
)
|
||
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
|
||
|
||
|
||
_REVIEWABLE_STATUSES = (Order.Status.PAID, Order.Status.FULFILLED)
|
||
|
||
|
||
def qualifying_order_for_review(user, product: Product) -> Order | None:
|
||
if not user or not getattr(user, "is_authenticated", False):
|
||
return None
|
||
email = (user.email or user.username or "").strip()
|
||
qs = (
|
||
Order.objects.filter(
|
||
items__product=product,
|
||
status__in=_REVIEWABLE_STATUSES,
|
||
)
|
||
.filter(Q(user=user) | Q(email__iexact=email))
|
||
.distinct()
|
||
.order_by("-paid_at", "-created_at")
|
||
)
|
||
return qs.first()
|
||
|
||
|
||
def user_has_reviewed(user, product: Product) -> bool:
|
||
if not user or not getattr(user, "is_authenticated", False):
|
||
return False
|
||
return ProductReview.objects.filter(user=user, product=product).exists()
|