web_django_template/site/shop/stats.py

"""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,
    }