1 Commits
Author SHA1 Message Date
westfarn 9cdce7a897 Add shop, POS sync, event ticketing, and shipping catalog apps (#6)
## Summary
- Closes #5
- Optional `shop`, `pos_sync`, `events`, and `shipping` apps gated by `FEATURE_*` flags
- Deps match the catalog: shop/events need email + Stripe; POS/shipping need shop
- Portal inventory, POS webhooks, capacity tickets, EasyPost/Pirate Ship shipping

## Test plan
- [ ] `manage.py test` (152 passed locally)
- [ ] Shop cart + paid order decrements stocked inventory
- [ ] POS inbound webhook decrements SKU; paid order queues outbound reserve
- [ ] Event capacity blocks overbook; paid order emails ticket codes
- [ ] Shipping stub label + Pirate Ship CSV of unshipped paid orders
- [ ] `validate-env.sh` rejects shop without email/payments, POS/shipping without shop

Reviewed-on: #6
2026-09-06 14:00:22 -07:00
82 changed files with 3908 additions and 1 deletions
+19
View File
@@ -35,6 +35,10 @@ FEATURE_BLOG=true
FEATURE_PAYMENTS=true
FEATURE_SOCIAL=true
FEATURE_SOCIAL_AI=true
FEATURE_SHOP=true
FEATURE_POS_SYNC=true
FEATURE_EVENTS=true
FEATURE_SHIPPING=true
# reCAPTCHA (optional locally — form skips captcha when empty)
RECAPTCHA_PUBLIC_KEY=
@@ -74,6 +78,21 @@ STRIPE_PUBLISHABLE_KEY=
STRIPE_WEBHOOK_SECRET=
STRIPE_CURRENCY=usd
# Shop — FEATURE_SHOP (requires email/SMS + payments)
# SHOP_PRINT_QUEUE_LIMIT_MINUTES=0
# POS sync — FEATURE_POS_SYNC (requires shop). Required in beta/prod.
POS_WEBHOOK_SECRET=
POS_API_BASE_URL=
POS_API_TOKEN=
# Shipping — FEATURE_SHIPPING (requires shop). EasyPost optional; CSV export always works.
EASYPOST_API_KEY=
SHIP_FROM_LINE1=
SHIP_FROM_CITY=
SHIP_FROM_STATE=
SHIP_FROM_ZIP=
# Social — SOCIAL_TOKEN_ENCRYPTION_KEY required when FEATURE_SOCIAL=true
# LinkedIn / Instagram App ID + Secret: Portal → Social accounts (not env).
META_APP_ID=
+19
View File
@@ -32,6 +32,10 @@ FEATURE_BLOG=false
FEATURE_PAYMENTS=false
FEATURE_SOCIAL=false
FEATURE_SOCIAL_AI=false
FEATURE_SHOP=false
FEATURE_POS_SYNC=false
FEATURE_EVENTS=false
FEATURE_SHIPPING=false
# Shared external Postgres (10.0.0.230)
DATABASE_URL=postgres://westfarn:replace-db-password@10.0.0.230:5432/client_site
@@ -63,6 +67,21 @@ STRIPE_PUBLISHABLE_KEY=
STRIPE_WEBHOOK_SECRET=
STRIPE_CURRENCY=usd
# Required when FEATURE_SHOP=true (also requires FEATURE_EMAIL_SMS + FEATURE_PAYMENTS)
# SHOP_PRINT_QUEUE_LIMIT_MINUTES=0
# Required when FEATURE_POS_SYNC=true in beta/prod (also requires FEATURE_SHOP)
POS_WEBHOOK_SECRET=
POS_API_BASE_URL=
POS_API_TOKEN=
# Optional when FEATURE_SHIPPING=true (also requires FEATURE_SHOP)
EASYPOST_API_KEY=
SHIP_FROM_LINE1=
SHIP_FROM_CITY=
SHIP_FROM_STATE=
SHIP_FROM_ZIP=
# Required when FEATURE_SOCIAL=true
META_APP_ID=
META_APP_SECRET=
+4
View File
@@ -23,6 +23,10 @@ separate Django apps. A flag that is off keeps the app out of
| Payments (Stripe) | `FEATURE_PAYMENTS` | `payments` (requires email/SMS) |
| Social consolidation | `FEATURE_SOCIAL` | `social` |
| AI social generator | `FEATURE_SOCIAL_AI` | `social_ai` (requires social) |
| E-commerce storefront | `FEATURE_SHOP` | `shop` (requires email/SMS + payments) |
| POS API sync | `FEATURE_POS_SYNC` | `pos_sync` (requires shop) |
| Event ticketing | `FEATURE_EVENTS` | `events` (requires email/SMS + payments) |
| Shipping automation | `FEATURE_SHIPPING` | `shipping` (requires shop) |
## New client
+4
View File
@@ -40,6 +40,10 @@ services:
FEATURE_PAYMENTS: ${FEATURE_PAYMENTS:-true}
FEATURE_SOCIAL: ${FEATURE_SOCIAL:-true}
FEATURE_SOCIAL_AI: ${FEATURE_SOCIAL_AI:-true}
FEATURE_SHOP: ${FEATURE_SHOP:-true}
FEATURE_POS_SYNC: ${FEATURE_POS_SYNC:-true}
FEATURE_EVENTS: ${FEATURE_EVENTS:-true}
FEATURE_SHIPPING: ${FEATURE_SHIPPING:-true}
PUBLIC_SITE_URL: ${PUBLIC_SITE_URL:-http://127.0.0.1:8000}
EMAIL_HOST: ${EMAIL_HOST:-mail.smtp2go.com}
EMAIL_HOST_USER: ${EMAIL_HOST_USER:-}
+35
View File
@@ -82,6 +82,41 @@ if _true "${FEATURE_SOCIAL_AI:-}"; then
exit 1
fi
fi
if _true "${FEATURE_SHOP:-}"; then
if ! _true "${FEATURE_EMAIL_SMS:-}"; then
echo "FEATURE_SHOP requires FEATURE_EMAIL_SMS" >&2
exit 1
fi
if ! _true "${FEATURE_PAYMENTS:-}"; then
echo "FEATURE_SHOP requires FEATURE_PAYMENTS" >&2
exit 1
fi
fi
if _true "${FEATURE_POS_SYNC:-}"; then
if ! _true "${FEATURE_SHOP:-}"; then
echo "FEATURE_POS_SYNC requires FEATURE_SHOP" >&2
exit 1
fi
if [[ "$DJANGO_ENV" == "prod" || "$DJANGO_ENV" == "beta" ]]; then
required_vars+=(POS_WEBHOOK_SECRET)
fi
fi
if _true "${FEATURE_EVENTS:-}"; then
if ! _true "${FEATURE_EMAIL_SMS:-}"; then
echo "FEATURE_EVENTS requires FEATURE_EMAIL_SMS" >&2
exit 1
fi
if ! _true "${FEATURE_PAYMENTS:-}"; then
echo "FEATURE_EVENTS requires FEATURE_PAYMENTS" >&2
exit 1
fi
fi
if _true "${FEATURE_SHIPPING:-}"; then
if ! _true "${FEATURE_SHOP:-}"; then
echo "FEATURE_SHIPPING requires FEATURE_SHOP" >&2
exit 1
fi
fi
missing=()
for var in "${required_vars[@]}"; do
+34
View File
@@ -115,11 +115,27 @@ FEATURE_BLOG = env_bool("FEATURE_BLOG", _feature_default)
FEATURE_PAYMENTS = env_bool("FEATURE_PAYMENTS", _feature_default)
FEATURE_SOCIAL = env_bool("FEATURE_SOCIAL", _feature_default)
FEATURE_SOCIAL_AI = env_bool("FEATURE_SOCIAL_AI", _feature_default)
FEATURE_SHOP = env_bool("FEATURE_SHOP", _feature_default)
FEATURE_POS_SYNC = env_bool("FEATURE_POS_SYNC", _feature_default)
FEATURE_EVENTS = env_bool("FEATURE_EVENTS", _feature_default)
FEATURE_SHIPPING = env_bool("FEATURE_SHIPPING", _feature_default)
if FEATURE_PAYMENTS and not FEATURE_EMAIL_SMS:
raise ImproperlyConfigured("FEATURE_PAYMENTS requires FEATURE_EMAIL_SMS")
if FEATURE_SOCIAL_AI and not FEATURE_SOCIAL:
raise ImproperlyConfigured("FEATURE_SOCIAL_AI requires FEATURE_SOCIAL")
if FEATURE_SHOP and not FEATURE_EMAIL_SMS:
raise ImproperlyConfigured("FEATURE_SHOP requires FEATURE_EMAIL_SMS")
if FEATURE_SHOP and not FEATURE_PAYMENTS:
raise ImproperlyConfigured("FEATURE_SHOP requires FEATURE_PAYMENTS")
if FEATURE_POS_SYNC and not FEATURE_SHOP:
raise ImproperlyConfigured("FEATURE_POS_SYNC requires FEATURE_SHOP")
if FEATURE_EVENTS and not FEATURE_EMAIL_SMS:
raise ImproperlyConfigured("FEATURE_EVENTS requires FEATURE_EMAIL_SMS")
if FEATURE_EVENTS and not FEATURE_PAYMENTS:
raise ImproperlyConfigured("FEATURE_EVENTS requires FEATURE_PAYMENTS")
if FEATURE_SHIPPING and not FEATURE_SHOP:
raise ImproperlyConfigured("FEATURE_SHIPPING requires FEATURE_SHOP")
CORE_APPS = [
"core.apps.CoreConfig",
@@ -137,6 +153,10 @@ OPTIONAL_APPS = [
(FEATURE_PAYMENTS, "payments.apps.PaymentsConfig"),
(FEATURE_SOCIAL, "social.apps.SocialConfig"),
(FEATURE_SOCIAL_AI, "social_ai.apps.SocialAiConfig"),
(FEATURE_SHOP, "shop.apps.ShopConfig"),
(FEATURE_POS_SYNC, "pos_sync.apps.PosSyncConfig"),
(FEATURE_EVENTS, "events.apps.EventsConfig"),
(FEATURE_SHIPPING, "shipping.apps.ShippingConfig"),
]
DJANGO_APPS = [
"django.contrib.admin",
@@ -352,6 +372,20 @@ STRIPE_PUBLISHABLE_KEY = env("STRIPE_PUBLISHABLE_KEY", "")
STRIPE_WEBHOOK_SECRET = env("STRIPE_WEBHOOK_SECRET", "")
STRIPE_CURRENCY = env("STRIPE_CURRENCY", "usd")
# --- Shop / POS / shipping (FEATURE_SHOP and add-ons) ---
# 0 = unlimited made-to-order print queue.
SHOP_PRINT_QUEUE_LIMIT_MINUTES = int(
env("SHOP_PRINT_QUEUE_LIMIT_MINUTES", "0") or "0"
)
POS_WEBHOOK_SECRET = env("POS_WEBHOOK_SECRET", "")
POS_API_BASE_URL = env("POS_API_BASE_URL", "")
POS_API_TOKEN = env("POS_API_TOKEN", "")
EASYPOST_API_KEY = env("EASYPOST_API_KEY", "")
SHIP_FROM_LINE1 = env("SHIP_FROM_LINE1", "")
SHIP_FROM_CITY = env("SHIP_FROM_CITY", "")
SHIP_FROM_STATE = env("SHIP_FROM_STATE", "")
SHIP_FROM_ZIP = env("SHIP_FROM_ZIP", "")
# --- Nominatim (address suggest; server-side proxy only) ---
# Self-hosted on ai-server-4080. Nominatim has no native API keys — gate with
# LAN UFW + this Django proxy. Optional NOMINATIM_API_KEY is forwarded as
+57
View File
@@ -264,6 +264,63 @@ body.portal {
line-height: 1.2;
white-space: nowrap;
}
.chart-placeholder.chart-daily {
gap: 3px;
padding: 16px 10px 8px;
overflow-x: auto;
}
.chart-placeholder.chart-daily .chart-bar-col {
flex: 1 0 8px;
min-width: 6px;
}
.chart-placeholder.chart-daily .chart-bar-col .bar {
min-height: 0;
}
.chart-placeholder.chart-daily .chart-bar-col .bar.is-zero {
height: 0 !important;
min-height: 0;
opacity: 0.2;
}
.chart-placeholder.chart-daily .chart-bar-label {
font-size: 10px;
}
.chart-placeholder.chart-revenue .chart-bar-col .bar {
background: var(--monica-primary-light);
}
.hbar-chart {
display: flex;
flex-direction: column;
gap: 12px;
}
.hbar-row {
display: grid;
grid-template-columns: minmax(72px, 160px) 1fr auto;
gap: 12px;
align-items: center;
}
.hbar-label {
font-size: 13px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.hbar-track {
height: 16px;
background: #e8f4f5;
overflow: hidden;
}
.hbar-fill {
height: 100%;
background: var(--monica-primary);
opacity: 0.8;
min-width: 0;
}
.hbar-value {
font-size: 13px;
font-weight: 600;
min-width: 2.5ch;
text-align: right;
}
.consent-pills { display: flex; gap: 6px; flex-wrap: wrap; }
.preview-pane {
+14
View File
@@ -48,5 +48,19 @@ if apps.is_installed("social_ai"):
urlpatterns += [
path("portal/social/api/generate/", include("social_ai.urls")),
]
if apps.is_installed("shop"):
urlpatterns += [
path("shop/", include("shop.public_urls")),
path("portal/shop/", include("shop.portal_urls")),
]
if apps.is_installed("pos_sync"):
urlpatterns += [path("portal/pos/", include("pos_sync.urls"))]
if apps.is_installed("events"):
urlpatterns += [
path("events/", include("events.public_urls")),
path("portal/events/", include("events.portal_urls")),
]
if apps.is_installed("shipping"):
urlpatterns += [path("portal/shipping/", include("shipping.urls"))]
handler404 = "public.views.page_not_found"
+1
View File
@@ -22,6 +22,7 @@ GROUP_ORDER = {
"Outreach": 20,
"Content": 30,
"Social": 40,
"Retail": 45,
"Billing": 50,
}
+37 -1
View File
@@ -22,16 +22,30 @@ class FeatureFlagSettingsTests(SimpleTestCase):
self.assertIn("FEATURE_PAYMENTS", src)
self.assertIn("FEATURE_SOCIAL", src)
self.assertIn("FEATURE_SOCIAL_AI", src)
self.assertIn("FEATURE_SHOP", src)
self.assertIn("FEATURE_POS_SYNC", src)
self.assertIn("FEATURE_EVENTS", src)
self.assertIn("FEATURE_SHIPPING", src)
self.assertIn("email_sms.apps.EmailSmsConfig", src)
self.assertIn("directmail.apps.DirectmailConfig", src)
self.assertIn("blog.apps.BlogConfig", src)
self.assertIn("payments.apps.PaymentsConfig", src)
self.assertIn("social_ai.apps.SocialAiConfig", src)
self.assertIn("shop.apps.ShopConfig", src)
self.assertIn("pos_sync.apps.PosSyncConfig", src)
self.assertIn("events.apps.EventsConfig", src)
self.assertIn("shipping.apps.ShippingConfig", src)
def test_payments_requires_email_sms_in_source(self):
src = _read_settings_source()
self.assertIn("FEATURE_PAYMENTS requires FEATURE_EMAIL_SMS", src)
self.assertIn("FEATURE_SOCIAL_AI requires FEATURE_SOCIAL", src)
self.assertIn("FEATURE_SHOP requires FEATURE_EMAIL_SMS", src)
self.assertIn("FEATURE_SHOP requires FEATURE_PAYMENTS", src)
self.assertIn("FEATURE_POS_SYNC requires FEATURE_SHOP", src)
self.assertIn("FEATURE_EVENTS requires FEATURE_EMAIL_SMS", src)
self.assertIn("FEATURE_EVENTS requires FEATURE_PAYMENTS", src)
self.assertIn("FEATURE_SHIPPING requires FEATURE_SHOP", src)
class AlwaysOnImportIsolationTests(SimpleTestCase):
@@ -44,6 +58,10 @@ class AlwaysOnImportIsolationTests(SimpleTestCase):
"payments",
"social",
"social_ai",
"shop",
"pos_sync",
"events",
"shipping",
}
ALWAYS_ON = [
@@ -82,7 +100,18 @@ class InstalledOptionalAppsTests(TestCase):
def test_dev_installs_catalog_apps(self):
labels = {c.label for c in apps.get_app_configs()}
for label in ("email_sms", "directmail", "blog", "payments", "social", "social_ai"):
for label in (
"email_sms",
"directmail",
"blog",
"payments",
"social",
"social_ai",
"shop",
"pos_sync",
"events",
"shipping",
):
self.assertIn(label, labels)
def test_optional_urls_resolve_when_installed(self):
@@ -92,3 +121,10 @@ class InstalledOptionalAppsTests(TestCase):
self.assertTrue(reverse("payments:invoice_list").startswith("/portal/payments/"))
self.assertTrue(reverse("social:composer").startswith("/portal/social/"))
self.assertTrue(reverse("social_ai:generate").startswith("/portal/social/api/generate/"))
self.assertTrue(reverse("shop:list").startswith("/shop/"))
self.assertTrue(reverse("shop_portal:product_list").startswith("/portal/shop/"))
self.assertTrue(reverse("shop_portal:sales").startswith("/portal/shop/sales/"))
self.assertTrue(reverse("pos_sync:event_list").startswith("/portal/pos/"))
self.assertTrue(reverse("events:list").startswith("/events/"))
self.assertTrue(reverse("events_portal:event_list").startswith("/portal/events/"))
self.assertTrue(reverse("shipping:shipment_list").startswith("/portal/shipping/"))
@@ -27,6 +27,36 @@
<div class="value">{{ open_invoices|default:0 }}</div>
</div>
{% endif %}
{% if 'shop' in enabled_features %}
<div class="stat-card">
<div class="label">Open shop orders</div>
<div class="value">{{ open_shop_orders|default:0 }}</div>
</div>
<div class="stat-card">
<div class="label">Sales (30 days)</div>
<div class="value">{{ shop_sales_30d|default:0 }}</div>
</div>
<div class="stat-card">
<div class="label">Shop revenue (30 days)</div>
<div class="value" style="font-size:22px">{{ shop_revenue_30d|default:0 }}</div>
</div>
<div class="stat-card">
<div class="label">Low stock</div>
<div class="value">{{ low_stock_products|default:0 }}</div>
</div>
{% endif %}
{% if 'events' in enabled_features %}
<div class="stat-card">
<div class="label">Upcoming events</div>
<div class="value">{{ upcoming_events|default:0 }}</div>
</div>
{% endif %}
{% if 'shipping' in enabled_features %}
<div class="stat-card">
<div class="label">Unshipped orders</div>
<div class="value">{{ unshipped_orders|default:0 }}</div>
</div>
{% endif %}
</div>
<div class="split">
View File
+24
View File
@@ -0,0 +1,24 @@
from django.contrib import admin
from events.models import Event, Ticket, TicketOrder
@admin.register(Event)
class EventAdmin(admin.ModelAdmin):
list_display = ("title", "starts_at", "capacity", "price", "is_published")
list_filter = ("is_published",)
search_fields = ("title", "venue")
prepopulated_fields = {"slug": ("title",)}
@admin.register(TicketOrder)
class TicketOrderAdmin(admin.ModelAdmin):
list_display = ("number", "event", "email", "quantity", "status")
list_filter = ("status",)
search_fields = ("number", "email")
@admin.register(Ticket)
class TicketAdmin(admin.ModelAdmin):
list_display = ("code", "event", "order")
search_fields = ("code",)
+12
View File
@@ -0,0 +1,12 @@
from django.apps import AppConfig
class EventsConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "events"
verbose_name = "Event ticketing"
def ready(self):
from events import hooks
hooks.register()
+33
View File
@@ -0,0 +1,33 @@
from core.registry import (
register_dashboard_collector,
register_feature,
register_portal_nav,
register_public_nav,
)
def register() -> None:
register_feature("events")
register_public_nav(
section="events", label="Events", url_name="events:list", order=35
)
register_portal_nav(
section="events",
label="Events",
url_name="events_portal:event_list",
group="Retail",
order=30,
)
register_dashboard_collector(_dashboard)
def _dashboard(request) -> dict:
from django.utils import timezone
from events.models import Event
return {
"upcoming_events": Event.objects.filter(
is_published=True, starts_at__gte=timezone.now()
).count()
}
+74
View File
@@ -0,0 +1,74 @@
# Generated by Django 6.1 on 2026-09-06 11:17
import django.db.models.deletion
import uuid
from decimal import Decimal
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Event',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('title', models.CharField(max_length=200)),
('slug', models.SlugField(max_length=220, unique=True)),
('description', models.TextField(blank=True)),
('starts_at', models.DateTimeField()),
('venue', models.CharField(blank=True, max_length=200)),
('capacity', models.PositiveIntegerField(default=0)),
('price', models.DecimalField(decimal_places=2, default=Decimal('0'), max_digits=10)),
('currency', models.CharField(default='usd', max_length=8)),
('is_published', models.BooleanField(default=False)),
],
options={
'ordering': ['starts_at'],
},
),
migrations.CreateModel(
name='TicketOrder',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('number', models.CharField(max_length=32, unique=True)),
('email', models.EmailField(max_length=254)),
('customer_name', models.CharField(blank=True, max_length=200)),
('quantity', models.PositiveIntegerField(default=1)),
('amount', models.DecimalField(decimal_places=2, max_digits=10)),
('currency', models.CharField(default='usd', max_length=8)),
('status', models.CharField(choices=[('draft', 'Draft'), ('open', 'Open'), ('paid', 'Paid'), ('cancelled', 'Cancelled')], default='draft', max_length=16)),
('stripe_checkout_session_id', models.CharField(blank=True, max_length=255)),
('hosted_checkout_url', models.URLField(blank=True)),
('paid_at', models.DateTimeField(blank=True, null=True)),
('event', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='ticket_orders', to='events.event')),
],
options={
'ordering': ['-created_at'],
},
),
migrations.CreateModel(
name='Ticket',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('code', models.CharField(max_length=16, unique=True)),
('attendee_name', models.CharField(blank=True, max_length=200)),
('event', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='tickets', to='events.event')),
('order', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='tickets', to='events.ticketorder')),
],
options={
'ordering': ['created_at'],
},
),
]
View File
+111
View File
@@ -0,0 +1,111 @@
from decimal import Decimal
from django.db import models
from django.urls import reverse
from django.utils import timezone
from django.utils.text import slugify
from core.models import TimeStampedModel, UUIDPrimaryKeyModel
class Event(UUIDPrimaryKeyModel, TimeStampedModel):
title = models.CharField(max_length=200)
slug = models.SlugField(max_length=220, unique=True)
description = models.TextField(blank=True)
starts_at = models.DateTimeField()
venue = models.CharField(max_length=200, blank=True)
capacity = models.PositiveIntegerField(default=0)
price = models.DecimalField(max_digits=10, decimal_places=2, default=Decimal("0"))
currency = models.CharField(max_length=8, default="usd")
is_published = models.BooleanField(default=False)
class Meta:
ordering = ["starts_at"]
def __str__(self) -> str:
return self.title
def get_absolute_url(self) -> str:
return reverse("events:detail", kwargs={"slug": self.slug})
def save(self, *args, **kwargs):
if not self.slug:
base = slugify(self.title)[:200] or "event"
slug = base
n = 2
while Event.objects.filter(slug=slug).exclude(pk=self.pk).exists():
slug = f"{base}-{n}"
n += 1
self.slug = slug
super().save(*args, **kwargs)
def tickets_held(self, *, exclude_order_id=None) -> int:
qs = self.ticket_orders.filter(
status__in=[
TicketOrder.Status.DRAFT,
TicketOrder.Status.OPEN,
TicketOrder.Status.PAID,
]
)
if exclude_order_id:
qs = qs.exclude(pk=exclude_order_id)
return qs.aggregate(total=models.Sum("quantity"))["total"] or 0
@property
def tickets_sold(self) -> int:
return self.tickets_held()
@property
def seats_remaining(self) -> int | None:
if not self.capacity:
return None
return max(self.capacity - self.tickets_sold, 0)
@property
def is_upcoming(self) -> bool:
return self.starts_at >= timezone.now()
class TicketOrder(UUIDPrimaryKeyModel, TimeStampedModel):
class Status(models.TextChoices):
DRAFT = "draft", "Draft"
OPEN = "open", "Open"
PAID = "paid", "Paid"
CANCELLED = "cancelled", "Cancelled"
number = models.CharField(max_length=32, unique=True)
event = models.ForeignKey(
Event, on_delete=models.PROTECT, related_name="ticket_orders"
)
email = models.EmailField()
customer_name = models.CharField(max_length=200, blank=True)
quantity = models.PositiveIntegerField(default=1)
amount = models.DecimalField(max_digits=10, decimal_places=2)
currency = models.CharField(max_length=8, default="usd")
status = models.CharField(
max_length=16, choices=Status.choices, default=Status.DRAFT
)
stripe_checkout_session_id = models.CharField(max_length=255, blank=True)
hosted_checkout_url = models.URLField(blank=True)
paid_at = models.DateTimeField(null=True, blank=True)
class Meta:
ordering = ["-created_at"]
def __str__(self) -> str:
return f"{self.number} · {self.event}"
class Ticket(UUIDPrimaryKeyModel, TimeStampedModel):
order = models.ForeignKey(
TicketOrder, on_delete=models.CASCADE, related_name="tickets"
)
event = models.ForeignKey(Event, on_delete=models.PROTECT, related_name="tickets")
code = models.CharField(max_length=16, unique=True)
attendee_name = models.CharField(max_length=200, blank=True)
class Meta:
ordering = ["created_at"]
def __str__(self) -> str:
return self.code
+13
View File
@@ -0,0 +1,13 @@
from django.urls import path
from events import views
app_name = "events_portal"
urlpatterns = [
path("", views.portal_event_list, name="event_list"),
path("new/", views.portal_event_edit, name="event_new"),
path("<uuid:pk>/", views.portal_event_detail, name="event_detail"),
path("<uuid:pk>/edit/", views.portal_event_edit, name="event_edit"),
path("webhooks/stripe/", views.stripe_webhook, name="stripe_webhook"),
]
+13
View File
@@ -0,0 +1,13 @@
from django.urls import path
from events import views
app_name = "events"
urlpatterns = [
path("", views.event_list, name="list"),
path("order/<uuid:pk>/success/", views.checkout_success, name="checkout_success"),
path("order/<uuid:pk>/cancel/", views.checkout_cancel, name="checkout_cancel"),
path("<slug:slug>/", views.event_detail, name="detail"),
path("<slug:slug>/buy/", views.buy_tickets, name="buy"),
]
+188
View File
@@ -0,0 +1,188 @@
"""Event capacity, Stripe ticket checkout, and confirmation email."""
from __future__ import annotations
import logging
import secrets
from datetime import date
from decimal import Decimal
from django.conf import settings
from django.db import transaction
from django.utils import timezone
from events.models import Event, Ticket, TicketOrder
logger = logging.getLogger(__name__)
class EventsError(RuntimeError):
pass
def _stripe():
secret = (settings.STRIPE_SECRET_KEY or "").strip()
if not secret:
raise EventsError("STRIPE_SECRET_KEY is not configured")
try:
import stripe
except ImportError as exc:
raise EventsError("stripe package is not installed") from exc
stripe.api_key = secret
return stripe
def next_ticket_order_number() -> str:
today = date.today().strftime("%Y%m%d")
prefix = f"TIX-{today}-"
existing = TicketOrder.objects.filter(number__startswith=prefix).count()
return f"{prefix}{existing + 1:03d}"
def new_ticket_code() -> str:
while True:
code = secrets.token_hex(4).upper()
if not Ticket.objects.filter(code=code).exists():
return code
def assert_capacity(
event: Event, quantity: int, *, exclude_order_id=None
) -> None:
if quantity < 1:
raise EventsError("Quantity must be at least 1.")
if not event.capacity:
return
held = event.tickets_held(exclude_order_id=exclude_order_id)
remaining = max(event.capacity - held, 0)
if quantity > remaining:
raise EventsError(f"Only {remaining} seats left for {event.title}.")
def create_ticket_order(
event: Event, *, email: str, customer_name: str = "", quantity: int = 1
) -> TicketOrder:
email = (email or "").strip()
if not email:
raise EventsError("Email is required.")
if not event.is_published:
raise EventsError("This event is not on sale.")
assert_capacity(event, quantity)
amount = event.price * quantity
return TicketOrder.objects.create(
number=next_ticket_order_number(),
event=event,
email=email,
customer_name=(customer_name or "").strip(),
quantity=quantity,
amount=amount,
currency=(event.currency or settings.STRIPE_CURRENCY or "usd").lower(),
status=TicketOrder.Status.DRAFT,
)
def create_checkout_session(
order: TicketOrder, *, success_url: str, cancel_url: str
) -> str:
stripe = _stripe()
session = stripe.checkout.Session.create(
mode="payment",
customer_email=order.email or None,
line_items=[
{
"quantity": order.quantity,
"price_data": {
"currency": (order.currency or "usd").lower(),
"unit_amount": int(
(order.event.price * Decimal("100")).quantize(Decimal("1"))
),
"product_data": {"name": order.event.title},
},
}
],
metadata={
"ticket_order_id": str(order.pk),
"ticket_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 = TicketOrder.Status.OPEN
order.save(
update_fields=[
"stripe_checkout_session_id",
"hosted_checkout_url",
"status",
"updated_at",
]
)
return session.url or ""
def _issue_tickets(order: TicketOrder) -> list[Ticket]:
created = []
for _ in range(order.quantity):
created.append(
Ticket.objects.create(
order=order,
event=order.event,
code=new_ticket_code(),
attendee_name=order.customer_name,
)
)
return created
def mark_paid(order: TicketOrder, *, stripe_id: str = "") -> None:
if order.status == TicketOrder.Status.PAID:
return
with transaction.atomic():
locked = TicketOrder.objects.select_for_update().get(pk=order.pk)
if locked.status == TicketOrder.Status.PAID:
return
assert_capacity(
locked.event, locked.quantity, exclude_order_id=locked.pk
)
_issue_tickets(locked)
locked.status = TicketOrder.Status.PAID
locked.paid_at = timezone.now()
locked.save(update_fields=["status", "paid_at", "updated_at"])
order.refresh_from_db()
try:
send_ticket_email(order)
except Exception:
logger.exception("ticket email failed for %s", order.number)
def send_ticket_email(order: TicketOrder) -> bool:
to_email = (order.email or "").strip()
if not to_email:
raise EventsError("Order has no email address")
from django.core.mail import EmailMultiAlternatives
codes = ", ".join(t.code for t in order.tickets.all())
name = order.customer_name or "there"
when = timezone.localtime(order.event.starts_at).strftime("%b %d, %Y %-I:%M %p")
subject = f"Tickets for {order.event.title}{settings.SITE_NAME}"
text = (
f"Hi {name},\n\n"
f"Your tickets for {order.event.title} on {when}:\n"
f"{codes}\n\nOrder {order.number}\n"
)
html = (
f"<p>Hi {name},</p>"
f"<p>Tickets for <strong>{order.event.title}</strong> on {when}:</p>"
f"<p><strong>{codes}</strong></p>"
f"<p>Order {order.number}</p>"
)
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
+17
View File
@@ -0,0 +1,17 @@
{% extends "base.html" %}
{% block title %}Tickets · {{ event.title }}{% endblock %}
{% block content %}
<section class="section section-lg">
<div class="container">
<h1>Tickets · {{ event.title }}</h1>
<p>{{ event.price }} {{ event.currency|upper }} each{% if event.seats_remaining is not None %} · {{ event.seats_remaining }} left{% endif %}</p>
<form method="post">
{% csrf_token %}
<p><label>Email <input name="email" type="email" required></label></p>
<p><label>Name <input name="customer_name"></label></p>
<p><label>Quantity <input name="quantity" type="number" min="1" value="1"></label></p>
<button class="button button-primary" type="submit">Pay with Stripe</button>
</form>
</div>
</section>
{% endblock %}
+8
View File
@@ -0,0 +1,8 @@
{% extends "base.html" %}
{% block title %}Checkout cancelled{% endblock %}
{% block content %}
<section class="section section-lg"><div class="container">
<h1>Checkout cancelled</h1>
<p>No charge for {{ order.event.title }}. <a href="{% url 'events:buy' order.event.slug %}">Try again</a>.</p>
</div></section>
{% endblock %}
+18
View File
@@ -0,0 +1,18 @@
{% extends "base.html" %}
{% block title %}{{ event.title }} · {{ SITE_NAME }}{% endblock %}
{% block content %}
<section class="section section-lg">
<div class="container">
<p><a href="{% url 'events:list' %}">← Events</a></p>
<h1>{{ event.title }}</h1>
<p class="muted">{{ event.starts_at|date:"F j, Y g:i A" }}{% if event.venue %} · {{ event.venue }}{% endif %}</p>
<div>{{ event.description|linebreaks }}</div>
<p>{{ event.price }} {{ event.currency|upper }}</p>
{% if event.seats_remaining == 0 %}
<p>Sold out.</p>
{% else %}
<p><a class="button button-primary" href="{% url 'events:buy' event.slug %}">Buy tickets</a></p>
{% endif %}
</div>
</section>
{% endblock %}
+18
View File
@@ -0,0 +1,18 @@
{% extends "base.html" %}
{% block title %}Events · {{ SITE_NAME }}{% endblock %}
{% block content %}
<section class="section section-lg">
<div class="container">
<h1 class="text-uppercase">Events</h1>
{% for event in events %}
<article style="margin:0 0 32px">
<h2><a href="{{ event.get_absolute_url }}">{{ event.title }}</a></h2>
<p class="muted">{{ event.starts_at|date:"F j, Y g:i A" }}{% if event.venue %} · {{ event.venue }}{% endif %}</p>
<p>{{ event.price }} {{ event.currency|upper }}{% if event.seats_remaining is not None %} · {{ event.seats_remaining }} seats left{% endif %}</p>
</article>
{% empty %}
<p>No upcoming events.</p>
{% endfor %}
</div>
</section>
{% endblock %}
@@ -0,0 +1,23 @@
{% extends "portal_base.html" %}
{% block title %}{{ event.title }} · Portal{% endblock %}
{% block topbar_title %}{{ event.title }}{% endblock %}
{% block portal_content %}
<p><a class="btn btn-ghost" href="{% url 'events_portal:event_edit' event.pk %}">Edit</a></p>
<p>{{ event.starts_at }} · {{ event.tickets_sold }} / {{ event.capacity|default:"∞" }} seats · {{ event.price }} {{ event.currency|upper }}</p>
<table class="table">
<thead><tr><th>Order</th><th>Email</th><th>Qty</th><th>Status</th><th>Codes</th></tr></thead>
<tbody>
{% for order in orders %}
<tr>
<td>{{ order.number }}</td>
<td>{{ order.email }}</td>
<td>{{ order.quantity }}</td>
<td>{{ order.get_status_display }}</td>
<td>{% for ticket in order.tickets.all %}{{ ticket.code }}{% if not forloop.last %}, {% endif %}{% endfor %}</td>
</tr>
{% empty %}
<tr><td colspan="5" class="empty-state">No ticket orders yet.</td></tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
@@ -0,0 +1,17 @@
{% extends "portal_base.html" %}
{% block title %}{% if event %}Edit{% else %}New{% endif %} event · Portal{% endblock %}
{% block topbar_title %}{% if event %}Edit event{% else %}New event{% endif %}{% endblock %}
{% block portal_content %}
<form method="post" class="form-grid">
{% csrf_token %}
<div class="field"><label>Title</label><input name="title" required value="{{ event.title|default:'' }}"></div>
<div class="field"><label>Slug</label><input name="slug" value="{{ event.slug|default:'' }}" placeholder="auto from title"></div>
<div class="field"><label>Starts</label><input name="starts_at" type="datetime-local" required value="{% if event %}{{ event.starts_at|date:'Y-m-d\\TH:i' }}{% endif %}"></div>
<div class="field"><label>Venue</label><input name="venue" value="{{ event.venue|default:'' }}"></div>
<div class="field"><label>Capacity</label><input name="capacity" type="number" min="0" value="{{ event.capacity|default:0 }}"></div>
<div class="field"><label>Price</label><input name="price" type="number" step="0.01" min="0" required value="{{ event.price|default:'0' }}"></div>
<div class="field"><label>Description</label><textarea name="description" style="min-height:120px">{{ event.description|default:'' }}</textarea></div>
<label><input type="checkbox" name="is_published" {% if event.is_published %}checked{% endif %}> Published</label>
<button class="btn btn-primary" type="submit">Save</button>
</form>
{% endblock %}
@@ -0,0 +1,21 @@
{% extends "portal_base.html" %}
{% block title %}Events · Portal{% endblock %}
{% block topbar_title %}Events{% endblock %}
{% block portal_content %}
<p><a class="btn btn-primary" href="{% url 'events_portal:event_new' %}">New event</a></p>
<table class="table">
<thead><tr><th>Title</th><th>When</th><th>Capacity</th><th>Status</th></tr></thead>
<tbody>
{% for event in events %}
<tr>
<td><a href="{% url 'events_portal:event_detail' event.pk %}">{{ event.title }}</a></td>
<td>{{ event.starts_at }}</td>
<td>{{ event.tickets_sold }} / {{ event.capacity|default:"∞" }}</td>
<td>{% if event.is_published %}Published{% else %}Draft{% endif %}</td>
</tr>
{% empty %}
<tr><td colspan="4" class="empty-state">No events yet.</td></tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
@@ -0,0 +1,9 @@
{% extends "base.html" %}
{% block title %}Tickets {{ order.number }}{% endblock %}
{% block content %}
<section class="section section-lg"><div class="container">
<h1>You're in</h1>
<p>Order {{ order.number }} for {{ order.event.title }} is {{ order.get_status_display|lower }}.</p>
<p>Ticket codes will be emailed to {{ order.email }}.</p>
</div></section>
{% endblock %}
+95
View File
@@ -0,0 +1,95 @@
from decimal import Decimal
from django.contrib.auth import get_user_model
from django.core import mail
from django.test import Client, TestCase
from django.urls import reverse
from django.utils import timezone
from events.models import Event, Ticket, TicketOrder
from events.services import EventsError, create_ticket_order, mark_paid
def _event(**kwargs):
defaults = dict(
title="Friday Night Prerelease",
starts_at=timezone.now() + timezone.timedelta(days=7),
venue="Main shop",
capacity=8,
price=Decimal("25.00"),
is_published=True,
)
defaults.update(kwargs)
return Event.objects.create(**defaults)
class EventPublicTests(TestCase):
def test_list_hides_drafts_and_past(self):
_event(title="Live prerelease")
_event(title="Draft night", is_published=False)
_event(
title="Last week",
starts_at=timezone.now() - timezone.timedelta(days=2),
)
response = Client().get(reverse("events:list"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Live prerelease")
self.assertNotContains(response, "Draft night")
self.assertNotContains(response, "Last week")
def test_buy_form_renders(self):
event = _event()
response = Client().get(reverse("events:buy", kwargs={"slug": event.slug}))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Pay with Stripe")
class EventCapacityTests(TestCase):
def test_capacity_blocks_overbook(self):
event = _event(capacity=2)
create_ticket_order(event, email="a@example.com", quantity=2)
with self.assertRaises(EventsError):
create_ticket_order(event, email="b@example.com", quantity=1)
def test_mark_paid_issues_tickets_and_email(self):
event = _event(capacity=4)
order = create_ticket_order(
event, email="fan@example.com", customer_name="Lee", quantity=2
)
mark_paid(order)
order.refresh_from_db()
self.assertEqual(order.status, TicketOrder.Status.PAID)
self.assertEqual(Ticket.objects.filter(order=order).count(), 2)
self.assertEqual(event.tickets_sold, 2)
self.assertEqual(len(mail.outbox), 1)
self.assertIn(event.title, mail.outbox[0].subject)
class EventPortalTests(TestCase):
def setUp(self):
User = get_user_model()
self.user = User.objects.create_user("host", password="test-pass-123")
self.client = Client()
self.client.login(username="host", password="test-pass-123")
def test_list_requires_login(self):
self.assertEqual(Client().get(reverse("events_portal:event_list")).status_code, 302)
def test_create_published_event(self):
starts = (timezone.now() + timezone.timedelta(days=3)).strftime("%Y-%m-%dT%H:%M")
response = self.client.post(
reverse("events_portal:event_new"),
{
"title": "Commander night",
"starts_at": starts,
"venue": "Back room",
"capacity": "16",
"price": "10.00",
"is_published": "on",
},
)
self.assertEqual(response.status_code, 302)
event = Event.objects.get()
self.assertEqual(event.slug, "commander-night")
self.assertTrue(event.is_published)
self.assertEqual(event.capacity, 16)
+194
View File
@@ -0,0 +1,194 @@
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 import timezone
from datetime import datetime
from django.utils.text import slugify
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
from events.models import Event, TicketOrder
from events.services import (
EventsError,
create_checkout_session,
create_ticket_order,
mark_paid,
)
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 event_list(request):
events = Event.objects.filter(is_published=True, starts_at__gte=timezone.now())
return render(request, "events/list.html", {"events": events})
def event_detail(request, slug):
event = get_object_or_404(Event, slug=slug, is_published=True)
return render(request, "events/detail.html", {"event": event})
@require_http_methods(["GET", "POST"])
def buy_tickets(request, slug):
event = get_object_or_404(Event, slug=slug, is_published=True)
if request.method == "POST":
email = (request.POST.get("email") or "").strip()
name = (request.POST.get("customer_name") or "").strip()
try:
quantity = int(request.POST.get("quantity") or "1")
except ValueError:
quantity = 0
try:
order = create_ticket_order(
event, email=email, customer_name=name, quantity=quantity
)
base = _site_base(request)
success = base + reverse(
"events:checkout_success", kwargs={"pk": order.pk}
)
cancel = base + reverse("events:checkout_cancel", kwargs={"pk": order.pk})
url = create_checkout_session(
order,
success_url=success + "?session_id={CHECKOUT_SESSION_ID}",
cancel_url=cancel,
)
except EventsError as exc:
messages.error(request, str(exc))
except Exception as exc: # noqa: BLE001
logger.exception("ticket checkout failed")
messages.error(request, f"Could not start checkout: {exc}")
else:
return redirect(url)
return render(request, "events/buy.html", {"event": event})
def checkout_success(request, pk):
order = get_object_or_404(TicketOrder.objects.select_related("event"), pk=pk)
return render(request, "events/success.html", {"order": order})
def checkout_cancel(request, pk):
order = get_object_or_404(TicketOrder.objects.select_related("event"), pk=pk)
return render(request, "events/cancel.html", {"order": order})
@login_required
def portal_event_list(request):
events = Event.objects.all().order_by("-starts_at")
return render(request, "events/portal/list.html", {"events": events})
@login_required
@require_http_methods(["GET", "POST"])
def portal_event_edit(request, pk=None):
event = get_object_or_404(Event, pk=pk) if pk else None
if request.method == "POST":
title = (request.POST.get("title") or "").strip()
description = (request.POST.get("description") or "").strip()
venue = (request.POST.get("venue") or "").strip()
slug = (request.POST.get("slug") or "").strip()
starts_raw = (request.POST.get("starts_at") or "").strip()
errors = []
if not title:
errors.append("Title is required.")
starts_at = None
if starts_raw:
try:
starts_at = datetime.fromisoformat(starts_raw)
except ValueError:
starts_at = None
if starts_at is None:
errors.append("Start date/time is required.")
elif timezone.is_naive(starts_at):
starts_at = timezone.make_aware(starts_at, timezone.get_current_timezone())
try:
capacity = int(request.POST.get("capacity") or "0")
except ValueError:
capacity = 0
errors.append("Capacity must be a number.")
try:
price = Decimal(request.POST.get("price") or "0")
if price < 0:
raise InvalidOperation
except Exception:
price = None
errors.append("Enter a valid price.")
if errors:
for err in errors:
messages.error(request, err)
else:
if event is None:
event = Event()
event.title = title
event.description = description
event.venue = venue
event.slug = slugify(slug)[:220] if slug else ""
event.starts_at = starts_at
event.capacity = max(capacity, 0)
event.price = price
event.currency = (settings.STRIPE_CURRENCY or "usd").lower()
event.is_published = request.POST.get("is_published") == "on"
event.save()
messages.success(request, f"Saved {event.title}.")
return redirect("events_portal:event_list")
return render(request, "events/portal/edit.html", {"event": event})
@login_required
def portal_event_detail(request, pk):
event = get_object_or_404(Event, pk=pk)
orders = event.ticket_orders.select_related().prefetch_related("tickets")
return render(
request,
"events/portal/detail.html",
{"event": event, "orders": orders},
)
@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("events 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("ticket_order_id") or ""
order = None
if order_id:
order = TicketOrder.objects.filter(pk=order_id).first()
if order is None:
session_id = obj.get("id") or ""
order = TicketOrder.objects.filter(
stripe_checkout_session_id=session_id
).first()
if order and order.status != TicketOrder.Status.PAID:
mark_paid(order, stripe_id=obj.get("id") or "")
logger.info("ticket order %s marked paid", order.number)
return HttpResponse("ok")
View File
+16
View File
@@ -0,0 +1,16 @@
from django.contrib import admin
from pos_sync.models import POSConnection, SyncEvent
@admin.register(POSConnection)
class POSConnectionAdmin(admin.ModelAdmin):
list_display = ("name", "provider", "is_active")
list_filter = ("provider", "is_active")
@admin.register(SyncEvent)
class SyncEventAdmin(admin.ModelAdmin):
list_display = ("direction", "sku", "quantity", "status", "created_at")
list_filter = ("direction", "status")
search_fields = ("sku", "external_id")
+12
View File
@@ -0,0 +1,12 @@
from django.apps import AppConfig
class PosSyncConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "pos_sync"
verbose_name = "POS API sync"
def ready(self):
from pos_sync import hooks
hooks.register()
+35
View File
@@ -0,0 +1,35 @@
from core.registry import (
register_dashboard_collector,
register_dispatcher,
register_feature,
register_portal_nav,
)
def register() -> None:
register_feature("pos_sync")
register_portal_nav(
section="pos_sync",
label="POS sync",
url_name="pos_sync:event_list",
group="Retail",
order=40,
)
register_dashboard_collector(_dashboard)
register_dispatcher(_dispatch)
def _dashboard(request) -> dict:
from pos_sync.models import SyncEvent
return {
"pos_failed_syncs": SyncEvent.objects.filter(
status=SyncEvent.Status.FAILED
).count()
}
def _dispatch() -> int:
from pos_sync.services import dispatch_pending_outbound
return dispatch_pending_outbound()
+52
View File
@@ -0,0 +1,52 @@
# Generated by Django 6.1 on 2026-09-06 11:17
import django.db.models.deletion
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='POSConnection',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('name', models.CharField(max_length=120)),
('provider', models.CharField(choices=[('generic', 'Generic REST'), ('square', 'Square'), ('lightspeed', 'Lightspeed'), ('binderpos', 'BinderPOS'), ('crystalcommerce', 'CrystalCommerce')], default='generic', max_length=32)),
('api_base_url', models.URLField(blank=True)),
('api_token', models.CharField(blank=True, max_length=255)),
('webhook_secret', models.CharField(blank=True, max_length=255)),
('is_active', models.BooleanField(default=True)),
],
options={
'ordering': ['name'],
},
),
migrations.CreateModel(
name='SyncEvent',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('direction', models.CharField(choices=[('inbound', 'POS → site'), ('outbound', 'Site → POS')], max_length=16)),
('status', models.CharField(choices=[('pending', 'Pending'), ('done', 'Done'), ('failed', 'Failed')], default='pending', max_length=16)),
('sku', models.CharField(max_length=64)),
('quantity', models.IntegerField(default=0)),
('external_id', models.CharField(blank=True, max_length=255)),
('payload', models.JSONField(blank=True, default=dict)),
('error', models.TextField(blank=True)),
('connection', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='events', to='pos_sync.posconnection')),
],
options={
'ordering': ['-created_at'],
},
),
]
+61
View File
@@ -0,0 +1,61 @@
from django.db import models
from core.models import TimeStampedModel, UUIDPrimaryKeyModel
class POSConnection(UUIDPrimaryKeyModel, TimeStampedModel):
class Provider(models.TextChoices):
GENERIC = "generic", "Generic REST"
SQUARE = "square", "Square"
LIGHTSPEED = "lightspeed", "Lightspeed"
BINDERPOS = "binderpos", "BinderPOS"
CRYSTALCOMMERCE = "crystalcommerce", "CrystalCommerce"
name = models.CharField(max_length=120)
provider = models.CharField(
max_length=32, choices=Provider.choices, default=Provider.GENERIC
)
api_base_url = models.URLField(blank=True)
api_token = models.CharField(max_length=255, blank=True)
webhook_secret = models.CharField(max_length=255, blank=True)
is_active = models.BooleanField(default=True)
class Meta:
ordering = ["name"]
def __str__(self) -> str:
return self.name
class SyncEvent(UUIDPrimaryKeyModel, TimeStampedModel):
class Direction(models.TextChoices):
INBOUND = "inbound", "POS → site"
OUTBOUND = "outbound", "Site → POS"
class Status(models.TextChoices):
PENDING = "pending", "Pending"
DONE = "done", "Done"
FAILED = "failed", "Failed"
connection = models.ForeignKey(
POSConnection,
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="events",
)
direction = models.CharField(max_length=16, choices=Direction.choices)
status = models.CharField(
max_length=16, choices=Status.choices, default=Status.PENDING
)
sku = models.CharField(max_length=64)
quantity = models.IntegerField(default=0)
external_id = models.CharField(max_length=255, blank=True)
payload = models.JSONField(default=dict, blank=True)
error = models.TextField(blank=True)
class Meta:
ordering = ["-created_at"]
def __str__(self) -> str:
return f"{self.direction} {self.sku} × {self.quantity} · {self.status}"
+131
View File
@@ -0,0 +1,131 @@
"""POS inventory sync. Inbound webhooks decrement shop stock; outbound queues reserve POS."""
from __future__ import annotations
import logging
import requests
from django.conf import settings
from django.db import transaction
from pos_sync.models import POSConnection, SyncEvent
from shop.models import Product
from shop.services import ShopError, adjust_stock
logger = logging.getLogger(__name__)
class POSSyncError(RuntimeError):
pass
def active_connection() -> POSConnection | None:
return POSConnection.objects.filter(is_active=True).order_by("created_at").first()
def apply_inbound_sale(
*,
sku: str,
quantity: int,
external_id: str = "",
payload: dict | None = None,
connection: POSConnection | None = None,
) -> SyncEvent:
sku = (sku or "").strip()
if not sku:
raise POSSyncError("SKU is required.")
if quantity < 1:
raise POSSyncError("Quantity must be at least 1.")
product = Product.objects.filter(sku__iexact=sku).first()
if product is None:
raise POSSyncError(f"Unknown SKU {sku}.")
event = SyncEvent.objects.create(
connection=connection or active_connection(),
direction=SyncEvent.Direction.INBOUND,
status=SyncEvent.Status.PENDING,
sku=product.sku,
quantity=quantity,
external_id=external_id,
payload=payload or {},
)
try:
with transaction.atomic():
locked = Product.objects.select_for_update().get(pk=product.pk)
if locked.track_inventory and locked.fulfillment == Product.Fulfillment.STOCKED:
adjust_stock(locked, -quantity)
event.status = SyncEvent.Status.DONE
event.save(update_fields=["status", "updated_at"])
except ShopError as exc:
event.status = SyncEvent.Status.FAILED
event.error = str(exc)
event.save(update_fields=["status", "error", "updated_at"])
raise POSSyncError(str(exc)) from exc
return event
def enqueue_online_sale(order) -> list[SyncEvent]:
"""Queue outbound POS reserves after an online shop sale."""
events = []
connection = active_connection()
for item in order.items.all():
sku = item.sku
if not sku:
continue
events.append(
SyncEvent.objects.create(
connection=connection,
direction=SyncEvent.Direction.OUTBOUND,
status=SyncEvent.Status.PENDING,
sku=sku,
quantity=item.quantity,
external_id=order.number,
payload={"order_id": str(order.pk), "order_number": order.number},
)
)
return events
def _post_reserve(connection: POSConnection, event: SyncEvent) -> None:
base = (connection.api_base_url or settings.POS_API_BASE_URL or "").rstrip("/")
token = connection.api_token or (settings.POS_API_TOKEN or "")
if not base:
raise POSSyncError("POS API base URL is not configured.")
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
response = requests.post(
f"{base}/inventory/reserve",
json={
"sku": event.sku,
"quantity": event.quantity,
"external_id": event.external_id,
"source": "online",
},
headers=headers,
timeout=15,
)
response.raise_for_status()
def dispatch_pending_outbound() -> int:
sent = 0
for event in SyncEvent.objects.filter(
direction=SyncEvent.Direction.OUTBOUND,
status=SyncEvent.Status.PENDING,
)[:50]:
connection = event.connection or active_connection()
try:
if connection is None:
raise POSSyncError("No active POS connection.")
_post_reserve(connection, event)
except Exception as exc: # noqa: BLE001
logger.exception("POS outbound failed for %s", event.sku)
event.status = SyncEvent.Status.FAILED
event.error = str(exc)
event.save(update_fields=["status", "error", "updated_at"])
else:
event.status = SyncEvent.Status.DONE
event.error = ""
event.save(update_fields=["status", "error", "updated_at"])
sent += 1
return sent
@@ -0,0 +1,22 @@
{% extends "portal_base.html" %}
{% block title %}POS sync · Portal{% endblock %}
{% block topbar_title %}POS sync{% endblock %}
{% block portal_content %}
<p class="muted">Inbound POS sales decrement shop stock. Online sales queue outbound reserves.</p>
<table class="table">
<thead><tr><th>When</th><th>Direction</th><th>SKU</th><th>Qty</th><th>Status</th></tr></thead>
<tbody>
{% for event in events %}
<tr>
<td>{{ event.created_at }}</td>
<td>{{ event.get_direction_display }}</td>
<td>{{ event.sku }}</td>
<td>{{ event.quantity }}</td>
<td>{{ event.get_status_display }}</td>
</tr>
{% empty %}
<tr><td colspan="5" class="empty-state">No sync events yet.</td></tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
+115
View File
@@ -0,0 +1,115 @@
from decimal import Decimal
from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.test import Client, TestCase, override_settings
from django.urls import reverse
from pos_sync.models import POSConnection, SyncEvent
from pos_sync.services import apply_inbound_sale, dispatch_pending_outbound, enqueue_online_sale
from shop.models import Product
from shop.services import add_to_cart, create_order_from_cart, mark_paid
class POSInboundTests(TestCase):
def setUp(self):
self.product = Product.objects.create(
name="Booster pack",
sku="TCG-PACK",
price=Decimal("4.99"),
stock_qty=10,
is_published=True,
)
def test_inbound_sale_decrements_stock(self):
event = apply_inbound_sale(sku="tcg-pack", quantity=3, external_id="pos-1")
self.product.refresh_from_db()
self.assertEqual(self.product.stock_qty, 7)
self.assertEqual(event.status, SyncEvent.Status.DONE)
@override_settings(POS_WEBHOOK_SECRET="pos-secret")
def test_webhook_requires_secret_and_updates_stock(self):
url = reverse("pos_sync:inventory_webhook")
anon = Client().post(
url,
data='{"sku":"TCG-PACK","quantity":1}',
content_type="application/json",
)
self.assertEqual(anon.status_code, 401)
ok = Client().post(
url,
data='{"sku":"TCG-PACK","quantity":2,"external_id":"reg-9"}',
content_type="application/json",
HTTP_AUTHORIZATION="Bearer pos-secret",
)
self.assertEqual(ok.status_code, 200)
self.product.refresh_from_db()
self.assertEqual(self.product.stock_qty, 8)
class POSOutboundTests(TestCase):
def setUp(self):
self.product = Product.objects.create(
name="Figure",
sku="FIG-1",
price=Decimal("12.00"),
stock_qty=5,
is_published=True,
)
self.connection = POSConnection.objects.create(
name="Counter",
api_base_url="https://pos.example.com",
api_token="tok",
is_active=True,
)
def test_paid_order_enqueues_outbound(self):
session = self.client.session
add_to_cart(session, self.product, 1)
session.save()
order = create_order_from_cart(self.client.session, email="a@example.com")
mark_paid(order)
event = SyncEvent.objects.get(direction=SyncEvent.Direction.OUTBOUND)
self.assertEqual(event.sku, "FIG-1")
self.assertEqual(event.status, SyncEvent.Status.PENDING)
def test_dispatch_posts_reserve(self):
enqueue_online_sale(
type(
"O",
(),
{
"number": "ORD-1",
"pk": "x",
"items": type(
"M",
(),
{
"all": lambda self: [
type("I", (), {"sku": "FIG-1", "quantity": 1})()
]
},
)(),
},
)()
)
with patch("pos_sync.services.requests.post") as post:
post.return_value.raise_for_status = lambda: None
sent = dispatch_pending_outbound()
self.assertEqual(sent, 1)
post.assert_called_once()
self.assertIn("/inventory/reserve", post.call_args.args[0])
event = SyncEvent.objects.get()
self.assertEqual(event.status, SyncEvent.Status.DONE)
class POSPortalTests(TestCase):
def test_list_requires_login(self):
self.assertEqual(Client().get(reverse("pos_sync:event_list")).status_code, 302)
def test_list_ok_when_logged_in(self):
User = get_user_model()
User.objects.create_user("clerk", password="test-pass-123")
client = Client()
client.login(username="clerk", password="test-pass-123")
self.assertEqual(client.get(reverse("pos_sync:event_list")).status_code, 200)
+10
View File
@@ -0,0 +1,10 @@
from django.urls import path
from pos_sync import views
app_name = "pos_sync"
urlpatterns = [
path("", views.event_list, name="event_list"),
path("webhooks/inventory/", views.inventory_webhook, name="inventory_webhook"),
]
+70
View File
@@ -0,0 +1,70 @@
import json
import logging
from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.http import HttpResponse, HttpResponseBadRequest, JsonResponse
from django.shortcuts import render
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
from pos_sync.models import POSConnection, SyncEvent
from pos_sync.services import POSSyncError, apply_inbound_sale
logger = logging.getLogger(__name__)
def _authorized(request) -> bool:
expected = (settings.POS_WEBHOOK_SECRET or "").strip()
if not expected:
conn = POSConnection.objects.filter(is_active=True).exclude(
webhook_secret=""
).first()
expected = (conn.webhook_secret if conn else "").strip()
if not expected:
return False
header = request.headers.get("Authorization", "")
token = ""
if header.startswith("Bearer "):
token = header[7:].strip()
token = token or request.GET.get("token") or ""
return token == expected
@login_required
def event_list(request):
events = SyncEvent.objects.select_related("connection")[:200]
connections = POSConnection.objects.all()
return render(
request,
"pos_sync/list.html",
{"events": events, "connections": connections},
)
@csrf_exempt
@require_http_methods(["POST"])
def inventory_webhook(request):
if not _authorized(request):
return HttpResponse("unauthorized", status=401)
try:
payload = json.loads(request.body.decode() or "{}")
except json.JSONDecodeError:
return HttpResponseBadRequest("invalid json")
sku = payload.get("sku") or payload.get("SKU") or ""
try:
quantity = int(payload.get("quantity") or payload.get("qty") or 0)
except (TypeError, ValueError):
return HttpResponseBadRequest("invalid quantity")
external_id = str(payload.get("external_id") or payload.get("id") or "")
try:
event = apply_inbound_sale(
sku=sku,
quantity=quantity,
external_id=external_id,
payload=payload,
)
except POSSyncError as exc:
logger.warning("POS inbound rejected: %s", exc)
return JsonResponse({"ok": False, "error": str(exc)}, status=400)
return JsonResponse({"ok": True, "id": str(event.pk), "status": event.status})
+9
View File
@@ -63,6 +63,15 @@ def tianji_tracking(request):
nav_section = "social_accounts"
elif namespace == "social":
nav_section = "social"
elif namespace == "shop_portal":
if url_name.startswith("product"):
nav_section = "shop_products"
elif url_name.startswith("order"):
nav_section = "shop_orders"
elif url_name == "sales":
nav_section = "shop_sales"
elif namespace == "shop":
nav_section = "shop"
return {
"tianji_enabled": getattr(settings, "TIANJI_ENABLED", False)
+4
View File
@@ -70,6 +70,10 @@ def sitemap_xml(request):
from django.apps import apps as django_apps
if django_apps.is_installed("blog"):
paths.append(("blog:list", "0.7", "weekly"))
if django_apps.is_installed("shop"):
paths.append(("shop:list", "0.8", "weekly"))
if django_apps.is_installed("events"):
paths.append(("events:list", "0.8", "weekly"))
urls = []
for name, priority, changefreq in paths:
path = reverse(name)
View File
+10
View File
@@ -0,0 +1,10 @@
from django.contrib import admin
from shipping.models import Shipment
@admin.register(Shipment)
class ShipmentAdmin(admin.ModelAdmin):
list_display = ("order", "status", "carrier", "tracking_number")
list_filter = ("status", "carrier")
search_fields = ("tracking_number", "order__number")
+12
View File
@@ -0,0 +1,12 @@
from django.apps import AppConfig
class ShippingConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "shipping"
verbose_name = "Shipping automation"
def ready(self):
from shipping import hooks
hooks.register()
+31
View File
@@ -0,0 +1,31 @@
from core.registry import (
register_dashboard_collector,
register_feature,
register_portal_nav,
)
def register() -> None:
register_feature("shipping")
register_portal_nav(
section="shipping",
label="Shipping",
url_name="shipping:shipment_list",
group="Retail",
order=50,
)
register_dashboard_collector(_dashboard)
def _dashboard(request) -> dict:
from shipping.models import Shipment
from shop.models import Order
labeled = Shipment.objects.filter(status=Shipment.Status.LABELED).values_list(
"order_id", flat=True
)
return {
"unshipped_orders": Order.objects.filter(status=Order.Status.PAID)
.exclude(pk__in=labeled)
.count()
}
+40
View File
@@ -0,0 +1,40 @@
# Generated by Django 6.1 on 2026-09-06 11:17
import django.db.models.deletion
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('shop', '0001_initial'),
]
operations = [
migrations.CreateModel(
name='Shipment',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('status', models.CharField(choices=[('draft', 'Draft'), ('rated', 'Rated'), ('labeled', 'Labeled'), ('void', 'Void')], default='draft', max_length=16)),
('carrier', models.CharField(blank=True, max_length=32)),
('service', models.CharField(blank=True, max_length=64)),
('tracking_number', models.CharField(blank=True, max_length=64)),
('label_url', models.URLField(blank=True)),
('rate_amount', models.DecimalField(blank=True, decimal_places=2, max_digits=10, null=True)),
('currency', models.CharField(default='usd', max_length=8)),
('provider_shipment_id', models.CharField(blank=True, max_length=255)),
('rates', models.JSONField(blank=True, default=list)),
('weight_oz', models.PositiveIntegerField(default=16)),
('notes', models.TextField(blank=True)),
('order', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='shipments', to='shop.order')),
],
options={
'ordering': ['-created_at'],
},
),
]
+33
View File
@@ -0,0 +1,33 @@
from django.db import models
from core.models import TimeStampedModel, UUIDPrimaryKeyModel
from shop.models import Order
class Shipment(UUIDPrimaryKeyModel, TimeStampedModel):
class Status(models.TextChoices):
DRAFT = "draft", "Draft"
RATED = "rated", "Rated"
LABELED = "labeled", "Labeled"
VOID = "void", "Void"
order = models.ForeignKey(Order, on_delete=models.PROTECT, related_name="shipments")
status = models.CharField(
max_length=16, choices=Status.choices, default=Status.DRAFT
)
carrier = models.CharField(max_length=32, blank=True)
service = models.CharField(max_length=64, blank=True)
tracking_number = models.CharField(max_length=64, blank=True)
label_url = models.URLField(blank=True)
rate_amount = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
currency = models.CharField(max_length=8, default="usd")
provider_shipment_id = models.CharField(max_length=255, blank=True)
rates = models.JSONField(default=list, blank=True)
weight_oz = models.PositiveIntegerField(default=16)
notes = models.TextField(blank=True)
class Meta:
ordering = ["-created_at"]
def __str__(self) -> str:
return f"{self.order.number} · {self.tracking_number or self.status}"
+184
View File
@@ -0,0 +1,184 @@
"""EasyPost-style rates/labels plus Pirate Ship CSV export."""
from __future__ import annotations
import csv
import io
import logging
from decimal import Decimal
import requests
from django.conf import settings
from shipping.models import Shipment
from shop.models import Order
logger = logging.getLogger(__name__)
class ShippingError(RuntimeError):
pass
def _easypost_key() -> str:
return (getattr(settings, "EASYPOST_API_KEY", "") or "").strip()
def quote_rates(shipment: Shipment) -> list[dict]:
"""Return carrier rates. Uses EasyPost when configured; otherwise a USPS stub."""
key = _easypost_key()
if not key:
rates = [
{
"id": "stub-usps-ground",
"carrier": "USPS",
"service": "GroundAdvantage",
"rate": "5.40",
"currency": "USD",
},
{
"id": "stub-usps-priority",
"carrier": "USPS",
"service": "Priority",
"rate": "9.80",
"currency": "USD",
},
]
shipment.rates = rates
shipment.status = Shipment.Status.RATED
shipment.save(update_fields=["rates", "status", "updated_at"])
return rates
addr = shipment.order.shipping_address or {}
response = requests.post(
"https://api.easypost.com/v2/shipments",
auth=(key, ""),
json={
"shipment": {
"to_address": {
"name": shipment.order.customer_name or shipment.order.email,
"street1": addr.get("line1") or "",
"street2": addr.get("line2") or "",
"city": addr.get("city") or "",
"state": addr.get("state") or "",
"zip": addr.get("zip") or "",
"country": addr.get("country") or "US",
},
"from_address": {
"name": settings.SITE_NAME,
"street1": getattr(settings, "SHIP_FROM_LINE1", "") or "",
"city": getattr(settings, "SHIP_FROM_CITY", "") or "",
"state": getattr(settings, "SHIP_FROM_STATE", "") or "",
"zip": getattr(settings, "SHIP_FROM_ZIP", "") or "",
"country": "US",
},
"parcel": {"weight": shipment.weight_oz},
}
},
timeout=20,
)
response.raise_for_status()
data = response.json()
rates = data.get("rates") or []
shipment.provider_shipment_id = data.get("id") or ""
shipment.rates = rates
shipment.status = Shipment.Status.RATED
shipment.save(
update_fields=["provider_shipment_id", "rates", "status", "updated_at"]
)
return rates
def buy_label(shipment: Shipment, *, rate_id: str = "") -> Shipment:
rates = shipment.rates or quote_rates(shipment)
chosen = None
if rate_id:
chosen = next((r for r in rates if str(r.get("id")) == str(rate_id)), None)
if chosen is None and rates:
chosen = rates[0]
if chosen is None:
raise ShippingError("No shipping rates available.")
key = _easypost_key()
if key and shipment.provider_shipment_id:
response = requests.post(
f"https://api.easypost.com/v2/shipments/{shipment.provider_shipment_id}/buy",
auth=(key, ""),
json={"rate": {"id": chosen.get("id")}},
timeout=20,
)
response.raise_for_status()
data = response.json()
postage = data.get("postage_label") or {}
tracking = data.get("tracking_code") or ""
shipment.label_url = postage.get("label_url") or ""
shipment.tracking_number = tracking
else:
shipment.tracking_number = f"STUB{shipment.order.number[-6:]}"
shipment.label_url = ""
shipment.carrier = chosen.get("carrier") or ""
shipment.service = chosen.get("service") or ""
try:
shipment.rate_amount = Decimal(str(chosen.get("rate") or "0"))
except Exception:
shipment.rate_amount = None
shipment.status = Shipment.Status.LABELED
shipment.save(
update_fields=[
"carrier",
"service",
"tracking_number",
"label_url",
"rate_amount",
"status",
"updated_at",
]
)
return shipment
def create_shipment_for_order(order: Order, *, weight_oz: int = 16) -> Shipment:
if order.status not in {Order.Status.PAID, Order.Status.FULFILLED}:
raise ShippingError("Ship paid orders only.")
return Shipment.objects.create(order=order, weight_oz=weight_oz)
def pirate_ship_csv(orders=None) -> str:
"""CSV Pirate Ship can map: name, address, city, state, zip, email."""
if orders is None:
shipped = Shipment.objects.filter(
status=Shipment.Status.LABELED
).values_list("order_id", flat=True)
orders = Order.objects.filter(status=Order.Status.PAID).exclude(pk__in=shipped)
buffer = io.StringIO()
writer = csv.writer(buffer)
writer.writerow(
[
"Order Number",
"Name",
"Address 1",
"Address 2",
"City",
"State",
"Zip",
"Country",
"Email",
]
)
for order in orders:
addr = order.shipping_address or {}
writer.writerow(
[
order.number,
order.customer_name or order.email,
addr.get("line1") or "",
addr.get("line2") or "",
addr.get("city") or "",
addr.get("state") or "",
addr.get("zip") or "",
addr.get("country") or "US",
order.email,
]
)
return buffer.getvalue()
@@ -0,0 +1,25 @@
{% extends "portal_base.html" %}
{% block title %}{{ shipment.order.number }} · Shipping{% endblock %}
{% block topbar_title %}{{ shipment.order.number }}{% endblock %}
{% block portal_content %}
<p>{{ shipment.order.email }} · {{ shipment.get_status_display }}</p>
{% if shipment.tracking_number %}<p>Tracking: {{ shipment.tracking_number }}</p>{% endif %}
{% if shipment.label_url %}<p><a href="{{ shipment.label_url }}">Download label</a></p>{% endif %}
{% if shipment.status != 'labeled' %}
<form method="post" action="{% url 'shipping:shipment_buy' shipment.pk %}">
{% csrf_token %}
{% for rate in shipment.rates %}
<label>
<input type="radio" name="rate_id" value="{{ rate.id }}" {% if forloop.first %}checked{% endif %}>
{{ rate.carrier }} {{ rate.service }} — {{ rate.rate }} {{ rate.currency }}
</label><br>
{% empty %}
<p>No rates yet.</p>
{% endfor %}
{% if shipment.rates %}
<button class="btn btn-primary" type="submit">Buy label</button>
{% endif %}
</form>
{% endif %}
<p><a href="{% url 'shipping:shipment_list' %}">← All shipments</a></p>
{% endblock %}
@@ -0,0 +1,43 @@
{% extends "portal_base.html" %}
{% block title %}Shipping · Portal{% endblock %}
{% block topbar_title %}Shipping{% endblock %}
{% block portal_content %}
<p><a class="btn btn-ghost" href="{% url 'shipping:pirate_ship_export' %}">Pirate Ship CSV</a></p>
<h2>Unshipped paid orders</h2>
<table class="table">
<thead><tr><th>Order</th><th>Email</th><th></th></tr></thead>
<tbody>
{% for order in unshipped %}
<tr>
<td>{{ order.number }}</td>
<td>{{ order.email }}</td>
<td>
<form method="post" action="{% url 'shipping:shipment_create' %}">
{% csrf_token %}
<input type="hidden" name="order" value="{{ order.pk }}">
<input name="weight_oz" type="number" min="1" value="16" style="width:5rem"> oz
<button class="btn btn-primary" type="submit">Quote rates</button>
</form>
</td>
</tr>
{% empty %}
<tr><td colspan="3" class="empty-state">No paid orders waiting to ship.</td></tr>
{% endfor %}
</tbody>
</table>
<h2>Shipments</h2>
<table class="table">
<thead><tr><th>Order</th><th>Status</th><th>Tracking</th></tr></thead>
<tbody>
{% for shipment in shipments %}
<tr>
<td><a href="{% url 'shipping:shipment_detail' shipment.pk %}">{{ shipment.order.number }}</a></td>
<td>{{ shipment.get_status_display }}</td>
<td>{{ shipment.tracking_number|default:"—" }}</td>
</tr>
{% empty %}
<tr><td colspan="3" class="empty-state">No shipments yet.</td></tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
+84
View File
@@ -0,0 +1,84 @@
from decimal import Decimal
from django.contrib.auth import get_user_model
from django.test import Client, TestCase
from django.urls import reverse
from shipping.models import Shipment
from shipping.services import buy_label, create_shipment_for_order, pirate_ship_csv, quote_rates
from shop.models import Order
class ShippingServiceTests(TestCase):
def setUp(self):
self.order = Order.objects.create(
number="ORD-20260101-001",
email="buyer@example.com",
customer_name="Pat Lee",
status=Order.Status.PAID,
amount=Decimal("18.00"),
shipping_address={
"line1": "123 Main",
"city": "Aurora",
"state": "IL",
"zip": "60506",
"country": "US",
},
)
def test_quote_and_buy_stub_label(self):
shipment = create_shipment_for_order(self.order)
rates = quote_rates(shipment)
self.assertGreaterEqual(len(rates), 1)
shipment.refresh_from_db()
self.assertEqual(shipment.status, Shipment.Status.RATED)
buy_label(shipment, rate_id=rates[0]["id"])
shipment.refresh_from_db()
self.assertEqual(shipment.status, Shipment.Status.LABELED)
self.assertTrue(shipment.tracking_number)
def test_pirate_ship_csv_includes_unshipped(self):
csv_body = pirate_ship_csv()
self.assertIn("ORD-20260101-001", csv_body)
self.assertIn("123 Main", csv_body)
self.assertIn("buyer@example.com", csv_body)
def test_csv_omits_labeled_orders(self):
shipment = create_shipment_for_order(self.order)
quote_rates(shipment)
buy_label(shipment)
self.assertNotIn(self.order.number, pirate_ship_csv())
class ShippingPortalTests(TestCase):
def setUp(self):
User = get_user_model()
self.user = User.objects.create_user("shipper", password="test-pass-123")
self.client = Client()
self.client.login(username="shipper", password="test-pass-123")
self.order = Order.objects.create(
number="ORD-20260101-002",
email="a@example.com",
status=Order.Status.PAID,
amount=Decimal("9.00"),
shipping_address={"line1": "9 Oak", "city": "Town", "state": "IL", "zip": "60189"},
)
def test_list_requires_login(self):
self.assertEqual(Client().get(reverse("shipping:shipment_list")).status_code, 302)
def test_create_shipment_from_portal(self):
response = self.client.post(
reverse("shipping:shipment_create"),
{"order": str(self.order.pk), "weight_oz": "8"},
)
self.assertEqual(response.status_code, 302)
shipment = Shipment.objects.get()
self.assertEqual(shipment.weight_oz, 8)
self.assertEqual(shipment.status, Shipment.Status.RATED)
def test_csv_export(self):
response = self.client.get(reverse("shipping:pirate_ship_export"))
self.assertEqual(response.status_code, 200)
self.assertEqual(response["Content-Type"], "text/csv")
self.assertIn(b"ORD-20260101-002", response.content)
+13
View File
@@ -0,0 +1,13 @@
from django.urls import path
from shipping import views
app_name = "shipping"
urlpatterns = [
path("", views.shipment_list, name="shipment_list"),
path("new/", views.shipment_create, name="shipment_create"),
path("export/pirate-ship.csv", views.pirate_ship_export, name="pirate_ship_export"),
path("<uuid:pk>/", views.shipment_detail, name="shipment_detail"),
path("<uuid:pk>/buy/", views.shipment_buy, name="shipment_buy"),
]
+86
View File
@@ -0,0 +1,86 @@
import logging
from django.contrib import messages
from django.contrib.auth.decorators import login_required
from django.http import HttpResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.views.decorators.http import require_http_methods, require_POST
from shipping.models import Shipment
from shipping.services import (
ShippingError,
buy_label,
create_shipment_for_order,
pirate_ship_csv,
quote_rates,
)
from shop.models import Order
logger = logging.getLogger(__name__)
@login_required
def shipment_list(request):
shipments = Shipment.objects.select_related("order")[:200]
labeled = Shipment.objects.filter(status=Shipment.Status.LABELED).values_list(
"order_id", flat=True
)
unshipped = Order.objects.filter(status=Order.Status.PAID).exclude(pk__in=labeled)
return render(
request,
"shipping/list.html",
{"shipments": shipments, "unshipped": unshipped},
)
@login_required
@require_POST
def shipment_create(request):
order = get_object_or_404(Order, pk=request.POST.get("order"))
try:
weight = int(request.POST.get("weight_oz") or "16")
except ValueError:
weight = 16
try:
shipment = create_shipment_for_order(order, weight_oz=weight)
quote_rates(shipment)
except ShippingError as exc:
messages.error(request, str(exc))
return redirect("shipping:shipment_list")
except Exception as exc: # noqa: BLE001
logger.exception("rate quote failed")
messages.error(request, f"Could not quote rates: {exc}")
return redirect("shipping:shipment_list")
return redirect("shipping:shipment_detail", pk=shipment.pk)
@login_required
def shipment_detail(request, pk):
shipment = get_object_or_404(Shipment.objects.select_related("order"), pk=pk)
return render(request, "shipping/detail.html", {"shipment": shipment})
@login_required
@require_POST
def shipment_buy(request, pk):
shipment = get_object_or_404(Shipment, pk=pk)
rate_id = (request.POST.get("rate_id") or "").strip()
try:
buy_label(shipment, rate_id=rate_id)
except ShippingError as exc:
messages.error(request, str(exc))
except Exception as exc: # noqa: BLE001
logger.exception("buy label failed")
messages.error(request, f"Could not buy label: {exc}")
else:
messages.success(request, f"Label created for {shipment.order.number}.")
return redirect("shipping:shipment_detail", pk=shipment.pk)
@login_required
@require_http_methods(["GET"])
def pirate_ship_export(request):
body = pirate_ship_csv()
response = HttpResponse(body, content_type="text/csv")
response["Content-Disposition"] = 'attachment; filename="pirate-ship-orders.csv"'
return response
View File
+24
View File
@@ -0,0 +1,24 @@
from django.contrib import admin
from shop.models import Order, OrderItem, Product
@admin.register(Product)
class ProductAdmin(admin.ModelAdmin):
list_display = ("name", "sku", "price", "stock_qty", "fulfillment", "is_published")
list_filter = ("fulfillment", "is_published")
search_fields = ("name", "sku")
prepopulated_fields = {"slug": ("name",)}
class OrderItemInline(admin.TabularInline):
model = OrderItem
extra = 0
@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
list_display = ("number", "email", "amount", "status", "created_at")
list_filter = ("status",)
search_fields = ("number", "email", "customer_name")
inlines = [OrderItemInline]
+12
View File
@@ -0,0 +1,12 @@
from django.apps import AppConfig
class ShopConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "shop"
verbose_name = "E-commerce"
def ready(self):
from shop import hooks
hooks.register()
+53
View File
@@ -0,0 +1,53 @@
from core.registry import (
register_dashboard_collector,
register_feature,
register_portal_nav,
register_public_nav,
)
def register() -> None:
register_feature("shop")
register_public_nav(section="shop", label="Shop", url_name="shop:list", order=30)
register_portal_nav(
section="shop_sales",
label="Sales",
url_name="shop_portal:sales",
group="Retail",
order=5,
)
register_portal_nav(
section="shop_products",
label="Products",
url_name="shop_portal:product_list",
group="Retail",
order=10,
)
register_portal_nav(
section="shop_orders",
label="Orders",
url_name="shop_portal:order_list",
group="Retail",
order=20,
)
register_dashboard_collector(_dashboard)
def _dashboard(request) -> dict:
from shop.models import Order, Product
from shop.stats import sales_summary
sales = sales_summary()
return {
"open_shop_orders": Order.objects.filter(
status__in=[Order.Status.OPEN, Order.Status.PAID]
).count(),
"low_stock_products": Product.objects.filter(
is_published=True,
track_inventory=True,
fulfillment=Product.Fulfillment.STOCKED,
stock_qty__lte=3,
).count(),
"shop_sales_30d": sales["order_count"],
"shop_revenue_30d": sales["revenue"],
}
+80
View File
@@ -0,0 +1,80 @@
# Generated by Django 6.1 on 2026-09-06 11:17
import django.db.models.deletion
import uuid
from decimal import Decimal
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Order',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('number', models.CharField(max_length=32, unique=True)),
('email', models.EmailField(max_length=254)),
('customer_name', models.CharField(blank=True, max_length=200)),
('status', models.CharField(choices=[('draft', 'Draft'), ('open', 'Open'), ('paid', 'Paid'), ('fulfilled', 'Fulfilled'), ('cancelled', 'Cancelled')], default='draft', max_length=16)),
('amount', models.DecimalField(decimal_places=2, default=Decimal('0'), max_digits=10)),
('currency', models.CharField(default='usd', max_length=8)),
('shipping_address', models.JSONField(blank=True, default=dict)),
('stripe_checkout_session_id', models.CharField(blank=True, max_length=255)),
('hosted_checkout_url', models.URLField(blank=True)),
('paid_at', models.DateTimeField(blank=True, null=True)),
('notes', models.TextField(blank=True)),
],
options={
'ordering': ['-created_at'],
},
),
migrations.CreateModel(
name='Product',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('name', models.CharField(max_length=200)),
('slug', models.SlugField(max_length=220, unique=True)),
('sku', models.CharField(max_length=64, unique=True)),
('description', models.TextField(blank=True)),
('price', models.DecimalField(decimal_places=2, max_digits=10)),
('currency', models.CharField(default='usd', max_length=8)),
('fulfillment', models.CharField(choices=[('stocked', 'On-hand stock'), ('made_to_order', 'Made to order')], default='stocked', max_length=16)),
('stock_qty', models.IntegerField(default=0)),
('print_minutes', models.PositiveIntegerField(default=0, help_text='Estimated print time per unit (made-to-order).')),
('filament_grams', models.PositiveIntegerField(default=0)),
('is_published', models.BooleanField(default=False)),
('track_inventory', models.BooleanField(default=True)),
],
options={
'ordering': ['name'],
},
),
migrations.CreateModel(
name='OrderItem',
fields=[
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('name', models.CharField(max_length=200)),
('sku', models.CharField(max_length=64)),
('quantity', models.PositiveIntegerField(default=1)),
('unit_price', models.DecimalField(decimal_places=2, max_digits=10)),
('print_minutes', models.PositiveIntegerField(default=0)),
('order', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='items', to='shop.order')),
('product', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='order_items', to='shop.product')),
],
options={
'ordering': ['created_at'],
},
),
]
View File
+118
View File
@@ -0,0 +1,118 @@
from decimal import Decimal
from django.db import models
from django.urls import reverse
from django.utils.text import slugify
from core.models import TimeStampedModel, UUIDPrimaryKeyModel
class Product(UUIDPrimaryKeyModel, TimeStampedModel):
class Fulfillment(models.TextChoices):
STOCKED = "stocked", "On-hand stock"
MADE_TO_ORDER = "made_to_order", "Made to order"
name = models.CharField(max_length=200)
slug = models.SlugField(max_length=220, unique=True)
sku = models.CharField(max_length=64, unique=True)
description = models.TextField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
currency = models.CharField(max_length=8, default="usd")
fulfillment = models.CharField(
max_length=16,
choices=Fulfillment.choices,
default=Fulfillment.STOCKED,
)
stock_qty = models.IntegerField(default=0)
print_minutes = models.PositiveIntegerField(
default=0,
help_text="Estimated print time per unit (made-to-order).",
)
filament_grams = models.PositiveIntegerField(default=0)
is_published = models.BooleanField(default=False)
track_inventory = models.BooleanField(default=True)
class Meta:
ordering = ["name"]
def __str__(self) -> str:
return f"{self.name} ({self.sku})"
def get_absolute_url(self) -> str:
return reverse("shop:detail", kwargs={"slug": self.slug})
@property
def amount_cents(self) -> int:
return int((self.price * Decimal("100")).quantize(Decimal("1")))
def save(self, *args, **kwargs):
if not self.slug:
base = slugify(self.name)[:200] or "product"
slug = base
n = 2
while Product.objects.filter(slug=slug).exclude(pk=self.pk).exists():
slug = f"{base}-{n}"
n += 1
self.slug = slug
if not self.sku:
self.sku = (self.slug or "sku").upper().replace("-", "")[:64]
super().save(*args, **kwargs)
class Order(UUIDPrimaryKeyModel, TimeStampedModel):
class Status(models.TextChoices):
DRAFT = "draft", "Draft"
OPEN = "open", "Open"
PAID = "paid", "Paid"
FULFILLED = "fulfilled", "Fulfilled"
CANCELLED = "cancelled", "Cancelled"
number = models.CharField(max_length=32, unique=True)
email = models.EmailField()
customer_name = models.CharField(max_length=200, blank=True)
status = models.CharField(
max_length=16, choices=Status.choices, default=Status.DRAFT
)
amount = models.DecimalField(max_digits=10, decimal_places=2, default=Decimal("0"))
currency = models.CharField(max_length=8, default="usd")
shipping_address = models.JSONField(default=dict, blank=True)
stripe_checkout_session_id = models.CharField(max_length=255, blank=True)
hosted_checkout_url = models.URLField(blank=True)
paid_at = models.DateTimeField(null=True, blank=True)
notes = models.TextField(blank=True)
class Meta:
ordering = ["-created_at"]
def __str__(self) -> str:
return f"{self.number} · {self.email} · {self.amount}"
@property
def amount_cents(self) -> int:
return int((self.amount * Decimal("100")).quantize(Decimal("1")))
class OrderItem(UUIDPrimaryKeyModel, TimeStampedModel):
order = models.ForeignKey(Order, on_delete=models.CASCADE, related_name="items")
product = models.ForeignKey(
Product,
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="order_items",
)
name = models.CharField(max_length=200)
sku = models.CharField(max_length=64)
quantity = models.PositiveIntegerField(default=1)
unit_price = models.DecimalField(max_digits=10, decimal_places=2)
print_minutes = models.PositiveIntegerField(default=0)
class Meta:
ordering = ["created_at"]
def __str__(self) -> str:
return f"{self.quantity}× {self.name}"
@property
def line_total(self) -> Decimal:
return self.unit_price * self.quantity
+20
View File
@@ -0,0 +1,20 @@
from django.urls import path
from shop import views
app_name = "shop_portal"
urlpatterns = [
path("sales/", views.portal_sales, name="sales"),
path("products/", views.portal_product_list, name="product_list"),
path("products/new/", views.portal_product_edit, name="product_new"),
path("products/<uuid:pk>/", views.portal_product_edit, name="product_edit"),
path(
"products/<uuid:pk>/stock/",
views.portal_stock_adjust,
name="product_stock",
),
path("orders/", views.portal_order_list, name="order_list"),
path("orders/<uuid:pk>/", views.portal_order_detail, name="order_detail"),
path("webhooks/stripe/", views.stripe_webhook, name="stripe_webhook"),
]
+16
View File
@@ -0,0 +1,16 @@
from django.urls import path
from shop import views
app_name = "shop"
urlpatterns = [
path("", views.product_list, name="list"),
path("cart/", views.cart_view, name="cart"),
path("cart/add/<slug:slug>/", views.cart_add, name="cart_add"),
path("cart/update/<slug:slug>/", views.cart_update, name="cart_update"),
path("checkout/", views.checkout, name="checkout"),
path("checkout/<uuid:pk>/success/", views.checkout_success, name="checkout_success"),
path("checkout/<uuid:pk>/cancel/", views.checkout_cancel, name="checkout_cancel"),
path("<slug:slug>/", views.product_detail, name="detail"),
]
+305
View File
@@ -0,0 +1,305 @@
"""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, Sum
from django.utils import timezone
from shop.models import Order, OrderItem, Product
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 = "",
) -> 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(),
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 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"):
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 = "") -> 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
+166
View File
@@ -0,0 +1,166 @@
"""Sales dashboard aggregates for FEATURE_SHOP."""
from __future__ import annotations
from datetime import date, datetime, time, timedelta
from decimal import Decimal
from django.conf import settings
from django.db.models import Count, F, Max, Q, Sum
from django.utils import timezone
from shop.models import Order, OrderItem
SOLD_STATUSES = (Order.Status.PAID, Order.Status.FULFILLED)
SALES_WINDOW_DAYS = 30
def _money(value) -> Decimal:
return (value or Decimal("0")).quantize(Decimal("0.01"))
def _window_start(days: int):
today = timezone.localdate()
start_date = today - timedelta(days=days - 1)
start_dt = timezone.make_aware(
datetime.combine(start_date, time.min),
timezone.get_current_timezone(),
)
return today, start_date, start_dt
def _sold_orders(start_dt=None):
qs = Order.objects.filter(status__in=SOLD_STATUSES)
if start_dt is None:
return qs
return qs.filter(
Q(paid_at__gte=start_dt) | Q(paid_at__isnull=True, created_at__gte=start_dt)
)
def _sale_date(order) -> date:
when = order.paid_at or order.created_at
return timezone.localtime(when).date()
def _bar_pct(value: Decimal | int | float, peak: Decimal | int | float) -> int:
if not peak:
return 0
if not value:
return 0
return max(8, int(round((float(value) / float(peak)) * 100)))
def sales_summary(*, days: int = SALES_WINDOW_DAYS) -> dict:
"""Order count and revenue for the rolling sales window."""
_, _, start_dt = _window_start(days)
totals = _sold_orders(start_dt).aggregate(
order_count=Count("id"), revenue=Sum("amount")
)
return {
"order_count": int(totals["order_count"] or 0),
"revenue": _money(totals["revenue"]),
}
def sales_dashboard(*, days: int = SALES_WINDOW_DAYS) -> dict:
"""Paid/fulfilled order stats, daily series, and top products."""
_, start_date, start_dt = _window_start(days)
sold = _sold_orders(start_dt)
totals = sold.aggregate(order_count=Count("id"), revenue=Sum("amount"))
order_count = int(totals["order_count"] or 0)
revenue = _money(totals["revenue"])
units = int(
OrderItem.objects.filter(order__in=sold).aggregate(total=Sum("quantity"))[
"total"
]
or 0
)
aov = _money(revenue / order_count) if order_count else Decimal("0.00")
currency = (settings.STRIPE_CURRENCY or "usd").lower()
by_day: dict = {
start_date + timedelta(days=offset): {
"count": 0,
"revenue": Decimal("0.00"),
}
for offset in range(days)
}
for order in sold.only("paid_at", "created_at", "amount"):
day = _sale_date(order)
bucket = by_day.get(day)
if bucket is None:
continue
bucket["count"] += 1
bucket["revenue"] += order.amount or Decimal("0")
peak_count = max((row["count"] for row in by_day.values()), default=0)
peak_revenue = max((row["revenue"] for row in by_day.values()), default=Decimal("0"))
daily_sales = []
daily_revenue = []
for index, (day, row) in enumerate(by_day.items()):
tick = index == 0 or index == days - 1 or day.weekday() == 0
label = f"{day.strftime('%b')} {day.day}"
daily_sales.append(
{
"date": day,
"label": label,
"count": row["count"],
"revenue": _money(row["revenue"]),
"pct": _bar_pct(row["count"], peak_count),
"tick": tick,
"tick_label": label if tick else "",
}
)
daily_revenue.append(
{
"date": day,
"label": label,
"count": row["count"],
"revenue": _money(row["revenue"]),
"pct": _bar_pct(row["revenue"], peak_revenue),
"tick": tick,
"tick_label": label if tick else "",
}
)
product_rows = list(
OrderItem.objects.filter(order__in=sold)
.values("sku")
.annotate(
units=Sum("quantity"),
revenue=Sum(F("unit_price") * F("quantity")),
product_name=Max("product__name"),
item_name=Max("name"),
)
.order_by("-units", "-revenue")[:8]
)
peak_units = max((int(row["units"] or 0) for row in product_rows), default=0)
top_products = []
for row in product_rows:
units_sold = int(row["units"] or 0)
top_products.append(
{
"sku": row["sku"],
"name": row["product_name"] or row["item_name"] or row["sku"],
"units": units_sold,
"revenue": _money(row["revenue"]),
"bar_pct": _bar_pct(units_sold, peak_units),
}
)
recent_orders = list(sold.order_by("-paid_at", "-created_at")[:8])
return {
"days": days,
"currency": currency,
"order_count": order_count,
"revenue": revenue,
"units_sold": units,
"aov": aov,
"daily_sales": daily_sales,
"daily_revenue": daily_revenue,
"top_products": top_products,
"recent_orders": recent_orders,
"has_sales": order_count > 0,
}
+8
View File
@@ -0,0 +1,8 @@
{% extends "base.html" %}
{% block title %}Checkout cancelled{% endblock %}
{% block content %}
<section class="section section-lg"><div class="container">
<h1>Checkout cancelled</h1>
<p>Order {{ order.number }} was not paid. You can return to the <a href="{% url 'shop:cart' %}">cart</a>.</p>
</div></section>
{% endblock %}
+26
View File
@@ -0,0 +1,26 @@
{% extends "base.html" %}
{% block title %}Cart · {{ SITE_NAME }}{% endblock %}
{% block content %}
<section class="section section-lg">
<div class="container">
<h1>Cart</h1>
{% for line in lines %}
<p>
<a href="{{ line.product.get_absolute_url }}">{{ line.product.name }}</a>
· {{ line.quantity }} × {{ line.unit_price }} = {{ line.line_total }}
</p>
<form method="post" action="{% url 'shop:cart_update' line.product.slug %}" style="margin:0 0 16px">
{% csrf_token %}
<input name="quantity" type="number" min="0" value="{{ line.quantity }}">
<button type="submit">Update</button>
</form>
{% empty %}
<p>Cart is empty.</p>
{% endfor %}
{% if lines %}
<p><strong>Total:</strong> {{ total }}</p>
<p><a class="button button-primary" href="{% url 'shop:checkout' %}">Checkout</a></p>
{% endif %}
</div>
</section>
{% endblock %}
+21
View File
@@ -0,0 +1,21 @@
{% extends "base.html" %}
{% block title %}Checkout · {{ SITE_NAME }}{% endblock %}
{% block content %}
<section class="section section-lg">
<div class="container">
<h1>Checkout</h1>
<p>Total: {{ total }}</p>
<form method="post">
{% csrf_token %}
<p><label>Email <input name="email" type="email" required></label></p>
<p><label>Name <input name="customer_name"></label></p>
<p><label>Address <input name="address_line1"></label></p>
<p><label>Address 2 <input name="address_line2"></label></p>
<p><label>City <input name="address_city"></label></p>
<p><label>State <input name="address_state"></label></p>
<p><label>ZIP <input name="address_zip"></label></p>
<button class="button button-primary" type="submit">Pay with Stripe</button>
</form>
</div>
</section>
{% endblock %}
+20
View File
@@ -0,0 +1,20 @@
{% extends "base.html" %}
{% block title %}{{ product.name }} · {{ SITE_NAME }}{% endblock %}
{% block content %}
<section class="section section-lg">
<div class="container">
<p><a href="{% url 'shop:list' %}">← Shop</a></p>
<h1>{{ product.name }}</h1>
<p class="muted">{{ product.price }} {{ product.currency|upper }} · {{ product.sku }}</p>
<div>{{ product.description|linebreaks }}</div>
{% if available is not None %}
<p>{{ available }} in stock</p>
{% endif %}
<form method="post" action="{% url 'shop:cart_add' product.slug %}">
{% csrf_token %}
<label>Qty <input name="quantity" type="number" min="1" value="1"></label>
<button class="button button-primary" type="submit">Add to cart</button>
</form>
</div>
</section>
{% endblock %}
+19
View File
@@ -0,0 +1,19 @@
{% extends "base.html" %}
{% block title %}Shop · {{ SITE_NAME }}{% endblock %}
{% block content %}
<section class="section section-lg">
<div class="container">
<h1 class="text-uppercase">Shop</h1>
<p><a href="{% url 'shop:cart' %}">View cart</a></p>
{% for product in products %}
<article style="margin:0 0 32px">
<h2><a href="{{ product.get_absolute_url }}">{{ product.name }}</a></h2>
<p class="muted">{{ product.price }} {{ product.currency|upper }} · {{ product.sku }}</p>
<p>{{ product.description|truncatewords:40 }}</p>
</article>
{% empty %}
<p>No products yet.</p>
{% endfor %}
</div>
</section>
{% endblock %}
@@ -0,0 +1,21 @@
{% extends "portal_base.html" %}
{% block title %}{{ order.number }} · Portal{% endblock %}
{% block topbar_title %}{{ order.number }}{% endblock %}
{% block portal_content %}
<p><strong>{{ order.email }}</strong> · {{ order.amount }} {{ order.currency|upper }} · {{ order.get_status_display }}</p>
{% if order.customer_name %}<p>{{ order.customer_name }}</p>{% endif %}
<table class="table">
<thead><tr><th>Item</th><th>SKU</th><th>Qty</th><th>Price</th></tr></thead>
<tbody>
{% for item in order.items.all %}
<tr>
<td>{{ item.name }}</td>
<td>{{ item.sku }}</td>
<td>{{ item.quantity }}</td>
<td>{{ item.unit_price }}</td>
</tr>
{% endfor %}
</tbody>
</table>
<p><a href="{% url 'shop_portal:order_list' %}">← All orders</a></p>
{% endblock %}
@@ -0,0 +1,20 @@
{% extends "portal_base.html" %}
{% block title %}Orders · Portal{% endblock %}
{% block topbar_title %}Orders{% endblock %}
{% block portal_content %}
<table class="table">
<thead><tr><th>Number</th><th>Email</th><th>Amount</th><th>Status</th></tr></thead>
<tbody>
{% for order in orders %}
<tr>
<td><a href="{% url 'shop_portal:order_detail' order.pk %}">{{ order.number }}</a></td>
<td>{{ order.email }}</td>
<td>{{ order.amount }} {{ order.currency|upper }}</td>
<td>{{ order.get_status_display }}</td>
</tr>
{% empty %}
<tr><td colspan="4" class="empty-state">No orders yet.</td></tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
@@ -0,0 +1,33 @@
{% extends "portal_base.html" %}
{% block title %}{% if product %}Edit{% else %}New{% endif %} product · Portal{% endblock %}
{% block topbar_title %}{% if product %}Edit product{% else %}New product{% endif %}{% endblock %}
{% block portal_content %}
<form method="post" class="form-grid">
{% csrf_token %}
<div class="field"><label>Name</label><input name="name" required value="{{ product.name|default:'' }}"></div>
<div class="field"><label>SKU</label><input name="sku" value="{{ product.sku|default:'' }}" placeholder="auto from name"></div>
<div class="field"><label>Slug</label><input name="slug" value="{{ product.slug|default:'' }}" placeholder="auto from name"></div>
<div class="field"><label>Price</label><input name="price" type="number" step="0.01" min="0" required value="{{ product.price|default:'' }}"></div>
<div class="field">
<label>Fulfillment</label>
<select name="fulfillment">
<option value="stocked" {% if product.fulfillment == 'stocked' or not product %}selected{% endif %}>On-hand stock</option>
<option value="made_to_order" {% if product.fulfillment == 'made_to_order' %}selected{% endif %}>Made to order</option>
</select>
</div>
<div class="field"><label>Stock qty</label><input name="stock_qty" type="number" value="{{ product.stock_qty|default:0 }}"></div>
<div class="field"><label>Print minutes</label><input name="print_minutes" type="number" min="0" value="{{ product.print_minutes|default:0 }}"></div>
<div class="field"><label>Filament grams</label><input name="filament_grams" type="number" min="0" value="{{ product.filament_grams|default:0 }}"></div>
<div class="field"><label>Description</label><textarea name="description" style="min-height:120px">{{ product.description|default:'' }}</textarea></div>
<label><input type="checkbox" name="is_published" {% if product.is_published %}checked{% endif %}> Published</label>
<label><input type="checkbox" name="track_inventory" {% if product.track_inventory or not product %}checked{% endif %}> Track inventory</label>
<button class="btn btn-primary" type="submit">Save</button>
</form>
{% if product %}
<form method="post" action="{% url 'shop_portal:product_stock' product.pk %}" class="form-grid" style="margin-top:24px">
{% csrf_token %}
<div class="field"><label>Adjust stock (+/)</label><input name="delta" type="number" required value="1"></div>
<button class="btn btn-ghost" type="submit">Apply adjustment</button>
</form>
{% endif %}
{% endblock %}
@@ -0,0 +1,22 @@
{% extends "portal_base.html" %}
{% block title %}Products · Portal{% endblock %}
{% block topbar_title %}Products{% endblock %}
{% block portal_content %}
<p><a class="btn btn-primary" href="{% url 'shop_portal:product_new' %}">New product</a></p>
<table class="table">
<thead><tr><th>Name</th><th>SKU</th><th>Price</th><th>Stock</th><th>Status</th></tr></thead>
<tbody>
{% for product in products %}
<tr>
<td><a href="{% url 'shop_portal:product_edit' product.pk %}">{{ product.name }}</a></td>
<td>{{ product.sku }}</td>
<td>{{ product.price }} {{ product.currency|upper }}</td>
<td>{{ product.stock_qty }}</td>
<td>{% if product.is_published %}Published{% else %}Draft{% endif %}</td>
</tr>
{% empty %}
<tr><td colspan="5" class="empty-state">No products yet.</td></tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
+147
View File
@@ -0,0 +1,147 @@
{% extends "portal_base.html" %}
{% block title %}Sales · Portal{% endblock %}
{% block topbar_title %}Sales{% endblock %}
{% block portal_content %}
<div class="stat-row">
<div class="stat-card">
<div class="label">Orders ({{ days }} days)</div>
<div class="value">{{ order_count }}</div>
</div>
<div class="stat-card">
<div class="label">Revenue ({{ days }} days)</div>
<div class="value">{{ revenue }} {{ currency|upper }}</div>
</div>
<div class="stat-card">
<div class="label">Units sold</div>
<div class="value">{{ units_sold }}</div>
</div>
<div class="stat-card">
<div class="label">Average order</div>
<div class="value">{{ aov }} {{ currency|upper }}</div>
</div>
</div>
<div class="panel">
<div class="panel-h">
<h2>Orders per day</h2>
<span class="muted">Last {{ days }} days</span>
</div>
<div class="panel-b">
<div class="chart-placeholder chart-daily" role="img"
aria-label="Paid orders per day for the last {{ days }} days">
{% for bar in daily_sales %}
<div class="chart-bar-col">
<div class="bar{% if not bar.count %} is-zero{% endif %}"
style="height:{{ bar.pct }}%"
title="{{ bar.label }}: {{ bar.count }} sale{{ bar.count|pluralize }} · {{ bar.revenue }} {{ currency|upper }}"></div>
<div class="chart-bar-meta">
<span class="chart-bar-label">{% if bar.tick %}{{ bar.tick_label }}{% else %}&nbsp;{% endif %}</span>
</div>
</div>
{% endfor %}
</div>
{% if not has_sales %}
<p class="empty-state" style="padding-top:12px">No paid orders in this window.</p>
{% endif %}
</div>
</div>
<div class="panel">
<div class="panel-h">
<h2>Revenue per day</h2>
<span class="muted">Last {{ days }} days</span>
</div>
<div class="panel-b">
<div class="chart-placeholder chart-daily chart-revenue" role="img"
aria-label="Revenue per day for the last {{ days }} days">
{% for bar in daily_revenue %}
<div class="chart-bar-col">
<div class="bar{% if not bar.revenue %} is-zero{% endif %}"
style="height:{{ bar.pct }}%"
title="{{ bar.label }}: {{ bar.revenue }} {{ currency|upper }}"></div>
<div class="chart-bar-meta">
<span class="chart-bar-label">{% if bar.tick %}{{ bar.tick_label }}{% else %}&nbsp;{% endif %}</span>
</div>
</div>
{% endfor %}
</div>
</div>
</div>
<div class="split">
<div class="panel">
<div class="panel-h">
<h2>Top products</h2>
<span class="muted">By units sold</span>
</div>
<div class="panel-b">
{% if top_products %}
<div class="hbar-chart" role="img" aria-label="Top products by units sold">
{% for row in top_products %}
<div class="hbar-row" title="{{ row.name }} · {{ row.units }} sold · {{ row.revenue }} {{ currency|upper }}">
<div class="hbar-label">{{ row.name }}</div>
<div class="hbar-track">
<div class="hbar-fill" style="width:{{ row.bar_pct }}%"></div>
</div>
<div class="hbar-value">{{ row.units }}</div>
</div>
{% endfor %}
</div>
{% else %}
<p class="empty-state" style="margin:0;padding:0">No product sales yet.</p>
{% endif %}
</div>
</div>
<div class="panel">
<div class="panel-h">
<h2>Product mix</h2>
<a class="btn btn-sm btn-ghost" href="{% url 'shop_portal:product_list' %}">Products</a>
</div>
<div class="panel-b" style="padding:0">
<table class="table">
<thead>
<tr><th>Product</th><th>SKU</th><th>Units</th><th>Revenue</th></tr>
</thead>
<tbody>
{% for row in top_products %}
<tr>
<td>{{ row.name }}</td>
<td>{{ row.sku }}</td>
<td>{{ row.units }}</td>
<td>{{ row.revenue }} {{ currency|upper }}</td>
</tr>
{% empty %}
<tr><td colspan="4" class="empty-state">No paid line items in the last {{ days }} days.</td></tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
<div class="panel">
<div class="panel-h">
<h2>Recent sales</h2>
<a class="btn btn-sm btn-ghost" href="{% url 'shop_portal:order_list' %}">All orders</a>
</div>
<div class="panel-b" style="padding:0">
<table class="table">
<thead>
<tr><th>Number</th><th>Customer</th><th>Amount</th><th>Status</th></tr>
</thead>
<tbody>
{% for order in recent_orders %}
<tr>
<td><a href="{% url 'shop_portal:order_detail' order.pk %}">{{ order.number }}</a></td>
<td>{{ order.customer_name|default:order.email }}</td>
<td>{{ order.amount }} {{ order.currency|upper }}</td>
<td>{{ order.get_status_display }}</td>
</tr>
{% empty %}
<tr><td colspan="4" class="empty-state">No paid orders yet.</td></tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
{% endblock %}
+9
View File
@@ -0,0 +1,9 @@
{% extends "base.html" %}
{% block title %}Order {{ order.number }}{% endblock %}
{% block content %}
<section class="section section-lg"><div class="container">
<h1>Thank you</h1>
<p>Order {{ order.number }} is {{ order.get_status_display|lower }}.</p>
<p>A confirmation will go to {{ order.email }}.</p>
</div></section>
{% endblock %}
+254
View File
@@ -0,0 +1,254 @@
from datetime import timedelta
from decimal import Decimal
from django.contrib.auth import get_user_model
from django.core import mail
from django.test import Client, TestCase, override_settings
from django.urls import reverse
from django.utils import timezone
from shop.models import Order, OrderItem, Product
from shop.services import (
ShopError,
add_to_cart,
adjust_stock,
available_qty,
create_order_from_cart,
mark_paid,
next_order_number,
)
from shop.stats import sales_dashboard
def _product(**kwargs):
defaults = dict(
name="Dragon figurine",
sku="TOY-001",
price=Decimal("18.00"),
stock_qty=5,
is_published=True,
fulfillment=Product.Fulfillment.STOCKED,
)
defaults.update(kwargs)
return Product.objects.create(**defaults)
class ShopPublicTests(TestCase):
def test_list_hides_unpublished(self):
_product(name="Live", sku="LIVE-1")
_product(name="Draft", sku="DRAFT-1", is_published=False)
response = Client().get(reverse("shop:list"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Live")
self.assertNotContains(response, "Draft")
def test_add_to_cart_then_checkout_form(self):
product = _product()
client = Client()
response = client.post(
reverse("shop:cart_add", kwargs={"slug": product.slug}),
{"quantity": "2"},
)
self.assertEqual(response.status_code, 302)
cart = client.get(reverse("shop:cart"))
self.assertContains(cart, "Dragon figurine")
self.assertContains(cart, "36.00")
checkout = client.get(reverse("shop:checkout"))
self.assertEqual(checkout.status_code, 200)
self.assertContains(checkout, "Pay with Stripe")
class ShopInventoryTests(TestCase):
def test_stocked_availability_and_adjust(self):
product = _product(stock_qty=4)
self.assertEqual(available_qty(product), 4)
adjust_stock(product, -2)
product.refresh_from_db()
self.assertEqual(product.stock_qty, 2)
with self.assertRaises(ShopError):
adjust_stock(product, -5)
def test_mark_paid_decrements_stock_and_emails(self):
product = _product(stock_qty=3)
session = self.client.session
add_to_cart(session, product, 2)
session.save()
order = create_order_from_cart(
self.client.session, email="buyer@example.com", customer_name="Pat"
)
mark_paid(order)
product.refresh_from_db()
self.assertEqual(product.stock_qty, 1)
order.refresh_from_db()
self.assertEqual(order.status, Order.Status.PAID)
self.assertEqual(len(mail.outbox), 1)
self.assertIn(order.number, mail.outbox[0].subject)
def test_insufficient_stock_blocks_order(self):
product = _product(stock_qty=1)
session = self.client.session
add_to_cart(session, product, 3)
session.save()
with self.assertRaises(ShopError):
create_order_from_cart(self.client.session, email="buyer@example.com")
@override_settings(SHOP_PRINT_QUEUE_LIMIT_MINUTES=60)
def test_made_to_order_queue_limit(self):
product = _product(
sku="MTO-1",
fulfillment=Product.Fulfillment.MADE_TO_ORDER,
print_minutes=30,
stock_qty=0,
)
self.assertEqual(available_qty(product), 2)
def test_next_number_increments(self):
n1 = next_order_number()
Order.objects.create(
number=n1,
email="a@example.com",
amount=Decimal("1.00"),
)
n2 = next_order_number()
self.assertNotEqual(n1, n2)
class ShopPortalTests(TestCase):
def setUp(self):
User = get_user_model()
self.user = User.objects.create_user("merchant", password="test-pass-123")
self.client = Client()
self.client.login(username="merchant", password="test-pass-123")
def test_list_requires_login(self):
anon = Client()
self.assertEqual(anon.get(reverse("shop_portal:product_list")).status_code, 302)
def test_create_and_adjust_stock(self):
response = self.client.post(
reverse("shop_portal:product_new"),
{
"name": "Booster box",
"sku": "TCG-BOX",
"price": "89.99",
"stock_qty": "10",
"fulfillment": "stocked",
"is_published": "on",
"track_inventory": "on",
},
)
self.assertEqual(response.status_code, 302)
product = Product.objects.get(sku="TCG-BOX")
self.assertTrue(product.is_published)
self.assertEqual(product.slug, "booster-box")
adjust = self.client.post(
reverse("shop_portal:product_stock", kwargs={"pk": product.pk}),
{"delta": "-3"},
)
self.assertEqual(adjust.status_code, 302)
product.refresh_from_db()
self.assertEqual(product.stock_qty, 7)
def _sold_order(product, *, qty=1, paid_at=None, status=None, number=None):
n = Order.objects.count() + 1
order = Order.objects.create(
number=number or f"ORD-TEST-{n:04d}",
email=f"buyer{n}@example.com",
customer_name="Pat",
status=status or Order.Status.PAID,
amount=product.price * qty,
currency="usd",
paid_at=paid_at if paid_at is not None else timezone.now(),
)
OrderItem.objects.create(
order=order,
product=product,
name=product.name,
sku=product.sku,
quantity=qty,
unit_price=product.price,
)
return order
class ShopSalesDashboardTests(TestCase):
def setUp(self):
User = get_user_model()
self.user = User.objects.create_user("merchant", password="test-pass-123")
self.client = Client()
self.client.login(username="merchant", password="test-pass-123")
self.dragon = _product(name="Dragon", sku="DRAGON", price=Decimal("18.00"))
self.fox = _product(name="Fox", sku="FOX", price=Decimal("12.00"))
def test_sales_requires_login(self):
anon = Client()
self.assertEqual(anon.get(reverse("shop_portal:sales")).status_code, 302)
def test_empty_dashboard(self):
data = sales_dashboard()
self.assertEqual(data["order_count"], 0)
self.assertEqual(data["revenue"], Decimal("0.00"))
self.assertEqual(len(data["daily_sales"]), 30)
self.assertFalse(data["has_sales"])
response = self.client.get(reverse("shop_portal:sales"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "No paid orders in this window")
self.assertContains(response, "Sales")
self.assertContains(response, "Orders per day")
self.assertContains(response, "Top products")
def test_counts_paid_and_fulfilled_in_window(self):
now = timezone.now()
_sold_order(self.dragon, qty=2, paid_at=now)
_sold_order(self.fox, qty=5, paid_at=now - timedelta(days=2))
_sold_order(
self.dragon,
qty=1,
paid_at=now - timedelta(days=1),
status=Order.Status.FULFILLED,
)
_sold_order(
self.fox,
qty=9,
paid_at=now - timedelta(days=40),
)
open_order = _sold_order(self.dragon, qty=3, paid_at=None, status=Order.Status.OPEN)
open_order.paid_at = None
open_order.save(update_fields=["paid_at"])
cancelled = _sold_order(
self.fox, qty=4, paid_at=now, status=Order.Status.CANCELLED
)
self.assertEqual(cancelled.status, Order.Status.CANCELLED)
data = sales_dashboard()
self.assertEqual(data["order_count"], 3)
self.assertEqual(data["units_sold"], 8)
self.assertEqual(data["revenue"], Decimal("114.00"))
self.assertEqual(data["aov"], Decimal("38.00"))
self.assertEqual([row["sku"] for row in data["top_products"]], ["FOX", "DRAGON"])
self.assertEqual(data["top_products"][0]["units"], 5)
self.assertEqual(data["top_products"][1]["units"], 3)
today_bar = data["daily_sales"][-1]
self.assertEqual(today_bar["count"], 1)
response = self.client.get(reverse("shop_portal:sales"))
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Dragon")
self.assertContains(response, "Fox")
self.assertContains(response, "114.00")
self.assertContains(response, "Orders per day")
self.assertContains(response, "Revenue per day")
def test_portal_nav_and_home_stats(self):
_sold_order(self.dragon, qty=1)
home = self.client.get(reverse("dashboard:home"))
self.assertEqual(home.status_code, 200)
self.assertEqual(home.context["shop_sales_30d"], 1)
self.assertEqual(home.context["shop_revenue_30d"], Decimal("18.00"))
self.assertContains(home, reverse("shop_portal:sales"))
self.assertContains(home, "Sales (30 days)")
products = self.client.get(reverse("shop_portal:product_list"))
self.assertContains(products, "Sales")
sales = self.client.get(reverse("shop_portal:sales"))
self.assertContains(sales, 'class="active"')
+274
View File
@@ -0,0 +1,274 @@
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")