from contextlib import suppress from datetime import date, timedelta from constance import config from django.db.models import Count, F, QuerySet, Sum from django.db.models.functions import Coalesce, TruncDate from django.utils.timezone import now from engine.core.models import Order, OrderProduct def get_period_order_products( period: timedelta = timedelta(days=30), statuses: list[str] | None = None ) -> QuerySet[OrderProduct]: if statuses is None: statuses = ["FINISHED"] current = now() perioded = current - period orders = Order.objects.filter(status="FINISHED", buy_time__lte=current, buy_time__gte=perioded) return OrderProduct.objects.filter(status__in=statuses, order__in=orders) def get_revenue(clear: bool = True, period: timedelta = timedelta(days=30)) -> float: order_products = get_period_order_products(period) total: float = ( order_products.aggregate(total=Coalesce(Sum(F("buy_price") * F("quantity")), 0.0)).get("total") or 0.0 ) try: total = float(total) except (TypeError, ValueError): total = 0.0 if not clear: return round(float(total), 2) try: tax_rate = float(config.TAX_RATE or 0) except (TypeError, ValueError): tax_rate = 0.0 tax_included = False with suppress(Exception): tax_included = bool(getattr(config, "TAX_INCLUDED", False)) if tax_rate <= 0: net = total else: if tax_included: divisor = 1.0 + (tax_rate / 100.0) if divisor <= 0: net = total else: net = total / divisor else: net = total return round(float(net or 0.0), 2) def get_returns(period: timedelta = timedelta(days=30)) -> float: order_products = get_period_order_products(period, ["RETURNED"]) total_returns: float = ( order_products.aggregate(total=Coalesce(Sum(F("buy_price") * F("quantity")), 0.0)).get("total") or 0.0 ) try: return round(float(total_returns), 2) except (TypeError, ValueError): return 0.0 def get_total_processed_orders(period: timedelta = timedelta(days=30)) -> int: return get_period_order_products(period, ["RETURNED", "FINISHED"]).count() def get_daily_finished_orders_count(period: timedelta = timedelta(days=30)) -> dict[date, int]: current = now() period_start = current - period qs = ( Order.objects.filter(status="FINISHED", buy_time__lte=current, buy_time__gte=period_start) .annotate(day=TruncDate("buy_time")) .values("day") .annotate(cnt=Count("id")) .order_by("day") ) result: dict[date, int] = {} for row in qs: d = row.get("day") c = int(row.get("cnt", 0) or 0) if d: result[d] = c return result def get_daily_gross_revenue(period: timedelta = timedelta(days=30)) -> dict[date, float]: qs = ( get_period_order_products(period, ["FINISHED"]) # OrderProduct queryset .annotate(day=TruncDate("order__buy_time")) .values("day") .annotate(total=Coalesce(Sum(F("buy_price") * F("quantity")), 0.0)) .order_by("day") ) result: dict[date, float] = {} for row in qs: d = row.get("day") total = row.get("total") or 0.0 try: total_f = round(float(total), 2) except (TypeError, ValueError): total_f = 0.0 if d: result[d] = total_f return result