generated from westfarn/web_django_template
Stand up Print Forge as a 3D-printed toy shop with color variants and printer photography.
Replaces the client_site template branding, adds shop/shipping, and points beta CI at master for easy deploy. Closes #1
This commit is contained in:
@@ -0,0 +1,610 @@
|
||||
"""Catalog, cart, inventory, and Stripe checkout for FEATURE_SHOP."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
import struct
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import transaction
|
||||
from django.db.models import F, Sum
|
||||
from django.utils import timezone
|
||||
|
||||
from core.models import StoredFile
|
||||
from shop.models import Order, OrderItem, Product, ProductColor, ProductImage
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
CART_SESSION_KEY = "shop_cart"
|
||||
_ALLOWED_IMAGE_TYPES = frozenset(
|
||||
{"image/jpeg", "image/png", "image/gif", "image/webp"}
|
||||
)
|
||||
_MAX_IMAGE_BYTES = 15 * 1024 * 1024
|
||||
_ALLOWED_STL_TYPES = frozenset(
|
||||
{
|
||||
"model/stl",
|
||||
"model/x.stl-ascii",
|
||||
"model/x.stl-binary",
|
||||
"application/sla",
|
||||
"application/vnd.ms-pki.stl",
|
||||
"application/octet-stream",
|
||||
"",
|
||||
}
|
||||
)
|
||||
_MAX_STL_BYTES = 25 * 1024 * 1024
|
||||
_HEX_RE = re.compile(r"^#?[0-9A-Fa-f]{6}$")
|
||||
|
||||
|
||||
class ShopError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def _read_upload_bytes(upload) -> bytes:
|
||||
"""Copy the request upload into memory, then drop any temp-file spool."""
|
||||
try:
|
||||
return upload.read()
|
||||
finally:
|
||||
closer = getattr(upload, "close", None)
|
||||
if callable(closer):
|
||||
closer()
|
||||
|
||||
|
||||
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 cart_line_key(product: Product, color: ProductColor | None = None) -> str:
|
||||
if color is None:
|
||||
return str(product.pk)
|
||||
return f"{product.pk}:{color.pk}"
|
||||
|
||||
|
||||
def _split_cart_key(key: str) -> tuple[str, str | None]:
|
||||
text = str(key)
|
||||
if ":" not in text:
|
||||
return text, None
|
||||
product_id, color_id = text.split(":", 1)
|
||||
return product_id, color_id or None
|
||||
|
||||
|
||||
def add_to_cart(
|
||||
session,
|
||||
product: Product,
|
||||
quantity: int = 1,
|
||||
color: ProductColor | None = None,
|
||||
) -> dict[str, int]:
|
||||
if quantity < 1:
|
||||
raise ShopError("Quantity must be at least 1.")
|
||||
cart = get_cart(session)
|
||||
key = cart_line_key(product, color)
|
||||
cart[key] = cart.get(key, 0) + quantity
|
||||
save_cart(session, cart)
|
||||
return cart
|
||||
|
||||
|
||||
def set_cart_qty(
|
||||
session,
|
||||
product: Product,
|
||||
quantity: int,
|
||||
color: ProductColor | None = None,
|
||||
) -> dict[str, int]:
|
||||
cart = get_cart(session)
|
||||
key = cart_line_key(product, color)
|
||||
if quantity < 1:
|
||||
cart.pop(key, None)
|
||||
else:
|
||||
cart[key] = quantity
|
||||
save_cart(session, cart)
|
||||
return cart
|
||||
|
||||
|
||||
def cart_lines(session) -> list[dict]:
|
||||
cart = get_cart(session)
|
||||
parsed: list[tuple[str, str | None, int]] = []
|
||||
product_ids: list[str] = []
|
||||
color_ids: list[str] = []
|
||||
for key, qty in cart.items():
|
||||
product_id, color_id = _split_cart_key(key)
|
||||
product_ids.append(product_id)
|
||||
if color_id:
|
||||
color_ids.append(color_id)
|
||||
parsed.append((product_id, color_id, qty))
|
||||
products = {
|
||||
str(p.pk): p
|
||||
for p in Product.objects.filter(
|
||||
pk__in=product_ids, is_published=True
|
||||
).select_related("image")
|
||||
}
|
||||
colors = {
|
||||
str(c.pk): c
|
||||
for c in ProductColor.objects.filter(pk__in=color_ids)
|
||||
.select_related("product")
|
||||
.prefetch_related("images")
|
||||
}
|
||||
lines = []
|
||||
for product_id, color_id, qty in parsed:
|
||||
product = products.get(product_id)
|
||||
if not product:
|
||||
continue
|
||||
color = colors.get(color_id) if color_id else None
|
||||
if color_id and (
|
||||
color is None or str(color.product_id) != product_id
|
||||
):
|
||||
continue
|
||||
image_url = ""
|
||||
if color is not None:
|
||||
image_url = color.image_url
|
||||
if not image_url:
|
||||
image_url = product.image_url
|
||||
lines.append(
|
||||
{
|
||||
"product": product,
|
||||
"color": color,
|
||||
"quantity": qty,
|
||||
"unit_price": product.price,
|
||||
"line_total": product.price * qty,
|
||||
"image_url": image_url,
|
||||
}
|
||||
)
|
||||
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, color: ProductColor | None = None
|
||||
) -> 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:
|
||||
if color is not None:
|
||||
return max(color.stock_qty, 0)
|
||||
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, color: ProductColor | None = None
|
||||
) -> 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, color=color)
|
||||
if avail is not None and quantity > avail:
|
||||
label = f"{product.name} ({color.name})" if color else product.name
|
||||
raise ShopError(f"Only {avail} of {label} available.")
|
||||
|
||||
|
||||
def adjust_stock(
|
||||
product: Product, delta: int, color: ProductColor | None = None
|
||||
) -> Product | ProductColor:
|
||||
"""Increment (positive) or decrement (negative) on-hand stock."""
|
||||
if color is not None:
|
||||
color.stock_qty = color.stock_qty + delta
|
||||
if color.stock_qty < 0:
|
||||
raise ShopError(f"Insufficient stock for {product.sku} ({color.name}).")
|
||||
color.save(update_fields=["stock_qty", "updated_at"])
|
||||
return color
|
||||
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 = "",
|
||||
) -> 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"], color=line.get("color"))
|
||||
currency = (settings.STRIPE_CURRENCY or "usd").lower()
|
||||
with transaction.atomic():
|
||||
order = Order.objects.create(
|
||||
number=next_order_number(),
|
||||
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"]
|
||||
color = line.get("color")
|
||||
display_name = (
|
||||
f"{product.name} ({color.name})" if color else product.name
|
||||
)
|
||||
OrderItem.objects.create(
|
||||
order=order,
|
||||
product=product,
|
||||
color=color,
|
||||
name=display_name,
|
||||
sku=product.sku,
|
||||
color_name=color.name if color else "",
|
||||
quantity=line["quantity"],
|
||||
unit_price=product.price,
|
||||
print_minutes=product.print_minutes,
|
||||
)
|
||||
return order
|
||||
|
||||
|
||||
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.")
|
||||
session = stripe.checkout.Session.create(
|
||||
mode="payment",
|
||||
customer_email=order.email or None,
|
||||
line_items=line_items,
|
||||
metadata={"shop_order_id": str(order.pk), "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 = 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", "color"):
|
||||
product = item.product
|
||||
if product is None or not product.track_inventory:
|
||||
continue
|
||||
if product.fulfillment == Product.Fulfillment.STOCKED:
|
||||
adjust_stock(product, -item.quantity, color=item.color)
|
||||
|
||||
|
||||
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 = "") -> 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()
|
||||
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
|
||||
|
||||
|
||||
def normalize_hex(value: str) -> str:
|
||||
raw = (value or "").strip()
|
||||
if not _HEX_RE.fullmatch(raw):
|
||||
return "#808080"
|
||||
if not raw.startswith("#"):
|
||||
raw = f"#{raw}"
|
||||
return raw.lower()
|
||||
|
||||
|
||||
def store_product_image(*, upload, user) -> StoredFile:
|
||||
content_type = (getattr(upload, "content_type", None) or "").lower()
|
||||
if content_type not in _ALLOWED_IMAGE_TYPES:
|
||||
raise ShopError("Use a JPEG, PNG, GIF, or WebP image.")
|
||||
data = _read_upload_bytes(upload)
|
||||
if len(data) > _MAX_IMAGE_BYTES:
|
||||
raise ShopError("Image must be 15 MB or smaller.")
|
||||
original = (getattr(upload, "name", None) or "product")[:255]
|
||||
try:
|
||||
from shop.imaging import prepare_product_photo, product_image_filename
|
||||
except ImportError as exc:
|
||||
logger.exception("product photo processing dependencies missing")
|
||||
raise ShopError(
|
||||
"Image processing is not installed. Rebuild the app container."
|
||||
) from exc
|
||||
try:
|
||||
data, content_type = prepare_product_photo(data)
|
||||
original = product_image_filename(original)
|
||||
except ShopError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.exception("product photo processing failed")
|
||||
raise ShopError("Could not process that image.") from exc
|
||||
return StoredFile.objects.create(
|
||||
kind=StoredFile.Kind.PRODUCT_IMAGE,
|
||||
filename=original,
|
||||
content_type=content_type,
|
||||
size=len(data),
|
||||
data=data,
|
||||
uploaded_by=user if getattr(user, "is_authenticated", False) else None,
|
||||
)
|
||||
|
||||
|
||||
def looks_like_stl(data: bytes, filename: str = "") -> bool:
|
||||
if not (filename or "").lower().endswith(".stl"):
|
||||
return False
|
||||
if len(data) < 84:
|
||||
return False
|
||||
head = data[:80].lstrip().lower()
|
||||
sample = data[:8192].lower()
|
||||
if head.startswith(b"solid") and b"facet" in sample:
|
||||
return True
|
||||
triangle_count = struct.unpack_from("<I", data, 80)[0]
|
||||
if triangle_count < 1:
|
||||
return False
|
||||
return 84 + triangle_count * 50 == len(data)
|
||||
|
||||
|
||||
def store_product_stl(*, upload, user) -> StoredFile:
|
||||
original = (getattr(upload, "name", None) or "model.stl")[:255]
|
||||
content_type = (getattr(upload, "content_type", None) or "").lower()
|
||||
if content_type not in _ALLOWED_STL_TYPES:
|
||||
raise ShopError("Use an STL file.")
|
||||
data = _read_upload_bytes(upload)
|
||||
if len(data) > _MAX_STL_BYTES:
|
||||
raise ShopError("STL must be 25 MB or smaller.")
|
||||
if not looks_like_stl(data, original):
|
||||
raise ShopError("That file does not look like a valid STL.")
|
||||
return StoredFile.objects.create(
|
||||
kind=StoredFile.Kind.PRODUCT_STL,
|
||||
filename=original,
|
||||
content_type="model/stl",
|
||||
size=len(data),
|
||||
data=data,
|
||||
uploaded_by=user if getattr(user, "is_authenticated", False) else None,
|
||||
)
|
||||
|
||||
|
||||
def sync_product_colors(
|
||||
product: Product,
|
||||
*,
|
||||
ids: list[str],
|
||||
names: list[str],
|
||||
hexes: list[str],
|
||||
keys: list[str] | None = None,
|
||||
stocks: list[str] | None = None,
|
||||
) -> dict[str, ProductColor]:
|
||||
keep: list[str] = []
|
||||
mapping: dict[str, ProductColor] = {}
|
||||
for index, raw_name in enumerate(names):
|
||||
name = (raw_name or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
hex_value = normalize_hex(hexes[index] if index < len(hexes) else "")
|
||||
color_id = (ids[index] if index < len(ids) else "").strip()
|
||||
key = ""
|
||||
if keys and index < len(keys):
|
||||
key = (keys[index] or "").strip()
|
||||
try:
|
||||
stock = int((stocks[index] if stocks and index < len(stocks) else "0") or "0")
|
||||
except (TypeError, ValueError):
|
||||
stock = 0
|
||||
color = None
|
||||
if color_id:
|
||||
color = ProductColor.objects.filter(pk=color_id, product=product).first()
|
||||
if color is None:
|
||||
color = ProductColor(product=product)
|
||||
color.name = name[:64]
|
||||
color.hex = hex_value
|
||||
color.stock_qty = max(stock, 0)
|
||||
color.sort_order = index
|
||||
color.save()
|
||||
keep.append(str(color.pk))
|
||||
mapping[str(color.pk)] = color
|
||||
if key:
|
||||
mapping[key] = color
|
||||
leftover = product.colors.exclude(pk__in=keep)
|
||||
stale_file_ids = list(
|
||||
ProductImage.objects.filter(color__in=leftover).values_list("file_id", flat=True)
|
||||
)
|
||||
leftover.delete()
|
||||
gc_product_image_files(stale_file_ids)
|
||||
return mapping
|
||||
|
||||
|
||||
def refresh_listing_image(product: Product) -> None:
|
||||
first = (
|
||||
product.images.filter(color_id__isnull=True)
|
||||
.order_by("sort_order", "created_at")
|
||||
.first()
|
||||
or product.images.order_by("sort_order", "created_at").first()
|
||||
)
|
||||
new_id = first.file_id if first else None
|
||||
if product.image_id != new_id:
|
||||
product.image_id = new_id
|
||||
product.save(update_fields=["image", "updated_at"])
|
||||
|
||||
|
||||
def gc_product_image_files(file_ids) -> None:
|
||||
ids = [item for item in file_ids if item]
|
||||
if not ids:
|
||||
return
|
||||
used = set(
|
||||
ProductImage.objects.filter(file_id__in=ids).values_list("file_id", flat=True)
|
||||
)
|
||||
used.update(
|
||||
Product.objects.filter(image_id__in=ids).values_list("image_id", flat=True)
|
||||
)
|
||||
stale = [pk for pk in ids if pk not in used]
|
||||
if stale:
|
||||
StoredFile.objects.filter(
|
||||
pk__in=stale, kind=StoredFile.Kind.PRODUCT_IMAGE
|
||||
).delete()
|
||||
|
||||
|
||||
def append_product_images(
|
||||
product: Product,
|
||||
*,
|
||||
uploads,
|
||||
user,
|
||||
color: ProductColor | None = None,
|
||||
) -> None:
|
||||
existing = product.images.filter(color=color).count()
|
||||
for offset, upload in enumerate(uploads):
|
||||
if not upload:
|
||||
continue
|
||||
stored = store_product_image(upload=upload, user=user)
|
||||
ProductImage.objects.create(
|
||||
product=product,
|
||||
color=color,
|
||||
file=stored,
|
||||
sort_order=existing + offset,
|
||||
)
|
||||
|
||||
|
||||
def remove_product_images(product: Product, image_ids: list[str]) -> None:
|
||||
ids = [item for item in image_ids if item]
|
||||
if not ids:
|
||||
return
|
||||
qs = ProductImage.objects.filter(product=product, pk__in=ids)
|
||||
file_ids = list(qs.values_list("file_id", flat=True))
|
||||
qs.delete()
|
||||
gc_product_image_files(file_ids)
|
||||
|
||||
|
||||
def product_media_payload(product: Product, colors: list[ProductColor]) -> dict:
|
||||
images = list(product.images.all())
|
||||
shared = [item.url for item in images if item.color_id is None]
|
||||
if not shared and product.image_url:
|
||||
shared = [product.image_url]
|
||||
color_data = {}
|
||||
for color in colors:
|
||||
colored = [item.url for item in images if item.color_id == color.pk]
|
||||
color_data[str(color.pk)] = {
|
||||
"name": color.name,
|
||||
"hex": color.hex,
|
||||
"images": colored,
|
||||
"available": available_qty(product, color),
|
||||
}
|
||||
return {"shared": shared, "stl": product.stl_url, "colors": color_data}
|
||||
Reference in New Issue
Block a user