Add a shop sales dashboard with 30-day charts (#4)
CI / test (pull_request) Successful in 33s

## Summary
- Closes #3
- Portal Retail **Sales** page: 30-day orders/revenue/units/AOV, daily charts, top products
- Home dashboard cards for 30-day shop sales and revenue
- Stacks on #2. Merge that first, then retarget `master` if needed.

## Test plan
- [ ] Sales page requires login; empty state with no paid orders
- [ ] Paid/fulfilled in the last 30 days counted; unpaid/old excluded
- [ ] Home dashboard shows 30-day sales count
- [ ] `manage.py test shop`

Reviewed-on: #4
This commit was merged in pull request #4.
This commit is contained in:
2026-09-06 18:35:20 -07:00
parent b2029b6c0f
commit c7d78fada8
15 changed files with 522 additions and 6 deletions
+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,
}