generated from westfarn/web_django_template
westfarn
dd37a2a268
## Summary - Slim the public contact form to email, interest, and message. Name, phone, and address live on the customer profile instead. - Customers can register, sign in, save shipping details, and view order history. Logged-in checkout creates a Stripe Customer and saves cards on Stripe (`setup_future_usage`); we only store `stripe_customer_id`. - Shipment tracking: EasyPost tracker lookup + webhook, plus paste-in numbers from Pirate Ship/Shippo. Customers see carrier status on their orders; `dispatch_due` refreshes open shipments. - Product reviews (1–5) only after a paid/fulfilled purchase of that product. Fixes #7 ## Test plan - [ ] Contact form submits with only email + message; extra name/phone/address fields are ignored - [ ] Register, sign in, save profile (name/phone/shipping) - [ ] Guest checkout still works; after signup, prior orders with that email show in history - [ ] Logged-in checkout prefills shipping and does not collect card data locally - [ ] Portal: buy label or paste a Pirate Ship tracking number, confirm status/events; customer order page shows tracking - [ ] Product page: non-buyers cannot review; buyers can leave one 1–5 star review - [ ] Non-staff users hitting `/portal/` redirect to `/account/` Reviewed-on: #8
696 lines
22 KiB
Python
696 lines
22 KiB
Python
"""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, Q, Sum
|
||
from django.utils import timezone
|
||
|
||
from core.models import StoredFile
|
||
from shop.models import Order, OrderItem, Product, ProductColor, ProductImage, ProductReview
|
||
|
||
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 = "",
|
||
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"], color=line.get("color"))
|
||
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"]
|
||
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 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", "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 = "", 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
|
||
|
||
|
||
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, smart_crop: bool = True) -> 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 open_image, 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:
|
||
if smart_crop:
|
||
data, content_type = prepare_product_photo(data)
|
||
original = product_image_filename(original)
|
||
else:
|
||
open_image(data)
|
||
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,
|
||
smart_crop: bool = True,
|
||
) -> 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, smart_crop=smart_crop
|
||
)
|
||
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}
|
||
|
||
|
||
_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()
|