from datetime import date, timedelta from django.contrib.auth.decorators import login_required from django.contrib import messages from django.core.cache import cache from django.core.paginator import Paginator from django.db.models import Count, DecimalField, ExpressionWrapper, F, Q, Sum from django.db.models.functions import TruncMonth from django.http import JsonResponse from django.shortcuts import get_object_or_404, redirect, render from django.utils import timezone from core.decorators import capability_required, role_required from core.models import AuditLog from core.permissions import Capability from core.utils import export_csv from inventory.models import ( Product, ProductBatch, ProductStock, PurchaseOrder, PurchaseOrderItem, StockAdjustmentRequest, StockMovement, StockTransfer, UserStoreAssignment, ) from sales.models import Customer, Invoice, Sale, SaleItem from . import reports from .forms import ( AccountForm, ExpenseCategoryForm, ExpenseForm, OtherIncomeForm, OwnerDrawForm, ) from .models import ( Account, AccountTransaction, Expense, ExpenseCategory, OtherIncome, OwnerDraw, ) @capability_required(Capability.FINANCE_READ) def account_list(request): accounts = Account.objects.filter(is_active=True) total_balance = sum(a.balance for a in accounts) return render( request, "finance/account_list.html", { "accounts": accounts, "total_balance": total_balance, }, ) @capability_required(Capability.FINANCE_CREATE) def account_create(request): if request.method == "POST": form = AccountForm(request.POST) if form.is_valid(): form.save() messages.success(request, "Account created.") return redirect("finance:account_list") else: form = AccountForm() return render(request, "finance/account_form.html", {"form": form}) @capability_required(Capability.FINANCE_READ) def account_detail(request, pk): account = get_object_or_404(Account, pk=pk) transactions = account.transactions.order_by("-date", "-created_at")[:50] return render( request, "finance/account_detail.html", { "account": account, "transactions": transactions, }, ) @capability_required(Capability.FINANCE_READ) def category_list(request): categories = ExpenseCategory.objects.all() return render(request, "finance/category_list.html", {"categories": categories}) @capability_required(Capability.FINANCE_CREATE) def category_create(request): if request.method == "POST": form = ExpenseCategoryForm(request.POST) if form.is_valid(): form.save() messages.success(request, "Category created.") return redirect("finance:category_list") else: form = ExpenseCategoryForm() return render(request, "finance/category_form.html", {"form": form}) @capability_required(Capability.FINANCE_READ) def expense_list(request): expenses = Expense.objects.select_related("category", "account").order_by( "-date", "-created_at" ) query = request.GET.get("q") if query: expenses = expenses.filter( Q(description__icontains=query) | Q(reference__icontains=query) ) category_id = request.GET.get("category") if category_id: expenses = expenses.filter(category_id=category_id) start = request.GET.get("start") end = request.GET.get("end") if start and end: expenses = expenses.filter(date__range=(start, end)) paginator = Paginator(expenses, 20) page = request.GET.get("page") expenses = paginator.get_page(page) categories = ExpenseCategory.objects.all() return render( request, "finance/expense_list.html", { "expenses": expenses, "categories": categories, "query": query or "", "current_category": category_id or "", }, ) @capability_required(Capability.FINANCE_CREATE) def expense_create(request): if request.method == "POST": form = ExpenseForm(request.POST) if form.is_valid(): expense = form.save(commit=False) expense.created_by = request.user expense.save() expense.account.withdraw(expense.amount) AccountTransaction.objects.create( account=expense.account, tx_type=AccountTransaction.TxType.EXPENSE, direction="out", amount=expense.amount, description=expense.description, reference=expense.reference, date=expense.date, ) messages.success(request, "Expense recorded.") return redirect("finance:expense_list") else: form = ExpenseForm(initial={"date": timezone.now().date()}) return render(request, "finance/expense_form.html", {"form": form}) @capability_required(Capability.FINANCE_READ) def expense_detail(request, pk): expense = get_object_or_404(Expense, pk=pk) return render(request, "finance/expense_detail.html", {"expense": expense}) @capability_required(Capability.FINANCE_READ) def income_list(request): incomes = OtherIncome.objects.select_related("account").order_by("-date", "-created_at") return render(request, "finance/income_list.html", {"incomes": incomes}) @capability_required(Capability.FINANCE_CREATE) def income_create(request): if request.method == "POST": form = OtherIncomeForm(request.POST) if form.is_valid(): income = form.save(commit=False) income.created_by = request.user income.save() income.account.deposit(income.amount) AccountTransaction.objects.create( account=income.account, tx_type=AccountTransaction.TxType.INCOME, direction="in", amount=income.amount, description=income.description, reference=income.reference, date=income.date, ) messages.success(request, "Income recorded.") return redirect("finance:income_list") else: form = OtherIncomeForm(initial={"date": timezone.now().date()}) return render(request, "finance/income_form.html", {"form": form}) @capability_required(Capability.FINANCE_READ) def draw_list(request): draws = OwnerDraw.objects.select_related("account").order_by("-date", "-created_at") return render(request, "finance/draw_list.html", {"draws": draws}) @role_required("owner", "super_admin") def draw_create(request): if request.method == "POST": form = OwnerDrawForm(request.POST) if form.is_valid(): draw = form.save(commit=False) draw.created_by = request.user draw.save() draw.account.withdraw(draw.amount) AccountTransaction.objects.create( account=draw.account, tx_type=AccountTransaction.TxType.DRAW, direction="out", amount=draw.amount, description=draw.description or "Owner Draw", date=draw.date, ) messages.success(request, "Owner draw recorded.") return redirect("finance:draw_list") else: form = OwnerDrawForm(initial={"date": timezone.now().date()}) return render(request, "finance/draw_form.html", {"form": form}) def _parse_dashboard_period(request): today = timezone.now().date() start_date = today.replace(day=1) end_date = today if request.GET.get("start") and request.GET.get("end"): try: from datetime import datetime start_date = datetime.strptime(request.GET["start"], "%Y-%m-%d").date() end_date = datetime.strptime(request.GET["end"], "%Y-%m-%d").date() except ValueError: pass return start_date, end_date, today def _revenue_expression(): return ExpressionWrapper( F("quantity") * F("unit_price"), output_field=DecimalField(max_digits=20, decimal_places=2), ) def _stock_value_expression(): return ExpressionWrapper( F("stock_quantity") * F("cost_price"), output_field=DecimalField(max_digits=20, decimal_places=2), ) def _pending_supplier_payments(limit=None): purchase_orders = ( PurchaseOrder.objects.filter(status=PurchaseOrder.Status.RECEIVED) .select_related("supplier") .order_by("due_date", "date") ) due_orders = [po for po in purchase_orders if po.balance_due > 0] return due_orders[:limit] if limit else due_orders def _common_notifications(today): expiry_cutoff = today + timedelta(days=30) return { "low_stock_count": Product.objects.filter( is_active=True, stock_quantity__lte=F("low_stock_alert"), ).count(), "overstock_count": Product.objects.filter( is_active=True, overstock_alert__gt=0, stock_quantity__gte=F("overstock_alert"), ).count(), "expiry_alert_count": ProductBatch.objects.filter( is_active=True, quantity__gt=0, expiry_date__isnull=False, expiry_date__lte=expiry_cutoff, ).count(), "pending_transfer_count": StockTransfer.objects.filter( status=StockTransfer.Status.REQUESTED ).count(), "pending_adjustment_count": StockAdjustmentRequest.objects.filter( status=StockAdjustmentRequest.Status.PENDING ).count(), "pending_supplier_payment_count": len(_pending_supplier_payments()), } def _monthly_sales_vs_purchases(months=6): today = timezone.now().date() start_date = (today.replace(day=1) - timedelta(days=32 * (months - 1))).replace(day=1) revenue_expr = _revenue_expression() purchase_expr = ExpressionWrapper( F("quantity") * F("unit_cost"), output_field=DecimalField(max_digits=20, decimal_places=2), ) sales_rows = ( SaleItem.objects.filter( sale__status=Sale.Status.COMPLETED, sale__date__gte=start_date, ) .annotate(month=TruncMonth("sale__date")) .values("month") .annotate(total=Sum(revenue_expr)) ) purchase_rows = ( PurchaseOrderItem.objects.filter( purchase_order__status=PurchaseOrder.Status.RECEIVED, purchase_order__date__gte=start_date, ) .annotate(month=TruncMonth("purchase_order__date")) .values("month") .annotate(total=Sum(purchase_expr)) ) sales_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in sales_rows} purchase_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in purchase_rows} labels = [] current = start_date while current <= today: labels.append(current.strftime("%Y-%m")) if current.month == 12: current = current.replace(year=current.year + 1, month=1) else: current = current.replace(month=current.month + 1) return [ { "month": month, "sales": sales_map.get(month, 0), "purchases": purchase_map.get(month, 0), } for month in labels ] def _stock_movement_trend(days=30): today = timezone.now().date() start_date = today - timedelta(days=days - 1) in_types = [ StockMovement.MovementType.STOCK_IN, StockMovement.MovementType.TRANSFER_IN, StockMovement.MovementType.RETURN, ] out_types = [ StockMovement.MovementType.SALE, StockMovement.MovementType.STOCK_OUT, StockMovement.MovementType.TRANSFER_OUT, StockMovement.MovementType.DAMAGED, StockMovement.MovementType.LOST, StockMovement.MovementType.EXPIRED, ] in_rows = ( StockMovement.objects.filter(date__gte=start_date, movement_type__in=in_types) .values("date") .annotate(total=Sum("quantity")) ) out_rows = ( StockMovement.objects.filter(date__gte=start_date, movement_type__in=out_types) .values("date") .annotate(total=Sum("quantity")) ) in_map = {row["date"]: row["total"] or 0 for row in in_rows} out_map = {row["date"]: row["total"] or 0 for row in out_rows} chart = [] day = start_date while day <= today: chart.append( { "day": day, "stock_in": in_map.get(day, 0), "stock_out": out_map.get(day, 0), } ) day += timedelta(days=1) return chart def _to_float(value): return float(value or 0) def _dashboard_payload_super_admin(start_date, end_date, today): summary = reports.get_summary(start_date, end_date) top_items = reports.get_top_products(start_date, end_date, limit=10) expense_breakdown = reports.get_expenses_by_category(start_date, end_date) return { "summary": { "revenue": _to_float(summary["revenue"]), "cogs": _to_float(summary["cogs"]), "gross_profit": _to_float(summary["gross_profit"]), "net_profit": _to_float(summary["net_profit"]), "total_expenses": _to_float(summary["total_expenses"]), "sales_count": int(summary["sales_count"]), }, "monthly_sales_vs_purchases": [ { "month": row["month"], "sales": _to_float(row["sales"]), "purchases": _to_float(row["purchases"]), } for row in _monthly_sales_vs_purchases(6) ], "stock_movement_trend": [ { "day": row["day"].isoformat(), "stock_in": _to_float(row["stock_in"]), "stock_out": _to_float(row["stock_out"]), } for row in _stock_movement_trend(30) ], "top_items": [ { "name": row["product__name"], "quantity": _to_float(row["total_qty"]), "revenue": _to_float(row["total_revenue"]), } for row in top_items ], "expense_breakdown": [ { "category": row["category__name"] or "Uncategorized", "amount": _to_float(row["total"]), } for row in expense_breakdown ], "notifications": _common_notifications(today), } def _dashboard_payload_store_manager(start_date, end_date, today): sales_window_start = today - timedelta(days=30) stock_by_category = ( Product.objects.filter(is_active=True) .values("category__name") .annotate( total_items=Count("id"), total_qty=Sum("stock_quantity"), total_value=Sum(_stock_value_expression()), ) .order_by("-total_value") ) fast_moving = ( SaleItem.objects.filter( sale__status=Sale.Status.COMPLETED, sale__date__gte=sales_window_start, ) .values("product__name") .annotate(total_qty=Sum("quantity")) .order_by("-total_qty")[:10] ) slow_moving = ( SaleItem.objects.filter( sale__status=Sale.Status.COMPLETED, sale__date__gte=sales_window_start, ) .values("product__name") .annotate(total_qty=Sum("quantity")) .order_by("total_qty")[:10] ) adjustments = ( StockAdjustmentRequest.objects.filter(created_at__date__range=(start_date, end_date)) .values("reason") .annotate(total=Count("id")) .order_by("-total") ) return { "stock_by_category": [ { "category": row["category__name"] or "Uncategorized", "items": int(row["total_items"] or 0), "quantity": _to_float(row["total_qty"]), "value": _to_float(row["total_value"]), } for row in stock_by_category ], "fast_moving_items": [ {"name": row["product__name"], "quantity": _to_float(row["total_qty"])} for row in fast_moving ], "slow_moving_items": [ {"name": row["product__name"], "quantity": _to_float(row["total_qty"])} for row in slow_moving ], "adjustments": [ {"reason": row["reason"], "total": int(row["total"] or 0)} for row in adjustments ], "low_stock_count": Product.objects.filter( is_active=True, stock_quantity__lte=F("low_stock_alert"), ).count(), "out_of_stock_count": Product.objects.filter( is_active=True, stock_quantity__lte=0, ).count(), "notifications": _common_notifications(today), } def _dashboard_payload_storekeeper(user, today): assignment = ( UserStoreAssignment.objects.select_related("location") .filter(user=user, is_active=True) .first() ) location = assignment.location if assignment else None stocks = ProductStock.objects.select_related("product", "location") movements = StockMovement.objects.select_related("product", "location") transfer_requests = StockTransfer.objects.filter(status=StockTransfer.Status.REQUESTED) if location: stocks = stocks.filter(location=location) movements = movements.filter(location=location) transfer_requests = transfer_requests.filter(from_location=location) low_stock = stocks.filter(quantity__lte=F("product__low_stock_alert")).order_by("quantity")[:10] return { "location": location.name if location else "", "counts": { "items_in_store": stocks.filter(quantity__gt=0).count(), "today_received": movements.filter( date=today, movement_type=StockMovement.MovementType.STOCK_IN ).count(), "today_issued": movements.filter( date=today, movement_type__in=[ StockMovement.MovementType.SALE, StockMovement.MovementType.STOCK_OUT, StockMovement.MovementType.TRANSFER_OUT, ], ).count(), "pending_requests": transfer_requests.count(), }, "low_stock_items": [ { "name": row.product.name, "quantity": _to_float(row.quantity), "unit": row.product.unit, } for row in low_stock ], "notifications": _common_notifications(today), } def _dashboard_payload_accountant(start_date, end_date, today): summary = reports.get_summary(start_date, end_date) received_purchase_orders = PurchaseOrder.objects.filter( status=PurchaseOrder.Status.RECEIVED ).select_related("supplier") total_purchases = sum( po.total for po in received_purchase_orders if start_date <= po.date <= end_date ) supplier_balances = {} for po in received_purchase_orders: if po.balance_due <= 0: continue supplier_name = po.supplier.name supplier_balances[supplier_name] = supplier_balances.get(supplier_name, 0) + po.balance_due supplier_balance_rows = sorted( ( {"supplier": supplier, "balance": balance} for supplier, balance in supplier_balances.items() ), key=lambda item: item["balance"], reverse=True, )[:10] customer_balance_rows = sorted( ( {"customer": customer.name, "balance": customer.balance_due} for customer in Customer.objects.filter(is_active=True) if customer.balance_due > 0 ), key=lambda item: item["balance"], reverse=True, )[:10] month_start = today.replace(day=1) months_back = (month_start - timedelta(days=160)).replace(day=1) revenue_expr = _revenue_expression() revenue_rows = ( SaleItem.objects.filter( sale__status=Sale.Status.COMPLETED, sale__date__gte=months_back, ) .annotate(month=TruncMonth("sale__date")) .values("month") .annotate(total=Sum(revenue_expr)) .order_by("month") ) expense_rows = ( Expense.objects.filter(date__gte=months_back) .annotate(month=TruncMonth("date")) .values("month") .annotate(total=Sum("amount")) .order_by("month") ) revenue_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in revenue_rows} expense_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in expense_rows} monthly_revenue_expense = [ { "month": month, "revenue": _to_float(revenue_map.get(month, 0)), "expense": _to_float(expense_map.get(month, 0)), } for month in sorted(set(revenue_map.keys()) | set(expense_map.keys())) ] cash_in_rows = ( AccountTransaction.objects.filter(date__gte=months_back, direction="in") .annotate(month=TruncMonth("date")) .values("month") .annotate(total=Sum("amount")) ) cash_out_rows = ( AccountTransaction.objects.filter(date__gte=months_back, direction="out") .annotate(month=TruncMonth("date")) .values("month") .annotate(total=Sum("amount")) ) cash_in_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in cash_in_rows} cash_out_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in cash_out_rows} cash_flow_trend = [ { "month": month, "cash_in": _to_float(cash_in_map.get(month, 0)), "cash_out": _to_float(cash_out_map.get(month, 0)), } for month in sorted(set(cash_in_map.keys()) | set(cash_out_map.keys())) ] return { "summary": { "revenue": _to_float(summary["revenue"]), "cogs": _to_float(summary["cogs"]), "gross_profit": _to_float(summary["gross_profit"]), "net_profit": _to_float(summary["net_profit"]), "total_expenses": _to_float(summary["total_expenses"]), "total_other_income": _to_float(summary["total_other_income"]), "sales_count": int(summary["sales_count"]), "total_purchases": _to_float(total_purchases), }, "supplier_balances": [ {"supplier": row["supplier"], "balance": _to_float(row["balance"])} for row in supplier_balance_rows ], "customer_balances": [ {"customer": row["customer"], "balance": _to_float(row["balance"])} for row in customer_balance_rows ], "monthly_revenue_expense": monthly_revenue_expense, "cash_flow_trend": cash_flow_trend, "notifications": _common_notifications(today), } def _dashboard_payload_sales_officer(user, today): revenue_expr = _revenue_expression() top_selling_today = ( SaleItem.objects.filter( sale__created_by=user, sale__status=Sale.Status.COMPLETED, sale__date=today, ) .values("product__name") .annotate(total_qty=Sum("quantity"), total_revenue=Sum(revenue_expr)) .order_by("-total_qty")[:10] ) recent_sales = Sale.objects.filter(created_by=user).order_by("-date", "-created_at")[:10] return { "top_selling_today": [ { "name": row["product__name"], "quantity": _to_float(row["total_qty"]), "revenue": _to_float(row["total_revenue"]), } for row in top_selling_today ], "recent_sales": [ { "date": sale.date.isoformat(), "reference": sale.reference, "total": _to_float(sale.total), "status": sale.status, } for sale in recent_sales ], "notifications": _common_notifications(today), } def _dashboard_payload_auditor(start_date, end_date, today): summary = reports.get_summary(start_date, end_date) movement_types = ( StockMovement.objects.values("movement_type") .annotate(total=Count("id")) .order_by("-total") ) actions = AuditLog.objects.values("action").annotate(total=Count("id")).order_by("-total") adjustment_status = ( StockAdjustmentRequest.objects.values("status") .annotate(total=Count("id")) .order_by("-total") ) return { "summary": { "revenue": _to_float(summary["revenue"]), "cogs": _to_float(summary["cogs"]), "gross_profit": _to_float(summary["gross_profit"]), "net_profit": _to_float(summary["net_profit"]), "total_expenses": _to_float(summary["total_expenses"]), }, "stock_movement_types": [ {"type": row["movement_type"], "count": int(row["total"] or 0)} for row in movement_types ], "audit_action_counts": [ {"action": row["action"], "count": int(row["total"] or 0)} for row in actions ], "adjustment_status_counts": [ {"status": row["status"], "count": int(row["total"] or 0)} for row in adjustment_status ], "notifications": _common_notifications(today), } @login_required def dashboard_data(request): start_date, end_date, today = _parse_dashboard_period(request) role = request.user.effective_role role_to_builder = { "super_admin": lambda: _dashboard_payload_super_admin(start_date, end_date, today), "store_manager": lambda: _dashboard_payload_store_manager(start_date, end_date, today), "storekeeper": lambda: _dashboard_payload_storekeeper(request.user, today), "accountant": lambda: _dashboard_payload_accountant(start_date, end_date, today), "sales_officer": lambda: _dashboard_payload_sales_officer(request.user, today), "auditor": lambda: _dashboard_payload_auditor(start_date, end_date, today), } if role not in role_to_builder: return JsonResponse({"detail": "Role has no dashboard data endpoint."}, status=403) cache_key = ( f"finance-dashboard-data:v1:{role}:{request.user.pk}:" f"{start_date.isoformat()}:{end_date.isoformat()}" ) payload = cache.get(cache_key) if payload is None: payload = { "role": role, "period": { "start": start_date.isoformat(), "end": end_date.isoformat(), }, "generated_at": timezone.now().isoformat(), "data": role_to_builder[role](), } cache.set(cache_key, payload, 300) return JsonResponse(payload) @login_required def dashboard(request): role_to_dashboard = { "super_admin": "finance:dashboard_super_admin", "store_manager": "finance:dashboard_store_manager", "storekeeper": "finance:dashboard_storekeeper", "accountant": "finance:dashboard_accountant", "sales_officer": "finance:dashboard_sales_officer", "auditor": "finance:dashboard_auditor", } target = role_to_dashboard.get(request.user.effective_role, "finance:dashboard_super_admin") return redirect(target) @role_required("super_admin", "owner") def dashboard_super_admin(request): start_date, end_date, today = _parse_dashboard_period(request) summary = reports.get_summary(start_date, end_date) total_stock_value = ( Product.objects.filter(is_active=True) .aggregate(total=Sum(_stock_value_expression())) .get("total") or 0 ) total_items = Product.objects.filter(is_active=True).count() total_purchases = sum( po.total for po in PurchaseOrder.objects.filter( status=PurchaseOrder.Status.RECEIVED, date__range=(start_date, end_date), ).prefetch_related("items") ) pending_approvals = ( StockTransfer.objects.filter(status=StockTransfer.Status.REQUESTED).count() + StockAdjustmentRequest.objects.filter( status=StockAdjustmentRequest.Status.PENDING ).count() ) low_stock_products = Product.objects.filter( is_active=True, stock_quantity__lte=F("low_stock_alert"), ).order_by("stock_quantity")[:10] context = { **summary, "start_date": start_date, "end_date": end_date, "today": today, "total_stock_value": total_stock_value, "total_items": total_items, "total_purchases": total_purchases, "pending_approvals": pending_approvals, "low_stock_products": low_stock_products, "monthly_sales_vs_purchases": _monthly_sales_vs_purchases(6), "stock_movement_trend": _stock_movement_trend(30), "top_items": reports.get_top_products(start_date, end_date, limit=10), "expense_breakdown": reports.get_expenses_by_category(start_date, end_date), "recent_activities": AuditLog.objects.select_related("actor").all()[:12], "notifications": _common_notifications(today), } return render(request, "finance/dashboards/super_admin.html", context) @role_required("store_manager", "manager", "super_admin", "owner") def dashboard_store_manager(request): start_date, end_date, today = _parse_dashboard_period(request) sales_window_start = today - timedelta(days=30) stock_by_category = ( Product.objects.filter(is_active=True) .values("category__name") .annotate( total_items=Count("id"), total_qty=Sum("stock_quantity"), total_value=Sum(_stock_value_expression()), ) .order_by("-total_value") ) fast_moving = ( SaleItem.objects.filter( sale__status=Sale.Status.COMPLETED, sale__date__gte=sales_window_start, ) .values("product__name") .annotate(total_qty=Sum("quantity")) .order_by("-total_qty")[:10] ) slow_moving = ( SaleItem.objects.filter( sale__status=Sale.Status.COMPLETED, sale__date__gte=sales_window_start, ) .values("product__name") .annotate(total_qty=Sum("quantity")) .order_by("total_qty")[:10] ) stock_adjustments_month = ( StockAdjustmentRequest.objects.filter(created_at__date__range=(start_date, end_date)) .values("reason") .annotate(total=Count("id")) .order_by("-total") ) context = { "start_date": start_date, "end_date": end_date, "today": today, "current_stock_levels": Product.objects.filter(is_active=True).order_by("-stock_quantity")[:20], "low_stock_items": Product.objects.filter( is_active=True, stock_quantity__lte=F("low_stock_alert"), ).order_by("stock_quantity")[:10], "out_of_stock_items": Product.objects.filter( is_active=True, stock_quantity__lte=0 ).order_by("name")[:10], "pending_transfer_requests": StockTransfer.objects.filter( status=StockTransfer.Status.REQUESTED ).select_related("from_location", "to_location", "requested_by")[:10], "recent_received_goods": PurchaseOrder.objects.filter( status=PurchaseOrder.Status.RECEIVED ).select_related("supplier").order_by("-received_date", "-date")[:10], "stock_by_category": stock_by_category, "fast_moving_items": fast_moving, "slow_moving_items": slow_moving, "stock_adjustments_month": stock_adjustments_month, "recent_activities": StockMovement.objects.select_related( "product", "location", "created_by" ).order_by("-created_at")[:12], "notifications": _common_notifications(today), } return render(request, "finance/dashboards/store_manager.html", context) @role_required("storekeeper", "store_manager", "manager", "super_admin", "owner") def dashboard_storekeeper(request): _, _, today = _parse_dashboard_period(request) assignment = ( UserStoreAssignment.objects.select_related("location") .filter(user=request.user, is_active=True) .first() ) location = assignment.location if assignment else None stocks = ProductStock.objects.select_related("product", "location") movements = StockMovement.objects.select_related("product", "location", "created_by") transfer_requests = StockTransfer.objects.select_related( "from_location", "to_location", "requested_by" ).filter(status=StockTransfer.Status.REQUESTED) pending_adjustments = StockAdjustmentRequest.objects.filter( status=StockAdjustmentRequest.Status.PENDING ) if location: stocks = stocks.filter(location=location) movements = movements.filter(location=location) transfer_requests = transfer_requests.filter(from_location=location) pending_adjustments = pending_adjustments.filter(location=location) context = { "today": today, "assigned_location": location, "items_in_store": stocks.filter(quantity__gt=0).order_by("-quantity")[:25], "today_goods_received": movements.filter( date=today, movement_type=StockMovement.MovementType.STOCK_IN ).order_by("-created_at")[:15], "today_issued_items": movements.filter( date=today, movement_type__in=[ StockMovement.MovementType.SALE, StockMovement.MovementType.STOCK_OUT, StockMovement.MovementType.TRANSFER_OUT, ], ).order_by("-created_at")[:15], "pending_issue_requests": transfer_requests[:10], "pending_adjustment_requests": pending_adjustments.select_related( "product", "requested_by" )[:10], "low_stock_alerts": stocks.filter( quantity__lte=F("product__low_stock_alert") ).order_by("quantity")[:10], "recent_activities": movements.order_by("-created_at")[:12], "notifications": _common_notifications(today), } return render(request, "finance/dashboards/storekeeper.html", context) @role_required("accountant", "super_admin", "owner") def dashboard_accountant(request): start_date, end_date, today = _parse_dashboard_period(request) summary = reports.get_summary(start_date, end_date) received_purchase_orders = PurchaseOrder.objects.filter( status=PurchaseOrder.Status.RECEIVED ).select_related("supplier") total_purchases = sum( po.total for po in received_purchase_orders if start_date <= po.date <= end_date ) supplier_balances = {} for po in received_purchase_orders: if po.balance_due <= 0: continue supplier_name = po.supplier.name supplier_balances[supplier_name] = supplier_balances.get(supplier_name, 0) + po.balance_due supplier_balance_rows = sorted( ( {"supplier": supplier, "balance": balance} for supplier, balance in supplier_balances.items() ), key=lambda item: item["balance"], reverse=True, )[:10] total_supplier_balances = sum(row["balance"] for row in supplier_balance_rows) customer_balance_rows = sorted( ( {"customer": customer, "balance": customer.balance_due} for customer in Customer.objects.filter(is_active=True) if customer.balance_due > 0 ), key=lambda item: item["balance"], reverse=True, )[:10] total_customer_balances = sum(row["balance"] for row in customer_balance_rows) month_start = today.replace(day=1) months_back = (month_start - timedelta(days=160)).replace(day=1) revenue_expr = _revenue_expression() revenue_rows = ( SaleItem.objects.filter( sale__status=Sale.Status.COMPLETED, sale__date__gte=months_back, ) .annotate(month=TruncMonth("sale__date")) .values("month") .annotate(total=Sum(revenue_expr)) .order_by("month") ) expense_rows = ( Expense.objects.filter(date__gte=months_back) .annotate(month=TruncMonth("date")) .values("month") .annotate(total=Sum("amount")) .order_by("month") ) revenue_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in revenue_rows} expense_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in expense_rows} monthly_revenue_expense = [ {"month": month, "revenue": revenue_map.get(month, 0), "expense": expense_map.get(month, 0)} for month in sorted(set(revenue_map.keys()) | set(expense_map.keys())) ] cash_in_rows = ( AccountTransaction.objects.filter(date__gte=months_back, direction="in") .annotate(month=TruncMonth("date")) .values("month") .annotate(total=Sum("amount")) ) cash_out_rows = ( AccountTransaction.objects.filter(date__gte=months_back, direction="out") .annotate(month=TruncMonth("date")) .values("month") .annotate(total=Sum("amount")) ) cash_in_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in cash_in_rows} cash_out_map = {row["month"].strftime("%Y-%m"): row["total"] or 0 for row in cash_out_rows} cash_flow_trend = [ {"month": month, "cash_in": cash_in_map.get(month, 0), "cash_out": cash_out_map.get(month, 0)} for month in sorted(set(cash_in_map.keys()) | set(cash_out_map.keys())) ] context = { **summary, "start_date": start_date, "end_date": end_date, "today": today, "total_purchases": total_purchases, "supplier_balances": supplier_balance_rows, "customer_balances": customer_balance_rows, "monthly_profit": summary["net_profit"], "outstanding_payables": total_supplier_balances, "outstanding_receivables": total_customer_balances, "monthly_revenue_expense": monthly_revenue_expense, "cash_flow_trend": cash_flow_trend, "recent_financial_activity": AccountTransaction.objects.select_related("account") .order_by("-created_at")[:12], "notifications": _common_notifications(today), } return render(request, "finance/dashboards/accountant.html", context) @role_required("sales_officer", "cashier", "store_manager", "manager", "super_admin", "owner") def dashboard_sales_officer(request): _, _, today = _parse_dashboard_period(request) month_start = today.replace(day=1) revenue_expr = _revenue_expression() own_sales_today = Sale.objects.filter( created_by=request.user, status=Sale.Status.COMPLETED, date=today, ) own_sales_month = Sale.objects.filter( created_by=request.user, status=Sale.Status.COMPLETED, date__range=(month_start, today), ) own_invoices_today = Invoice.objects.filter(created_by=request.user, issue_date=today) top_selling_today = ( SaleItem.objects.filter( sale__created_by=request.user, sale__status=Sale.Status.COMPLETED, sale__date=today, ) .values("product__name") .annotate(total_qty=Sum("quantity"), total_revenue=Sum(revenue_expr)) .order_by("-total_qty")[:10] ) context = { "today": today, "today_sales_total": sum(sale.total for sale in own_sales_today), "today_sales_count": own_sales_today.count(), "today_invoices_count": own_invoices_today.count(), "monthly_sales_total": sum(sale.total for sale in own_sales_month), "monthly_sales_count": own_sales_month.count(), "top_selling_today": top_selling_today, "recent_sales": Sale.objects.filter(created_by=request.user).order_by("-date", "-created_at")[:10], "recent_invoices": Invoice.objects.filter(created_by=request.user).order_by( "-issue_date", "-created_at" )[:10], "notifications": _common_notifications(today), } return render(request, "finance/dashboards/sales_officer.html", context) @role_required("auditor", "super_admin", "owner") def dashboard_auditor(request): start_date, end_date, today = _parse_dashboard_period(request) summary = reports.get_summary(start_date, end_date) context = { **summary, "start_date": start_date, "end_date": end_date, "today": today, "stock_movement_logs": StockMovement.objects.select_related( "product", "location", "created_by" ).order_by("-created_at")[:20], "voided_deleted_transactions": AuditLog.objects.select_related("actor").filter( action__in=[ AuditLog.Action.DELETE, AuditLog.Action.VOID, AuditLog.Action.REVERSE, ] )[:20], "adjustment_history": StockAdjustmentRequest.objects.select_related( "product", "location", "requested_by", "approved_by" ).order_by("-created_at")[:15], "user_activity_logs": AuditLog.objects.select_related("actor").order_by("-created_at")[:20], "notifications": _common_notifications(today), } return render(request, "finance/dashboards/auditor.html", context) @capability_required(Capability.REPORTS_VIEW) def profit_loss(request): today = date.today() start_date = today.replace(day=1) end_date = today if request.GET.get("start") and request.GET.get("end"): try: from datetime import datetime start_date = datetime.strptime(request.GET["start"], "%Y-%m-%d").date() end_date = datetime.strptime(request.GET["end"], "%Y-%m-%d").date() except ValueError: pass summary = reports.get_summary(start_date, end_date) expenses_by_category = reports.get_expenses_by_category(start_date, end_date) return render( request, "finance/profit_loss.html", { **summary, "expenses_by_category": expenses_by_category, }, ) @capability_required(Capability.REPORTS_EXPORT) def export_expenses_csv(request): expenses = Expense.objects.select_related("category", "account", "created_by").order_by( "-date" ) start = request.GET.get("start") end = request.GET.get("end") if start and end: expenses = expenses.filter(date__range=(start, end)) category_id = request.GET.get("category") if category_id: expenses = expenses.filter(category_id=category_id) headers = [ "Date", "Description", "Category", "Account", "Amount", "Reference", "Recorded By", ] rows = [] for expense in expenses: rows.append( [ expense.date.strftime("%d/%m/%Y"), expense.description, expense.category.name, expense.account.name, expense.amount, expense.reference or "", expense.created_by.username if expense.created_by else "", ] ) return export_csv("expenses", headers, rows) @capability_required(Capability.REPORTS_EXPORT) def export_profit_loss_csv(request): from datetime import datetime today = date.today() start_date = today.replace(day=1) end_date = today if request.GET.get("start") and request.GET.get("end"): try: start_date = datetime.strptime(request.GET["start"], "%Y-%m-%d").date() end_date = datetime.strptime(request.GET["end"], "%Y-%m-%d").date() except ValueError: pass summary = reports.get_summary(start_date, end_date) expenses_by_cat = reports.get_expenses_by_category(start_date, end_date) headers = ["Item", "Amount"] rows = [ ["Period", f"{start_date.strftime('%d/%m/%Y')} - {end_date.strftime('%d/%m/%Y')}"], ["", ""], ["INCOME", ""], ["Sales Revenue", summary["revenue"]], ["Other Income", summary["total_other_income"]], ["Total Income", summary["revenue"] + summary["total_other_income"]], ["", ""], ["COST OF GOODS SOLD", ""], ["COGS", summary["cogs"]], ["", ""], ["GROSS PROFIT", summary["gross_profit"]], ["", ""], ["EXPENSES", ""], ] for cat in expenses_by_cat: rows.append([cat["category__name"], cat["total"]]) rows += [ ["Total Expenses", summary["total_expenses"]], ["", ""], ["NET PROFIT", summary["net_profit"]], ["", ""], ["OWNER DRAWS", summary["total_draws"]], ] return export_csv("profit-loss", headers, rows)