from collections import defaultdict from datetime import date, timedelta from decimal import Decimal from django.db.models import Count, Min, Max, Q from django.utils import timezone from rest_framework.exceptions import ValidationError from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from account.permissions import HasTabPermission, assert_shop_access from shop_management.models.data.inventory import StockMovement from shop_management.models.data.organization import Warehouse from shop_management.models.data.sale import Sale from shop_management.models.data.sale_return import SaleReturn ZERO = Decimal("0") def amount(value): return str(Decimal(value or 0).quantize(Decimal("0.01"))) def totals(): return {"sales": ZERO, "returns": ZERO, "cogs": ZERO, "returned_cogs": ZERO, "invoice_count": 0, "return_count": 0, "cost_complete": True} def presentation(values): net_sales = values["sales"] - values["returns"] net_cogs = values["cogs"] - values["returned_cogs"] profit = net_sales - net_cogs return { "sales": amount(values["sales"]), "returns": amount(values["returns"]), "net_sales": amount(net_sales), "net_cogs": amount(net_cogs) if values["cost_complete"] else None, "gross_profit": amount(profit) if values["cost_complete"] else None, "margin_percent": amount(profit / net_sales * 100) if net_sales > 0 and values["cost_complete"] else None, "invoice_count": values["invoice_count"], "return_count": values["return_count"], "cost_complete": values["cost_complete"], } class WarehouseProfitReportView(APIView): permission_classes = [IsAuthenticated, HasTabPermission] required_tab = "reports" def get(self, request): try: end = date.fromisoformat(request.query_params.get("end_date") or str(timezone.localdate())) all_dates = request.query_params.get("period") == "all" start = date.min if all_dates else date.fromisoformat(request.query_params.get("start_date") or str(end - timedelta(days=29))) shop_id = int(request.query_params.get("shop", "")) warehouse_id = int(request.query_params["warehouse"]) if request.query_params.get("warehouse") else None except (ValueError, TypeError): raise ValidationError({"filters": "Choose a branch and use valid warehouse IDs and YYYY-MM-DD dates."}) if start > end: raise ValidationError({"end_date": "End date cannot be before start date."}) shop = assert_shop_access(request.user, shop_id, request) warehouses = Warehouse.objects.filter(shop=shop) if warehouse_id is not None and not warehouses.filter(pk=warehouse_id).exists(): raise ValidationError({"warehouse": "Warehouse does not belong to this branch."}) history = Sale.objects.filter(shop=shop, uses_inventory_engine=True) if warehouse_id is not None: history = history.filter( Q(sale_items__warehouse_id=warehouse_id) | Q(sale_items__warehouse__isnull=True, warehouse_id=warehouse_id) ).distinct() activity = history.aggregate( posted=Count("id", filter=Q(document_status="POSTED")), cancelled=Count("id", filter=Q(document_status="VOID")), drafts=Count("id", filter=Q(document_status="DRAFT")), first_sale=Min("invoice_date", filter=Q(document_status="POSTED")), last_sale=Max("invoice_date", filter=Q(document_status="POSTED")), ) sales = Sale.objects.filter( shop=shop, uses_inventory_engine=True, document_status=Sale.DocumentStatus.POSTED, invoice_date__range=(start, end), ).select_related("warehouse", "currency").prefetch_related("sale_items__warehouse") returns = SaleReturn.objects.filter( shop=shop, uses_inventory_engine=True, document_status=SaleReturn.DocumentStatus.POSTED, sale__uses_inventory_engine=True, sale__document_status=Sale.DocumentStatus.POSTED, return_date__date__range=(start, end), ).select_related("sale__warehouse", "sale__currency").prefetch_related( "return_items__sale_item__warehouse", "return_items__batch_restorations" ) sales, returns = list(sales), list(returns) original_sales = {str(sale.pk): sale.currency_id for sale in sales} original_sales.update({str(row.sale_id): row.sale.currency_id for row in returns}) incomplete_costs = set() for movement in StockMovement.objects.filter(warehouse__shop=shop, reference_type="Sale", reference_id__in=original_sales, movement_type=StockMovement.MovementType.SALE): if movement.currency_id != original_sales[movement.reference_id]: incomplete_costs.add(movement.reference_id) grouped = defaultdict(totals) labels = {} for sale in sales: items = list(sale.sale_items.all()) subtotal = sum((Decimal(item.quantity or 0) * Decimal(item.sale_price or 0) for item in items), ZERO) touched = set() for item in items: item_warehouse = item.warehouse or sale.warehouse item_warehouse_id = item_warehouse.id if item_warehouse else None if warehouse_id is not None and item_warehouse_id != warehouse_id: continue key = (item_warehouse_id, sale.currency_id) labels[key] = (item_warehouse.name if item_warehouse else "Unassigned warehouse", sale.currency.title if sale.currency else "Unknown") row = grouped[key] row["cost_complete"] = row["cost_complete"] and str(sale.pk) not in incomplete_costs line_total = Decimal(item.quantity or 0) * Decimal(item.sale_price or 0) row["sales"] += (sale.calculated_total * line_total / subtotal) if subtotal else ZERO row["cogs"] += item.cogs_total touched.add(key) for key in touched: grouped[key]["invoice_count"] += 1 for returned in returns: sale = returned.sale touched = set() for item in returned.return_items.all(): sale_item = item.sale_item item_warehouse = sale_item.warehouse or sale.warehouse item_warehouse_id = item_warehouse.id if item_warehouse else None if warehouse_id is not None and item_warehouse_id != warehouse_id: continue key = (item_warehouse_id, sale.currency_id) labels[key] = (item_warehouse.name if item_warehouse else "Unassigned warehouse", sale.currency.title if sale.currency else "Unknown") row = grouped[key] row["cost_complete"] = row["cost_complete"] and str(sale.pk) not in incomplete_costs row["returns"] += item.refund_amount row["returned_cogs"] += sum((restoration.quantity * restoration.unit_cost for restoration in item.batch_restorations.all()), ZERO) touched.add(key) for key in touched: grouped[key]["return_count"] += 1 currency_totals = defaultdict(totals) result = [] for key, values in sorted(grouped.items(), key=lambda item: labels[item[0]]): warehouse, currency = labels[key] result.append({"warehouse_id": key[0], "warehouse": warehouse, "currency_id": key[1], "currency": currency, **presentation(values)}) for field, value in values.items(): if field == "cost_complete": currency_totals[key[1]][field] = currency_totals[key[1]][field] and value else: currency_totals[key[1]][field] += value currency_names = {key[1]: labels[key][1] for key in labels} return Response({ "shop": shop.name, "shop_id": shop.pk, "start_date": start, "end_date": end, "rows": result, "period": "all" if all_dates else "custom", "activity": activity, "summary_by_currency": [{"currency_id": currency_id, "currency": currency_names[currency_id], **presentation(values)} for currency_id, values in currency_totals.items()], "incomplete_costs": bool(incomplete_costs), "basis": "Posted inventory sales after discounts minus period returns and net saved cost of goods sold. Mixed-warehouse invoices are allocated by line value and each return follows its original line warehouse. Operating expenses and legacy sales are excluded. Currencies are not combined.", })