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
526 lines
18 KiB
Python
526 lines
18 KiB
Python
import logging
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
from django.conf import settings
|
|
from django.contrib import messages
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.db import transaction
|
|
from django.db.models import Avg, Count, Prefetch
|
|
from django.http import HttpResponse, HttpResponseBadRequest
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.urls import reverse
|
|
from django.utils.text import slugify
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.views.decorators.http import require_http_methods, require_POST
|
|
|
|
from contacts.models import Contact
|
|
from shop.models import Order, Product, ProductColor, ProductImage, ProductReview
|
|
from shop.services import (
|
|
ShopError,
|
|
add_to_cart,
|
|
adjust_stock,
|
|
append_product_images,
|
|
available_qty,
|
|
cart_lines,
|
|
cart_total,
|
|
create_checkout_session,
|
|
create_order_from_cart,
|
|
mark_paid,
|
|
product_media_payload,
|
|
qualifying_order_for_review,
|
|
refresh_listing_image,
|
|
remove_product_images,
|
|
save_cart,
|
|
set_cart_qty,
|
|
store_product_stl,
|
|
sync_product_colors,
|
|
user_has_reviewed,
|
|
)
|
|
from shop.stats import sales_dashboard
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _delete_replaced_file(previous, current_id, kind):
|
|
if previous and previous.pk != current_id and previous.kind == kind:
|
|
previous.delete()
|
|
|
|
|
|
def _site_base(request) -> str:
|
|
base = (settings.PUBLIC_SITE_URL or "").rstrip("/")
|
|
if base:
|
|
return base
|
|
return request.build_absolute_uri("/").rstrip("/")
|
|
|
|
|
|
def _gallery_prefetch():
|
|
def images():
|
|
return ProductImage.objects.select_related("file").order_by(
|
|
"sort_order", "created_at"
|
|
)
|
|
|
|
return (
|
|
Prefetch("images", queryset=images()),
|
|
Prefetch(
|
|
"colors",
|
|
queryset=ProductColor.objects.prefetch_related(
|
|
Prefetch("images", queryset=images())
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def product_list(request):
|
|
products = Product.objects.filter(is_published=True).select_related(
|
|
"image"
|
|
).prefetch_related("colors")
|
|
return render(request, "shop/list.html", {"products": products})
|
|
|
|
|
|
def product_detail(request, slug):
|
|
product = get_object_or_404(
|
|
Product.objects.select_related("image", "stl").prefetch_related(
|
|
*_gallery_prefetch()
|
|
),
|
|
slug=slug,
|
|
is_published=True,
|
|
)
|
|
colors = list(product.colors.all())
|
|
selected = colors[0] if colors else None
|
|
reviews = list(product.reviews.select_related("user").all()[:50])
|
|
stats = product.reviews.aggregate(avg=Avg("rating"), n=Count("id"))
|
|
can_review = False
|
|
already_reviewed = False
|
|
if request.user.is_authenticated:
|
|
already_reviewed = user_has_reviewed(request.user, product)
|
|
can_review = (
|
|
not already_reviewed
|
|
and qualifying_order_for_review(request.user, product) is not None
|
|
)
|
|
return render(
|
|
request,
|
|
"shop/detail.html",
|
|
{
|
|
"product": product,
|
|
"colors": colors,
|
|
"selected_color": selected,
|
|
"available": available_qty(product, selected),
|
|
"gallery_items": product.gallery_items(selected),
|
|
"gallery_photos": product.photos_for(selected),
|
|
"gallery_data": product_media_payload(product, colors),
|
|
"reviews": reviews,
|
|
"review_avg": stats["avg"],
|
|
"review_count": stats["n"] or 0,
|
|
"can_review": can_review,
|
|
"already_reviewed": already_reviewed,
|
|
},
|
|
)
|
|
|
|
|
|
@login_required(login_url="account:login")
|
|
@require_POST
|
|
def product_review(request, slug):
|
|
product = get_object_or_404(Product, slug=slug, is_published=True)
|
|
if user_has_reviewed(request.user, product):
|
|
messages.info(request, "You already reviewed this product.")
|
|
return redirect("shop:detail", slug=product.slug)
|
|
order = qualifying_order_for_review(request.user, product)
|
|
if order is None:
|
|
messages.error(request, "Only customers who purchased this product can review it.")
|
|
return redirect("shop:detail", slug=product.slug)
|
|
try:
|
|
rating = int(request.POST.get("rating") or "0")
|
|
except ValueError:
|
|
rating = 0
|
|
if rating < 1 or rating > 5:
|
|
messages.error(request, "Choose a rating from 1 to 5.")
|
|
return redirect("shop:detail", slug=product.slug)
|
|
title = (request.POST.get("title") or "").strip()[:120]
|
|
body = (request.POST.get("body") or "").strip()
|
|
ProductReview.objects.create(
|
|
product=product,
|
|
user=request.user,
|
|
order=order,
|
|
rating=rating,
|
|
title=title,
|
|
body=body,
|
|
)
|
|
messages.success(request, "Thanks for the review.")
|
|
return redirect("shop:detail", slug=product.slug)
|
|
|
|
|
|
def cart_view(request):
|
|
lines = cart_lines(request.session)
|
|
return render(
|
|
request,
|
|
"shop/cart.html",
|
|
{"lines": lines, "total": cart_total(lines)},
|
|
)
|
|
|
|
|
|
@require_POST
|
|
def cart_add(request, slug):
|
|
product = get_object_or_404(
|
|
Product.objects.prefetch_related("colors"),
|
|
slug=slug,
|
|
is_published=True,
|
|
)
|
|
try:
|
|
qty = int(request.POST.get("quantity") or "1")
|
|
except ValueError:
|
|
qty = 1
|
|
color = None
|
|
colors = list(product.colors.all())
|
|
if colors:
|
|
color_id = (request.POST.get("color") or "").strip()
|
|
color = next((item for item in colors if str(item.pk) == color_id), None)
|
|
if color is None:
|
|
messages.error(request, "Choose a color.")
|
|
return redirect("shop:detail", slug=product.slug)
|
|
try:
|
|
add_to_cart(request.session, product, qty, color=color)
|
|
except ShopError as exc:
|
|
messages.error(request, str(exc))
|
|
return redirect("shop:detail", slug=product.slug)
|
|
label = f"{product.name} ({color.name})" if color else product.name
|
|
messages.success(request, f"Added {label} to cart.")
|
|
return redirect("shop:cart")
|
|
|
|
|
|
@require_POST
|
|
def cart_update(request, slug):
|
|
product = get_object_or_404(Product, slug=slug)
|
|
try:
|
|
qty = int(request.POST.get("quantity") or "0")
|
|
except ValueError:
|
|
qty = 0
|
|
color = None
|
|
color_id = (request.POST.get("color") or "").strip()
|
|
if color_id:
|
|
color = get_object_or_404(ProductColor, pk=color_id, product=product)
|
|
set_cart_qty(request.session, product, qty, color=color)
|
|
return redirect("shop:cart")
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def checkout(request):
|
|
lines = cart_lines(request.session)
|
|
if not lines:
|
|
messages.error(request, "Cart is empty.")
|
|
return redirect("shop:cart")
|
|
if request.method == "POST":
|
|
email = (request.POST.get("email") or "").strip()
|
|
name = (request.POST.get("customer_name") or "").strip()
|
|
address = Contact.make_postal_address(
|
|
line1=request.POST.get("address_line1") or "",
|
|
line2=request.POST.get("address_line2") or "",
|
|
city=request.POST.get("address_city") or "",
|
|
state=request.POST.get("address_state") or "",
|
|
zip_code=request.POST.get("address_zip") or "",
|
|
)
|
|
buyer = request.user if request.user.is_authenticated else None
|
|
if buyer:
|
|
email = (buyer.email or buyer.username or email).strip()
|
|
if not name:
|
|
name = buyer.get_full_name()
|
|
try:
|
|
order = create_order_from_cart(
|
|
request.session,
|
|
email=email,
|
|
customer_name=name,
|
|
shipping_address=address,
|
|
user=buyer,
|
|
)
|
|
base = _site_base(request)
|
|
success = base + reverse("shop:checkout_success", kwargs={"pk": order.pk})
|
|
cancel = base + reverse("shop:checkout_cancel", kwargs={"pk": order.pk})
|
|
url = create_checkout_session(
|
|
order,
|
|
success_url=success + "?session_id={CHECKOUT_SESSION_ID}",
|
|
cancel_url=cancel,
|
|
)
|
|
except ShopError as exc:
|
|
messages.error(request, str(exc))
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("shop checkout failed")
|
|
messages.error(request, f"Could not start checkout: {exc}")
|
|
else:
|
|
save_cart(request.session, {})
|
|
return redirect(url)
|
|
checkout_initial = {
|
|
"email": "",
|
|
"customer_name": "",
|
|
"address": {},
|
|
}
|
|
if request.user.is_authenticated:
|
|
from accounts.services import get_customer_profile
|
|
|
|
profile = get_customer_profile(request.user)
|
|
checkout_initial = {
|
|
"email": request.user.email or request.user.username,
|
|
"customer_name": request.user.get_full_name(),
|
|
"address": profile.shipping_address or {},
|
|
}
|
|
return render(
|
|
request,
|
|
"shop/checkout.html",
|
|
{
|
|
"lines": lines,
|
|
"total": cart_total(lines),
|
|
"checkout_initial": checkout_initial,
|
|
},
|
|
)
|
|
|
|
|
|
def checkout_success(request, pk):
|
|
order = get_object_or_404(Order, pk=pk)
|
|
return render(request, "shop/success.html", {"order": order})
|
|
|
|
|
|
def checkout_cancel(request, pk):
|
|
order = get_object_or_404(Order, pk=pk)
|
|
return render(request, "shop/cancel.html", {"order": order})
|
|
|
|
|
|
@login_required
|
|
def portal_product_list(request):
|
|
products = Product.objects.select_related("image").prefetch_related("colors")
|
|
return render(request, "shop/portal/products.html", {"products": products})
|
|
|
|
|
|
@login_required
|
|
@require_http_methods(["GET", "POST"])
|
|
def portal_product_edit(request, pk=None):
|
|
product = (
|
|
get_object_or_404(
|
|
Product.objects.select_related("image", "stl").prefetch_related(
|
|
*_gallery_prefetch()
|
|
),
|
|
pk=pk,
|
|
)
|
|
if pk
|
|
else None
|
|
)
|
|
if request.method == "POST":
|
|
name = (request.POST.get("name") or "").strip()
|
|
sku = (request.POST.get("sku") or "").strip()
|
|
description = (request.POST.get("description") or "").strip()
|
|
slug = (request.POST.get("slug") or "").strip()
|
|
fulfillment = request.POST.get("fulfillment") or Product.Fulfillment.STOCKED
|
|
errors = []
|
|
if not name:
|
|
errors.append("Name is required.")
|
|
try:
|
|
price = Decimal(request.POST.get("price") or "")
|
|
if price < 0:
|
|
raise InvalidOperation
|
|
except Exception:
|
|
price = None
|
|
errors.append("Enter a valid price.")
|
|
try:
|
|
stock_qty = int(request.POST.get("stock_qty") or "0")
|
|
except ValueError:
|
|
stock_qty = 0
|
|
errors.append("Stock must be a number.")
|
|
try:
|
|
print_minutes = int(request.POST.get("print_minutes") or "0")
|
|
filament_grams = int(request.POST.get("filament_grams") or "0")
|
|
except ValueError:
|
|
print_minutes = 0
|
|
filament_grams = 0
|
|
stl_upload = request.FILES.get("stl")
|
|
if errors:
|
|
for err in errors:
|
|
messages.error(request, err)
|
|
else:
|
|
try:
|
|
with transaction.atomic():
|
|
stored_stl = None
|
|
if stl_upload:
|
|
stored_stl = store_product_stl(
|
|
upload=stl_upload, user=request.user
|
|
)
|
|
if product is None:
|
|
product = Product()
|
|
previous_stl = product.stl
|
|
product.name = name
|
|
product.sku = sku
|
|
product.description = description
|
|
product.slug = slugify(slug)[:220] if slug else ""
|
|
product.price = price
|
|
product.currency = (settings.STRIPE_CURRENCY or "usd").lower()
|
|
product.fulfillment = fulfillment
|
|
product.stock_qty = stock_qty
|
|
product.print_minutes = max(print_minutes, 0)
|
|
product.filament_grams = max(filament_grams, 0)
|
|
product.is_published = request.POST.get("is_published") == "on"
|
|
product.track_inventory = request.POST.get("track_inventory") == "on"
|
|
if stored_stl is not None:
|
|
product.stl = stored_stl
|
|
elif request.POST.get("clear_stl") == "on":
|
|
product.stl = None
|
|
product.save()
|
|
colors = sync_product_colors(
|
|
product,
|
|
ids=request.POST.getlist("color_id"),
|
|
names=request.POST.getlist("color_name"),
|
|
hexes=request.POST.getlist("color_hex"),
|
|
keys=request.POST.getlist("color_key"),
|
|
stocks=request.POST.getlist("color_stock"),
|
|
)
|
|
remove_product_images(
|
|
product, request.POST.getlist("remove_image")
|
|
)
|
|
smart_crop = request.POST.get("smart_crop", "on") == "on"
|
|
append_product_images(
|
|
product,
|
|
uploads=request.FILES.getlist("images"),
|
|
user=request.user,
|
|
smart_crop=smart_crop,
|
|
)
|
|
for key, color in colors.items():
|
|
uploads = request.FILES.getlist(f"color_images_{key}")
|
|
if not uploads:
|
|
continue
|
|
append_product_images(
|
|
product,
|
|
uploads=uploads,
|
|
user=request.user,
|
|
color=color,
|
|
smart_crop=smart_crop,
|
|
)
|
|
refresh_listing_image(product)
|
|
_delete_replaced_file(
|
|
previous_stl,
|
|
product.stl_id,
|
|
previous_stl.Kind.PRODUCT_STL if previous_stl else None,
|
|
)
|
|
except ShopError as exc:
|
|
messages.error(request, str(exc))
|
|
else:
|
|
messages.success(request, f"Saved {product.name}.")
|
|
return redirect("shop_portal:product_list")
|
|
return render(request, "shop/portal/product_edit.html", {"product": product})
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def portal_stock_adjust(request, pk):
|
|
product = get_object_or_404(Product, pk=pk)
|
|
try:
|
|
delta = int(request.POST.get("delta") or "0")
|
|
except ValueError:
|
|
messages.error(request, "Enter a whole-number adjustment.")
|
|
return redirect("shop_portal:product_edit", pk=product.pk)
|
|
try:
|
|
adjust_stock(product, delta)
|
|
except ShopError as exc:
|
|
messages.error(request, str(exc))
|
|
else:
|
|
messages.success(request, f"{product.sku} stock is now {product.stock_qty}.")
|
|
return redirect("shop_portal:product_edit", pk=product.pk)
|
|
|
|
|
|
@login_required
|
|
def portal_sales(request):
|
|
return render(request, "shop/portal/sales.html", sales_dashboard())
|
|
|
|
|
|
@login_required
|
|
def portal_order_list(request):
|
|
orders = Order.objects.all()[:200]
|
|
return render(request, "shop/portal/orders.html", {"orders": orders})
|
|
|
|
|
|
@login_required
|
|
def portal_order_detail(request, pk):
|
|
order = get_object_or_404(
|
|
Order.objects.prefetch_related("items", "shipments"), pk=pk
|
|
)
|
|
return render(request, "shop/portal/order_detail.html", {"order": order})
|
|
|
|
|
|
@login_required(login_url="account:login")
|
|
def account_order_list(request):
|
|
from accounts.services import claim_orders_for_user
|
|
|
|
claim_orders_for_user(request.user)
|
|
orders = (
|
|
Order.objects.filter(user=request.user)
|
|
.prefetch_related("items", "shipments")
|
|
.exclude(status=Order.Status.DRAFT)
|
|
)
|
|
return render(request, "shop/account/orders.html", {"orders": orders})
|
|
|
|
|
|
@login_required(login_url="account:login")
|
|
def account_order_detail(request, pk):
|
|
order = get_object_or_404(
|
|
Order.objects.prefetch_related("items", "shipments"),
|
|
pk=pk,
|
|
user=request.user,
|
|
)
|
|
from django.apps import apps as django_apps
|
|
|
|
if django_apps.is_installed("shipping"):
|
|
from datetime import timedelta
|
|
|
|
from django.utils import timezone
|
|
from shipping.models import Shipment
|
|
from shipping.services import refresh_tracking
|
|
|
|
stale_after = timezone.now() - timedelta(minutes=15)
|
|
for shipment in order.shipments.all():
|
|
if (
|
|
shipment.status == Shipment.Status.LABELED
|
|
and shipment.tracking_number
|
|
and shipment.tracking_status != Shipment.TrackingStatus.DELIVERED
|
|
and (
|
|
shipment.last_tracked_at is None
|
|
or shipment.last_tracked_at < stale_after
|
|
)
|
|
):
|
|
try:
|
|
refresh_tracking(shipment)
|
|
except Exception:
|
|
logger.exception("order tracking refresh failed for %s", order.number)
|
|
return render(request, "shop/account/order_detail.html", {"order": order})
|
|
|
|
|
|
@csrf_exempt
|
|
@require_http_methods(["POST"])
|
|
def stripe_webhook(request):
|
|
secret = (settings.STRIPE_WEBHOOK_SECRET or "").strip()
|
|
if not secret:
|
|
logger.error("STRIPE_WEBHOOK_SECRET unset")
|
|
return HttpResponseBadRequest("webhook not configured")
|
|
try:
|
|
import stripe
|
|
except ImportError:
|
|
return HttpResponseBadRequest("stripe not installed")
|
|
sig = request.headers.get("Stripe-Signature", "")
|
|
try:
|
|
event = stripe.Webhook.construct_event(request.body, sig, secret)
|
|
except Exception:
|
|
logger.exception("shop stripe webhook signature failed")
|
|
return HttpResponseBadRequest("invalid signature")
|
|
|
|
obj = event.get("data", {}).get("object", {}) or {}
|
|
if event.get("type") != "checkout.session.completed":
|
|
return HttpResponse("ok")
|
|
order_id = (obj.get("metadata") or {}).get("shop_order_id") or ""
|
|
order = None
|
|
if order_id:
|
|
order = Order.objects.filter(pk=order_id).first()
|
|
if order is None:
|
|
session_id = obj.get("id") or ""
|
|
order = Order.objects.filter(stripe_checkout_session_id=session_id).first()
|
|
if order and order.status != Order.Status.PAID:
|
|
mark_paid(
|
|
order,
|
|
stripe_id=obj.get("id") or "",
|
|
stripe_customer_id=obj.get("customer") or "",
|
|
)
|
|
logger.info("shop order %s marked paid", order.number)
|
|
return HttpResponse("ok")
|