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.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 from shop.services import ( ShopError, add_to_cart, adjust_stock, available_qty, cart_lines, cart_total, create_checkout_session, create_order_from_cart, mark_paid, save_cart, set_cart_qty, ) from shop.stats import sales_dashboard logger = logging.getLogger(__name__) def _site_base(request) -> str: base = (settings.PUBLIC_SITE_URL or "").rstrip("/") if base: return base return request.build_absolute_uri("/").rstrip("/") def product_list(request): products = Product.objects.filter(is_published=True) return render(request, "shop/list.html", {"products": products}) def product_detail(request, slug): product = get_object_or_404(Product, slug=slug, is_published=True) return render( request, "shop/detail.html", {"product": product, "available": available_qty(product)}, ) 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, slug=slug, is_published=True) try: qty = int(request.POST.get("quantity") or "1") except ValueError: qty = 1 try: add_to_cart(request.session, product, qty) except ShopError as exc: messages.error(request, str(exc)) return redirect("shop:detail", slug=product.slug) messages.success(request, f"Added {product.name} 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 set_cart_qty(request.session, product, qty) 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 "", ) try: order = create_order_from_cart( request.session, email=email, customer_name=name, shipping_address=address, ) 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) return render( request, "shop/checkout.html", {"lines": lines, "total": cart_total(lines)}, ) 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.all() 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, 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 if errors: for err in errors: messages.error(request, err) else: if product is None: product = Product() 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" product.save() 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"), pk=pk) return render(request, "shop/portal/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 "") logger.info("shop order %s marked paid", order.number) return HttpResponse("ok")