from django.contrib import messages from django.core.exceptions import PermissionDenied from django.core.paginator import Paginator from django.db import transaction from django.db.models import Q from django.shortcuts import get_object_or_404, redirect, render from django.utils import timezone from business.models import Business from core.decorators import capability_required from core.permissions import Capability from core.utils import export_csv from finance.models import AccountTransaction from .forms import ( CustomerForm, InvoiceForm, InvoiceItemFormSet, InvoicePaymentForm, SaleForm, SaleItemFormSet, SalesReturnForm, SalesReturnItemFormSet, ) from .models import ( Customer, CustomerLedgerEntry, Invoice, InvoicePayment, PaymentAllocation, Sale, SaleItem, SalesReturn, ) from .utils import process_sale, record_customer_ledger_entry def generate_reference(): """Generate next sale reference e.g. SALE-0042.""" last = Sale.objects.order_by("-id").first() next_id = (last.id + 1) if last else 1 return f"SALE-{next_id:04d}" def generate_invoice_reference(): last = Invoice.objects.order_by("-id").first() next_id = (last.id + 1) if last else 1 return f"INV-{next_id:04d}" def _can_read_all_sales(user): return user.has_capability(Capability.SALES_READ_ALL) def _can_read_all_invoices(user): return user.has_capability(Capability.INVOICE_READ_ALL) def _ensure_sale_visible(request, sale): if _can_read_all_sales(request.user): return if sale.created_by_id != request.user.id: raise PermissionDenied def _ensure_invoice_visible(request, invoice): if _can_read_all_invoices(request.user): return if invoice.created_by_id != request.user.id: raise PermissionDenied @capability_required(Capability.SALES_READ_ALL, Capability.SALES_READ_OWN, any_of=True) def sale_list(request): sales = Sale.objects.prefetch_related("items").order_by("-date", "-created_at") if not _can_read_all_sales(request.user): sales = sales.filter(created_by=request.user) query = request.GET.get("q") if query: sales = sales.filter( Q(reference__icontains=query) | Q(customer_name__icontains=query) | Q(customer__name__icontains=query) ) status = request.GET.get("status") if status: sales = sales.filter(status=status) start = request.GET.get("start") end = request.GET.get("end") if start and end: sales = sales.filter(date__range=(start, end)) paginator = Paginator(sales, 20) page = request.GET.get("page") sales = paginator.get_page(page) return render( request, "sales/sale_list.html", { "sales": sales, "query": query or "", "current_status": status or "", }, ) @capability_required(Capability.SALES_CREATE) def sale_create(request): if request.method == "POST": form = SaleForm(request.POST) formset = SaleItemFormSet(request.POST) if form.is_valid() and formset.is_valid(): sale = form.save(commit=False) sale.reference = generate_reference() if sale.customer and not sale.customer_name: sale.customer_name = sale.customer.name sale.created_by = request.user sale.save() items = formset.save(commit=False) for item in items: item.sale = sale item.unit_cost = item.product.cost_price item.save() for obj in formset.deleted_objects: obj.delete() process_sale(sale, user=request.user) if sale.customer and sale.payment_method == Sale.PaymentMethod.CREDIT and sale.total > 0: record_customer_ledger_entry( customer=sale.customer, entry_type=CustomerLedgerEntry.EntryType.SALE, debit=sale.total, reference=sale.reference, note="Credit sale posted to customer ledger", date=sale.date, created_by=request.user, sale=sale, ) messages.success(request, f"Sale {sale.reference} recorded successfully.") return redirect("sales:sale_detail", pk=sale.pk) else: form = SaleForm(initial={"date": timezone.now().date()}) formset = SaleItemFormSet() return render( request, "sales/sale_form.html", { "form": form, "formset": formset, }, ) @capability_required(Capability.SALES_READ_ALL, Capability.SALES_READ_OWN, any_of=True) def sale_detail(request, pk): sale = get_object_or_404(Sale, pk=pk) _ensure_sale_visible(request, sale) return render(request, "sales/sale_detail.html", {"sale": sale}) @capability_required(Capability.SALES_VOID) def sale_cancel(request, pk): sale = get_object_or_404(Sale, pk=pk) _ensure_sale_visible(request, sale) if sale.status == Sale.Status.CANCELLED: messages.warning(request, "Sale already cancelled.") return redirect("sales:sale_detail", pk=pk) from inventory.utils import add_stock with transaction.atomic(): for item in sale.items.all(): add_stock( product=item.product, quantity=item.quantity, reference=sale.reference, note=f"Cancellation of {sale.reference}", user=request.user, ) if sale.account and sale.payment_method != "credit": sale.account.withdraw(sale.total) AccountTransaction.objects.create( account=sale.account, tx_type=AccountTransaction.TxType.SALE, direction="out", amount=sale.total, description=f"Cancellation of {sale.reference}", reference=sale.reference, date=sale.date, ) if sale.customer and sale.payment_method == Sale.PaymentMethod.CREDIT and sale.total > 0: record_customer_ledger_entry( customer=sale.customer, entry_type=CustomerLedgerEntry.EntryType.ADJUSTMENT, credit=sale.total, reference=sale.reference, note="Cancelled credit sale reversal", date=timezone.now().date(), created_by=request.user, sale=sale, ) sale.status = Sale.Status.CANCELLED sale.save() messages.success(request, f"{sale.reference} has been cancelled.") return redirect("sales:sale_detail", pk=pk) @capability_required(Capability.INVOICE_READ_ALL, Capability.INVOICE_READ_OWN, any_of=True) def invoice_list(request): invoices = Invoice.objects.prefetch_related("items", "payments").order_by( "-issue_date", "-created_at" ) if not _can_read_all_invoices(request.user): invoices = invoices.filter(created_by=request.user) query = request.GET.get("q") if query: invoices = invoices.filter( Q(reference__icontains=query) | Q(customer_name__icontains=query) | Q(customer__name__icontains=query) ) status = request.GET.get("status") if status: invoices = invoices.filter(status=status) paginator = Paginator(invoices, 20) page = request.GET.get("page") invoices = paginator.get_page(page) return render( request, "sales/invoice_list.html", { "invoices": invoices, "current_status": status or "", "query": query or "", }, ) @capability_required(Capability.INVOICE_CREATE) def invoice_create(request): if request.method == "POST": form = InvoiceForm(request.POST) formset = InvoiceItemFormSet(request.POST) if form.is_valid() and formset.is_valid(): invoice = form.save(commit=False) invoice.reference = generate_invoice_reference() if invoice.customer: if not invoice.customer_name: invoice.customer_name = invoice.customer.name if not invoice.customer_phone: invoice.customer_phone = invoice.customer.phone if not invoice.customer_email: invoice.customer_email = invoice.customer.email if not invoice.customer_address: invoice.customer_address = invoice.customer.address invoice.created_by = request.user invoice.save() items = formset.save(commit=False) for item in items: item.invoice = invoice if item.product: item.unit_cost = item.product.cost_price item.save() for obj in formset.deleted_objects: obj.delete() if invoice.customer and invoice.total > 0: record_customer_ledger_entry( customer=invoice.customer, entry_type=CustomerLedgerEntry.EntryType.INVOICE, debit=invoice.total, reference=invoice.reference, note="Invoice issued", date=invoice.issue_date, created_by=request.user, invoice=invoice, ) messages.success(request, f"{invoice.reference} created.") return redirect("sales:invoice_detail", pk=invoice.pk) else: form = InvoiceForm(initial={"issue_date": timezone.now().date()}) formset = InvoiceItemFormSet() return render( request, "sales/invoice_form.html", { "form": form, "formset": formset, }, ) @capability_required(Capability.INVOICE_READ_ALL, Capability.INVOICE_READ_OWN, any_of=True) def invoice_detail(request, pk): invoice = get_object_or_404(Invoice, pk=pk) _ensure_invoice_visible(request, invoice) payments = invoice.payments.filter(is_cancelled=False).order_by("-date") payment_form = InvoicePaymentForm(invoice=invoice) return render( request, "sales/invoice_detail.html", { "invoice": invoice, "payments": payments, "payment_form": payment_form, }, ) @capability_required(Capability.INVOICE_PAYMENT_MANAGE) def invoice_payment(request, pk): invoice = get_object_or_404(Invoice, pk=pk) _ensure_invoice_visible(request, invoice) if invoice.status == Invoice.Status.PAID: messages.warning(request, "Invoice is already fully paid.") return redirect("sales:invoice_detail", pk=pk) if invoice.status == Invoice.Status.CANCELLED: messages.error(request, "Cannot pay a cancelled invoice.") return redirect("sales:invoice_detail", pk=pk) if request.method == "POST": form = InvoicePaymentForm(request.POST, invoice=invoice) if form.is_valid(): payment = form.save(commit=False) payment.invoice = invoice payment.created_by = request.user payment.save() PaymentAllocation.objects.create( payment=payment, invoice=invoice, amount=payment.amount, created_by=request.user, ) payment.account.deposit(payment.amount) AccountTransaction.objects.create( account=payment.account, tx_type=AccountTransaction.TxType.SALE, direction="in", amount=payment.amount, description=f"Payment for {invoice.reference}", reference=invoice.reference, date=payment.date, ) if invoice.customer and payment.amount > 0: record_customer_ledger_entry( customer=invoice.customer, entry_type=CustomerLedgerEntry.EntryType.PAYMENT, credit=payment.amount, reference=invoice.reference, note=f"Payment posted for {invoice.reference}", date=payment.date, created_by=request.user, invoice=invoice, payment=payment, ) invoice.update_status() messages.success( request, f"Payment of {payment.amount} recorded. " f"Balance due: {invoice.balance_due}", ) return redirect("sales:invoice_detail", pk=pk) return redirect("sales:invoice_detail", pk=pk) @capability_required(Capability.INVOICE_VOID) def invoice_cancel(request, pk): invoice = get_object_or_404(Invoice, pk=pk) _ensure_invoice_visible(request, invoice) if invoice.status == Invoice.Status.CANCELLED: messages.warning(request, "Already cancelled.") return redirect("sales:invoice_detail", pk=pk) if invoice.total_paid > 0: messages.error(request, "Cannot cancel an invoice with payments recorded.") return redirect("sales:invoice_detail", pk=pk) if invoice.customer and invoice.total > 0: record_customer_ledger_entry( customer=invoice.customer, entry_type=CustomerLedgerEntry.EntryType.ADJUSTMENT, credit=invoice.total, reference=invoice.reference, note="Invoice cancellation reversal", date=timezone.now().date(), created_by=request.user, invoice=invoice, ) invoice.status = Invoice.Status.CANCELLED invoice.save() messages.success(request, f"{invoice.reference} cancelled.") return redirect("sales:invoice_list") @capability_required(Capability.SALES_DELETE) def sale_delete(request, pk): sale = get_object_or_404(Sale, pk=pk) _ensure_sale_visible(request, sale) if sale.status != Sale.Status.CANCELLED: messages.error(request, "Only cancelled sales can be deleted. Cancel it first.") return redirect("sales:sale_detail", pk=pk) sale.delete() messages.success(request, f"Sale {sale.reference} permanently deleted.") return redirect("sales:sale_list") @capability_required(Capability.SALES_PRINT) def sale_receipt_pdf(request, pk): sale = get_object_or_404(Sale, pk=pk) _ensure_sale_visible(request, sale) business = Business.get() return render( request, "sales/pdf/receipt.html", { "sale": sale, "business": business, }, ) @capability_required(Capability.INVOICE_PRINT) def invoice_pdf(request, pk): invoice = get_object_or_404(Invoice, pk=pk) _ensure_invoice_visible(request, invoice) business = Business.get() return render( request, "sales/pdf/invoice.html", { "invoice": invoice, "business": business, }, ) @capability_required(Capability.SALES_READ_ALL, Capability.SALES_READ_OWN, any_of=True) def customer_list(request): customers = Customer.objects.filter(is_active=True).order_by("name") query = request.GET.get("q") if query: customers = customers.filter( Q(name__icontains=query) | Q(phone__icontains=query) | Q(email__icontains=query) ) paginator = Paginator(customers, 20) page = request.GET.get("page") customers = paginator.get_page(page) return render( request, "sales/customer_list.html", {"customers": customers, "query": query or ""}, ) @capability_required(Capability.SALES_CREATE) def customer_create(request): if request.method == "POST": form = CustomerForm(request.POST) if form.is_valid(): customer = form.save() messages.success(request, f"{customer.name} created.") return redirect("sales:customer_detail", pk=customer.pk) else: form = CustomerForm() return render(request, "sales/customer_form.html", {"form": form}) @capability_required(Capability.SALES_READ_ALL, Capability.SALES_READ_OWN, any_of=True) def customer_detail(request, pk): customer = get_object_or_404(Customer, pk=pk) sales_qs = customer.sales.order_by("-date", "-created_at") invoices_qs = customer.invoices.order_by("-issue_date", "-created_at") ledger_qs = customer.ledger_entries.select_related( "invoice", "sale", "payment", "created_by" ).order_by("-date", "-created_at") if not _can_read_all_sales(request.user): sales_qs = sales_qs.filter(created_by=request.user) if not _can_read_all_invoices(request.user): invoices_qs = invoices_qs.filter(created_by=request.user) ledger_qs = ledger_qs.filter( Q(created_by=request.user) | Q(sale__created_by=request.user) | Q(invoice__created_by=request.user) ).distinct() return render( request, "sales/customer_detail.html", { "customer": customer, "sales": sales_qs[:10], "invoices": invoices_qs[:10], "ledger_entries": ledger_qs[:30], }, ) def generate_sales_return_reference(): last = SalesReturn.objects.order_by("-id").first() next_id = (last.id + 1) if last else 1 return f"SR-{next_id:04d}" @capability_required(Capability.SALES_READ_ALL, Capability.SALES_READ_OWN, any_of=True) def sales_return_list(request): returns = SalesReturn.objects.select_related("sale", "created_by").order_by("-date") if not _can_read_all_sales(request.user): returns = returns.filter(created_by=request.user) paginator = Paginator(returns, 20) page = request.GET.get("page") returns = paginator.get_page(page) return render(request, "sales/sales_return_list.html", {"returns": returns}) @capability_required(Capability.SALES_CREATE) def sales_return_create(request): if request.method == "POST": form = SalesReturnForm(request.POST) formset = SalesReturnItemFormSet(request.POST) if form.is_valid() and formset.is_valid(): sales_return = form.save(commit=False) sales_return.reference = generate_sales_return_reference() sales_return.created_by = request.user sales_return.save() items = formset.save(commit=False) for item in items: item.sales_return = sales_return if item.unit_price <= 0: item.unit_price = item.product.selling_price item.unit_cost = item.product.cost_price item.save() for obj in formset.deleted_objects: obj.delete() messages.success(request, f"{sales_return.reference} created.") return redirect("sales:sales_return_detail", pk=sales_return.pk) else: form = SalesReturnForm(initial={"date": timezone.now().date()}) formset = SalesReturnItemFormSet() return render( request, "sales/sales_return_form.html", {"form": form, "formset": formset}, ) @capability_required(Capability.SALES_READ_ALL, Capability.SALES_READ_OWN, any_of=True) def sales_return_detail(request, pk): sales_return = get_object_or_404(SalesReturn, pk=pk) _ensure_sale_visible(request, sales_return.sale) return render(request, "sales/sales_return_detail.html", {"sales_return": sales_return}) @capability_required(Capability.SALES_VOID) def sales_return_process(request, pk): sales_return = get_object_or_404(SalesReturn, pk=pk) _ensure_sale_visible(request, sales_return.sale) if sales_return.status != SalesReturn.Status.DRAFT: messages.warning(request, "Only draft returns can be processed.") return redirect("sales:sales_return_detail", pk=pk) if request.method == "POST": from inventory.utils import add_stock with transaction.atomic(): for item in sales_return.items.select_related("product"): add_stock( product=item.product, quantity=item.quantity, reference=sales_return.reference, note=f"Sales return {sales_return.reference}", user=request.user, ) sale = sales_return.sale if sale.account and sale.payment_method != "credit": if sale.account.balance < sales_return.total: messages.error( request, f"Insufficient balance in {sale.account.name} to refund return." ) return redirect("sales:sales_return_detail", pk=pk) sale.account.withdraw(sales_return.total) AccountTransaction.objects.create( account=sale.account, tx_type=AccountTransaction.TxType.SALE, direction="out", amount=sales_return.total, description=f"Refund for {sales_return.reference}", reference=sales_return.reference, date=sales_return.date, ) if sale.customer and sales_return.total > 0: record_customer_ledger_entry( customer=sale.customer, entry_type=CustomerLedgerEntry.EntryType.RETURN, credit=sales_return.total, reference=sales_return.reference, note=f"Sales return processed for {sale.reference}", date=sales_return.date, created_by=request.user, sale=sale, ) sales_return.status = SalesReturn.Status.PROCESSED sales_return.processed_by = request.user sales_return.processed_date = timezone.now().date() sales_return.save() messages.success(request, f"{sales_return.reference} processed.") return redirect("sales:sales_return_detail", pk=pk) return render( request, "sales/sales_return_process_confirm.html", {"sales_return": sales_return}, ) @capability_required(Capability.REPORTS_EXPORT) def export_sales_csv(request): sales = Sale.objects.prefetch_related("items").filter(status="completed").order_by("-date") if not _can_read_all_sales(request.user): sales = sales.filter(created_by=request.user) start = request.GET.get("start") end = request.GET.get("end") if start and end: sales = sales.filter(date__range=(start, end)) headers = [ "Reference", "Date", "Customer", "Payment Method", "Subtotal", "Discount", "Total", "COGS", "Gross Profit", "Account", "Created By", ] rows = [] for sale in sales: rows.append( [ sale.reference, sale.date.strftime("%d/%m/%Y"), sale.customer_display, sale.get_payment_method_display(), sale.subtotal, sale.discount, sale.total, sale.total_cost, sale.gross_profit, sale.account.name if sale.account else "", sale.created_by.username if sale.created_by else "", ] ) return export_csv("sales", headers, rows) @capability_required(Capability.REPORTS_EXPORT) def export_sale_items_csv(request): items = ( SaleItem.objects.select_related("sale", "product") .filter(sale__status="completed") .order_by("-sale__date") ) if not _can_read_all_sales(request.user): items = items.filter(sale__created_by=request.user) start = request.GET.get("start") end = request.GET.get("end") if start and end: items = items.filter(sale__date__range=(start, end)) headers = [ "Sale Reference", "Date", "Customer", "Product", "Quantity", "Unit", "Unit Price", "Unit Cost", "Line Total", "Line Cost", "Line Profit", ] rows = [] for item in items: rows.append( [ item.sale.reference, item.sale.date.strftime("%d/%m/%Y"), item.sale.customer_display, item.product.name, item.quantity, item.product.get_unit_display(), item.unit_price, item.unit_cost, item.line_total, item.line_cost, item.line_profit, ] ) return export_csv("sale-items", headers, rows) @capability_required(Capability.REPORTS_EXPORT) def export_invoices_csv(request): invoices = Invoice.objects.prefetch_related("items", "payments").order_by("-issue_date") if not _can_read_all_invoices(request.user): invoices = invoices.filter(created_by=request.user) status = request.GET.get("status") if status: invoices = invoices.filter(status=status) start = request.GET.get("start") end = request.GET.get("end") if start and end: invoices = invoices.filter(issue_date__range=(start, end)) headers = [ "Reference", "Customer", "Phone", "Issue Date", "Due Date", "Total", "Amount Paid", "Balance Due", "Status", ] rows = [] for inv in invoices: rows.append( [ inv.reference, inv.customer_display, inv.customer_phone, inv.issue_date.strftime("%d/%m/%Y"), inv.due_date.strftime("%d/%m/%Y") if inv.due_date else "", inv.total, inv.total_paid, inv.balance_due, inv.get_status_display(), ] ) return export_csv("invoices", headers, rows)