"""
inventory/admin.py
───────────────────
Django admin for inventory & finance.
Includes custom forms for stock operations (receive, transfer, adjust)
and sale recording — all accessible directly from the admin UI.
"""
import csv
import json
import logging
from datetime import date
from decimal import Decimal
from django import forms
from django.conf import settings
from django.contrib import admin, messages
from django.contrib.admin.widgets import AutocompleteSelect
from django.db import transaction
from django.db.models import Q
from django.forms import formset_factory
from django.http import HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import path, reverse
from django.utils import timezone
from django.utils.html import format_html
from catalog_app.models import Book
from inventory.models import (
AuditLog,
AuthorPayout,
AuthorPayoutItem,
Bookstore,
BookstoreListing,
Customer,
Event,
Expense,
ExpenseCategory,
InventoryBalance,
InventoryTransaction,
PendingInventoryTransaction,
PosMpesaPayment,
Sale,
SaleItem,
StockLocation,
)
from inventory.services.inventory_service import (
adjust_stock,
receive_stock,
transfer_stock,
)
from inventory.services.author_statement_service import send_author_statement_email
from inventory.services.notification_service import notify_approvers, notify_maker_decision
from inventory.services.payout_service import approve_payout
from inventory.services import pos_mpesa_service
from inventory.services.receipt_service import send_receipt_email
from system_logs.models import SystemLog
from system_logs.services import log_event
def _log_refusal(request, module, event, message, *, level=SystemLog.LEVEL_WARNING, **kwargs):
"""Record an action staff tried that was refused or failed (they saw the message once)."""
log_event(module, event, message, level=level, user=request.user, **kwargs)
def _sale_item_summary(items, channel, location) -> dict:
return {
"channel": channel,
"location": str(location or ""),
"items": [
{
"book": getattr(i.get("book"), "title", str(i.get("book") or i.get("bookId", ""))),
"quantity": i.get("quantity", i.get("qty")),
"unit_price": str(i.get("unit_price", i.get("price", ""))),
"discount_percent": str(i.get("discount_percent", i.get("discount", 0))),
}
for i in items
],
}
from inventory.services.sales_service import record_sale as svc_record_sale
from inventory.services.sales_service import void_sale as svc_void_sale
from inventory.services.sales_service import void_sale_item as svc_void_sale_item
logger = logging.getLogger(__name__)
def _is_approver(user) -> bool:
"""True if the user may approve/reject pending inventory transactions."""
return user.is_superuser or user.groups.filter(name="Inventory Approvers").exists()
# ---------------------------------------------------------------------------
# Forms for custom admin views
# ---------------------------------------------------------------------------
def _book_autocomplete_widget():
"""Return an AutocompleteSelect widget for Book, reusing SaleItem.book FK."""
return AutocompleteSelect(
SaleItem._meta.get_field("book"),
admin.site,
attrs={"data-placeholder": "Type to search books…"},
)
class StockReceiveForm(forms.Form):
book = forms.ModelChoiceField(
queryset=Book.objects.all().select_related("author").order_by("title"),
widget=_book_autocomplete_widget(),
help_text="Type the book title to search.",
)
location = forms.ModelChoiceField(
queryset=StockLocation.objects.filter(is_active=True).order_by("name"),
empty_label="— Select location —",
help_text="Where the books are being stored.",
)
quantity = forms.IntegerField(min_value=1, label="Quantity received")
reference = forms.CharField(
max_length=100, required=False,
label="Reference / delivery note #",
help_text="Optional — delivery receipt number or author name.",
)
notes = forms.CharField(
widget=forms.Textarea(attrs={"rows": 3}),
required=False,
)
class StockTransferForm(forms.Form):
book = forms.ModelChoiceField(
queryset=Book.objects.all().select_related("author").order_by("title"),
widget=_book_autocomplete_widget(),
)
source_location = forms.ModelChoiceField(
queryset=StockLocation.objects.filter(is_active=True).order_by("name"),
label="From location",
)
destination_location = forms.ModelChoiceField(
queryset=StockLocation.objects.filter(is_active=True).order_by("name"),
label="To location",
)
quantity = forms.IntegerField(min_value=1)
reference = forms.CharField(max_length=100, required=False)
notes = forms.CharField(
widget=forms.Textarea(attrs={"rows": 3}),
required=False,
)
def clean(self):
data = super().clean()
if data.get("source_location") and data.get("destination_location"):
if data["source_location"] == data["destination_location"]:
raise forms.ValidationError("Source and destination must be different.")
return data
class StockAdjustForm(forms.Form):
book = forms.ModelChoiceField(
queryset=Book.objects.all().select_related("author").order_by("title"),
widget=_book_autocomplete_widget(),
)
location = forms.ModelChoiceField(
queryset=StockLocation.objects.filter(is_active=True).order_by("name"),
)
new_quantity = forms.IntegerField(
min_value=0,
label="New quantity (physical count)",
help_text="Enter the actual count after a stocktake. The difference will be logged.",
)
reference = forms.CharField(
max_length=100, required=False,
label="Reference / reason",
help_text="e.g. 'Annual stocktake', 'Damaged copies'",
)
notes = forms.CharField(
widget=forms.Textarea(attrs={"rows": 3}),
required=False,
)
class SaleHeaderForm(forms.Form):
MODE_SINGLE = "single"
MODE_MULTIPLE = "multiple"
SALE_MODES = [
(MODE_SINGLE, "Single sale — one invoice, one payment"),
(MODE_MULTIPLE, "Multiple sales — one invoice per row (bulk bookstore reporting)"),
]
sale_mode = forms.ChoiceField(choices=SALE_MODES, initial=MODE_SINGLE, label="Recording mode")
sale_channel = forms.ChoiceField(choices=Sale.CHANNELS, label="Sale channel")
sale_date = forms.DateField(
widget=forms.DateInput(attrs={"type": "date"}),
initial=date.today,
)
customer_name = forms.CharField(max_length=255, required=False)
customer_phone = forms.CharField(max_length=20, required=False)
stock_location = forms.ModelChoiceField(
queryset=StockLocation.objects.filter(is_active=True).order_by("name"),
empty_label="— Location —",
required=False,
label="Location",
help_text="Ignored for a Bookstore-channel sale — stock is always deducted from that bookstore's own location.",
)
payment_method = forms.ChoiceField(
choices=Sale.PAYMENT_METHODS, required=False, initial=Sale.PAYMENT_MPESA,
label="Payment method",
help_text="In Multiple sales mode, only used if \"Same payment for all rows\" is checked.",
)
mpesa_code = forms.CharField(
max_length=20, required=False, label="M-Pesa confirmation code",
widget=forms.TextInput(attrs={"style": "text-transform:uppercase;", "placeholder": "e.g. SI97XXXXXXX"}),
)
same_payment_for_all = forms.BooleanField(
required=False, label="Same payment for all rows",
help_text="Multiple sales mode only — use the payment method/code above for every row instead of each row's own.",
)
bookstore = forms.ModelChoiceField(
# Excludes the internal Tai Bookstore/website row — that's resolved
# automatically for walk-in/website sales, never picked by staff.
queryset=Bookstore.objects.filter(is_active=True, is_internal=False).order_by("name"),
required=False, empty_label="— None —",
)
event = forms.ModelChoiceField(
queryset=Event.objects.all().order_by("name"),
required=False, empty_label="— None —",
)
notes = forms.CharField(
widget=forms.Textarea(attrs={"rows": 2}),
required=False,
)
def clean(self):
cleaned = super().clean()
channel = cleaned.get("sale_channel")
if channel == Sale.CHANNEL_BOOKSTORE:
if not cleaned.get("bookstore"):
self.add_error("bookstore", "Choose which bookstore this sale is for.")
elif not cleaned.get("stock_location"):
self.add_error("stock_location", "Choose the location this stock is being sold from.")
return cleaned
class SaleItemForm(forms.Form):
book = forms.ModelChoiceField(
queryset=Book.objects.filter(is_listed=True).select_related("author").order_by("title"),
widget=_book_autocomplete_widget(),
required=False,
)
quantity = forms.IntegerField(min_value=1, required=False, initial=1)
unit_price = forms.DecimalField(
max_digits=10, decimal_places=2, required=False,
label="Unit price (KES)",
)
discount_percent = forms.DecimalField(
max_digits=5, decimal_places=2, required=False, initial=Decimal("0"),
label="Discount %",
)
# Only used/shown in "Multiple sales" mode — each row is its own
# real-world sale, so its customer and (unless
# SaleHeaderForm.same_payment_for_all is checked) its payment can differ
# from every other row.
customer_name = forms.CharField(max_length=255, required=False, label="Customer")
customer_phone = forms.CharField(max_length=20, required=False, label="Phone")
payment_method = forms.ChoiceField(
choices=Sale.PAYMENT_METHODS, required=False, initial=Sale.PAYMENT_MPESA,
label="Payment",
)
mpesa_code = forms.CharField(
max_length=20, required=False, label="M-Pesa code",
widget=forms.TextInput(attrs={"style": "text-transform:uppercase;", "placeholder": "SI97XXXXXXX"}),
)
SaleItemFormSet = formset_factory(SaleItemForm, extra=3, can_delete=True)
# ---------------------------------------------------------------------------
# Stock Location
# ---------------------------------------------------------------------------
@admin.register(StockLocation)
class StockLocationAdmin(admin.ModelAdmin):
list_display = ["name", "code", "location_type", "is_active", "created_at"]
list_filter = ["location_type", "is_active"]
search_fields = ["name", "code"]
ordering = ["name"]
# ---------------------------------------------------------------------------
# Inventory Transaction (ledger + custom stock operation views)
# ---------------------------------------------------------------------------
@admin.register(InventoryTransaction)
class InventoryTransactionAdmin(admin.ModelAdmin):
change_list_template = "admin/inventory/inventorytransaction/change_list.html"
list_display = [
"id", "book", "transaction_type", "quantity",
"source_location", "destination_location",
"transaction_reference", "is_reversed", "created_at",
]
list_filter = ["transaction_type", "is_reversed", "source_location", "destination_location"]
search_fields = ["book__title", "transaction_reference", "notes"]
readonly_fields = [
"id", "book", "transaction_type", "quantity",
"source_location", "destination_location",
"transaction_reference", "is_reversed", "reversed_by",
"performed_by", "created_at", "notes",
]
ordering = ["-created_at"]
def has_add_permission(self, request):
return False
def has_change_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return request.user.is_superuser
# ------------------------------------------------------------------
# Custom URLs for stock operations
# ------------------------------------------------------------------
def get_urls(self):
custom = [
path(
"receive/",
self.admin_site.admin_view(self.stock_receive_view),
name="inventory_stock_receive",
),
path(
"transfer/",
self.admin_site.admin_view(self.stock_transfer_view),
name="inventory_stock_transfer",
),
path(
"adjust/",
self.admin_site.admin_view(self.stock_adjust_view),
name="inventory_stock_adjust",
),
]
return custom + super().get_urls()
def _op_context(self, request, title, form, action_url):
return {
**self.admin_site.each_context(request),
"title": title,
"form": form,
"media": self.media + form.media,
"action_url": action_url,
"opts": self.model._meta,
}
def stock_receive_view(self, request):
if request.method == "POST":
form = StockReceiveForm(request.POST)
if form.is_valid():
pending = PendingInventoryTransaction.objects.create(
op_type=PendingInventoryTransaction.OP_RECEIVE,
book=form.cleaned_data["book"],
dest_location=form.cleaned_data["location"],
quantity=form.cleaned_data["quantity"],
reference=form.cleaned_data.get("reference", ""),
notes=form.cleaned_data.get("notes", ""),
submitted_by=request.user,
)
notify_approvers(pending)
messages.success(
request,
f"Receive request for {pending.quantity} × '{pending.book.title}' "
f"submitted for approval.",
)
return redirect("..")
else:
form = StockReceiveForm()
return render(request, "inventory/admin/stock_receive.html",
self._op_context(request, "Receive Stock from Author / Supplier", form,
"receive/"))
def stock_transfer_view(self, request):
if request.method == "POST":
form = StockTransferForm(request.POST)
if form.is_valid():
pending = PendingInventoryTransaction.objects.create(
op_type=PendingInventoryTransaction.OP_TRANSFER,
book=form.cleaned_data["book"],
source_location=form.cleaned_data["source_location"],
dest_location=form.cleaned_data["destination_location"],
quantity=form.cleaned_data["quantity"],
reference=form.cleaned_data.get("reference", ""),
notes=form.cleaned_data.get("notes", ""),
submitted_by=request.user,
)
notify_approvers(pending)
messages.success(
request,
f"Transfer request for {pending.quantity} × '{pending.book.title}' "
f"from {pending.source_location} → {pending.dest_location} submitted for approval.",
)
return redirect("..")
else:
form = StockTransferForm()
return render(request, "inventory/admin/stock_transfer.html",
self._op_context(request, "Transfer Stock Between Locations", form,
"transfer/"))
def stock_adjust_view(self, request):
if request.method == "POST":
form = StockAdjustForm(request.POST)
if form.is_valid():
pending = PendingInventoryTransaction.objects.create(
op_type=PendingInventoryTransaction.OP_ADJUST,
book=form.cleaned_data["book"],
source_location=form.cleaned_data["location"],
new_quantity=form.cleaned_data["new_quantity"],
reference=form.cleaned_data.get("reference", ""),
notes=form.cleaned_data.get("notes", ""),
submitted_by=request.user,
)
notify_approvers(pending)
messages.success(
request,
f"Adjustment request for '{pending.book.title}' at "
f"{pending.source_location} to qty {pending.new_quantity} submitted for approval.",
)
return redirect("..")
else:
form = StockAdjustForm()
return render(request, "inventory/admin/stock_adjust.html",
self._op_context(request, "Adjust Stock (Stocktake Correction)", form,
"adjust/"))
# ---------------------------------------------------------------------------
# Inventory Balance
# ---------------------------------------------------------------------------
@admin.register(InventoryBalance)
class InventoryBalanceAdmin(admin.ModelAdmin):
list_display = [
"book", "location", "quantity_on_hand",
"quantity_reserved", "quantity_available_col", "low_stock_badge",
]
list_filter = ["location", "location__location_type"]
search_fields = ["book__title", "book__author__name", "location__name"]
readonly_fields = ["book", "location", "quantity_on_hand", "quantity_reserved"]
ordering = ["book__title"]
@admin.display(description="Available")
def quantity_available_col(self, obj):
return obj.quantity_available
@admin.display(description="Alert")
def low_stock_badge(self, obj):
if obj.quantity_on_hand <= obj.book.reorder_level:
return format_html('⚠ Low')
return format_html('OK')
def has_add_permission(self, request):
return False
def has_change_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return request.user.is_superuser
# ---------------------------------------------------------------------------
# Pending Inventory Transaction (maker-checker)
# ---------------------------------------------------------------------------
@admin.register(PendingInventoryTransaction)
class PendingInventoryTransactionAdmin(admin.ModelAdmin):
list_display = [
"id", "op_type_display", "book", "qty_display",
"status_badge", "submitted_by", "submitted_at", "review_link",
]
list_filter = ["status", "op_type", "submitted_by"]
search_fields = ["book__title", "reference", "notes"]
ordering = ["-submitted_at"]
def has_add_permission(self, request):
return False
def has_change_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return request.user.is_superuser
@admin.display(description="Operation")
def op_type_display(self, obj):
return obj.get_op_type_display()
@admin.display(description="Qty")
def qty_display(self, obj):
if obj.op_type == PendingInventoryTransaction.OP_ADJUST:
return f"→ {obj.new_quantity}"
return obj.quantity
@admin.display(description="Status")
def status_badge(self, obj):
colours = {
PendingInventoryTransaction.STATUS_PENDING: ("#B8860B", "#FFF8E1"),
PendingInventoryTransaction.STATUS_APPROVED: ("#1B5E20", "#E8F5E9"),
PendingInventoryTransaction.STATUS_REJECTED: ("#B71C1C", "#FFEBEE"),
}
colour, bg = colours.get(obj.status, ("#555", "#eee"))
return format_html(
'{}',
colour, bg, obj.get_status_display(),
)
@admin.display(description="")
def review_link(self, obj):
return format_html(
'Review →',
obj.pk,
)
def get_urls(self):
custom = [
path(
"/review/",
self.admin_site.admin_view(self.review_view),
name="inventory_pending_review",
),
]
return custom + super().get_urls()
def review_view(self, request, pk):
pending = get_object_or_404(PendingInventoryTransaction, pk=pk)
approver = _is_approver(request.user)
error = None
if request.method == "POST" and approver:
action = request.POST.get("action")
if pending.status != PendingInventoryTransaction.STATUS_PENDING:
messages.error(request, "This transaction has already been reviewed.")
return redirect(".")
if pending.submitted_by == request.user and not request.user.is_superuser:
messages.error(request, "You cannot approve your own submission.")
return redirect(".")
if action == "approve":
try:
with transaction.atomic():
if pending.op_type == PendingInventoryTransaction.OP_RECEIVE:
tx = receive_stock(
book=pending.book,
location=pending.dest_location,
quantity=pending.quantity,
reference=pending.reference,
notes=pending.notes,
user=request.user,
)
elif pending.op_type == PendingInventoryTransaction.OP_TRANSFER:
tx = transfer_stock(
book=pending.book,
source=pending.source_location,
destination=pending.dest_location,
quantity=pending.quantity,
reference=pending.reference,
notes=pending.notes,
user=request.user,
)
else: # adjust
tx = adjust_stock(
book=pending.book,
location=pending.source_location,
new_quantity=pending.new_quantity,
reference=pending.reference,
notes=pending.notes,
user=request.user,
)
pending.status = PendingInventoryTransaction.STATUS_APPROVED
pending.reviewed_by = request.user
pending.reviewed_at = timezone.now()
pending.linked_tx = tx
pending.save()
AuditLog.objects.create(
user=request.user,
action=AuditLog.ACTION_APPROVE,
model_name="PendingInventoryTransaction",
object_id=str(pending.pk),
object_repr=str(pending),
)
notify_maker_decision(pending)
messages.success(request, f"Approved — stock updated successfully.")
return redirect("..")
except ValueError as exc:
error = str(exc)
_log_refusal(
request, SystemLog.MODULE_INVENTORY, f"stock.{pending.op_type}.approval_failed",
f"Approving {pending.get_op_type_display().lower()} of {pending.quantity or pending.new_quantity} × "
f'"{pending.book.title}" failed: {exc}',
reference=pending.reference, related=pending,
request_data={
"book": pending.book.title, "quantity": pending.quantity,
"new_quantity": pending.new_quantity,
"from": str(pending.source_location or ""), "to": str(pending.dest_location or ""),
},
)
elif action == "reject":
reason = request.POST.get("rejection_reason", "").strip()
if not reason:
error = "Please provide a rejection reason."
else:
pending.status = PendingInventoryTransaction.STATUS_REJECTED
pending.reviewed_by = request.user
pending.reviewed_at = timezone.now()
pending.rejection_reason = reason
pending.save()
AuditLog.objects.create(
user=request.user,
action=AuditLog.ACTION_REJECT,
model_name="PendingInventoryTransaction",
object_id=str(pending.pk),
object_repr=str(pending),
)
notify_maker_decision(pending)
messages.success(request, "Transaction rejected and submitter notified.")
return redirect("..")
context = {
**self.admin_site.each_context(request),
"title": f"Review: {pending.get_op_type_display()}",
"opts": self.model._meta,
"pending": pending,
"approver": approver,
"error": error,
}
return render(request, "inventory/admin/pending_review.html", context)
# ---------------------------------------------------------------------------
# Bookstore
# ---------------------------------------------------------------------------
@admin.register(Bookstore)
class BookstoreAdmin(admin.ModelAdmin):
list_display = [
"name", "contact_person", "phone", "email", "stock_location",
"default_bookstore_cut_percent", "is_internal", "is_active",
]
list_filter = ["is_active", "is_internal"]
search_fields = ["name", "contact_person", "email"]
# ---------------------------------------------------------------------------
# BookstoreListing (versioned distribution terms)
# ---------------------------------------------------------------------------
@admin.register(BookstoreListing)
class BookstoreListingAdmin(admin.ModelAdmin):
list_display = [
"book", "bookstore", "agreed_unit_price",
"bookstore_cut_percent", "payee_cut_percent", "effective_from", "created_by",
]
list_filter = ["bookstore", "effective_from"]
search_fields = ["book__title", "bookstore__name"]
date_hierarchy = "effective_from"
autocomplete_fields = ["book", "bookstore"]
readonly_fields = ["created_by", "created_at"]
# Terms are locked after creation — editing them in place would silently
# change history out from under sales already recorded against this row
# (SaleItem snapshots its payout at sale time and never re-reads the
# listing afterward, so an in-place edit just makes the two permanently
# disagree with no visible warning). Add a new row with a later
# effective_from instead; see the fieldset description below.
TERMS_FIELDS = ["book", "bookstore", "agreed_unit_price", "bookstore_cut_percent", "payee_cut_percent", "effective_from"]
def get_readonly_fields(self, request, obj=None):
if obj is not None:
return self.readonly_fields + self.TERMS_FIELDS
return self.readonly_fields
fieldsets = [
(None, {
"fields": [
"book", "bookstore", "agreed_unit_price",
"bookstore_cut_percent", "payee_cut_percent", "effective_from", "notes",
],
"description": (
"To change terms, add a NEW listing with a later effective_from date rather "
"than editing an existing one. Sales already recorded keep the terms that "
"applied when they happened either way, but a new row lets you look back and "
"see exactly what was agreed on any given date."
),
}),
("Record", {"fields": ["created_by", "created_at"], "classes": ["collapse"]}),
]
ordering = ["book__title", "bookstore__name", "-effective_from"]
def save_model(self, request, obj, form, change):
if not obj.pk:
obj.created_by = request.user
super().save_model(request, obj, form, change)
# ---------------------------------------------------------------------------
# Event
# ---------------------------------------------------------------------------
@admin.register(Event)
class EventAdmin(admin.ModelAdmin):
list_display = ["name", "venue", "start_date", "end_date", "stock_location"]
list_filter = ["start_date"]
search_fields = ["name", "venue"]
date_hierarchy = "start_date"
# ---------------------------------------------------------------------------
# Customer
# ---------------------------------------------------------------------------
@admin.register(Customer)
class CustomerAdmin(admin.ModelAdmin):
list_display = ["name", "phone", "email", "created_at"]
search_fields = ["name", "phone", "email"]
ordering = ["name"]
readonly_fields = ["created_at", "updated_at"]
def get_urls(self):
custom = [
path(
"search/",
self.admin_site.admin_view(self.customer_search_view),
name="inventory_customer_search",
),
]
return custom + super().get_urls()
def customer_search_view(self, request):
q = request.GET.get("q", "").strip()
if len(q) < 2:
return JsonResponse({"results": []})
customers = Customer.objects.filter(
Q(phone__icontains=q) | Q(name__icontains=q)
)[:10]
return JsonResponse({"results": [
{"id": c.id, "name": c.name, "phone": c.phone, "email": c.email}
for c in customers
]})
# ---------------------------------------------------------------------------
# Sale (read + custom Record Sale view)
# ---------------------------------------------------------------------------
class SaleItemInline(admin.TabularInline):
model = SaleItem
extra = 0
# Profit/payout margin fields (my_keep_percent, author_pay_percent,
# my_profit_amount, author_payout_amount, payout_included) are
# deliberately excluded from the invoice-editing view — that's business
# margin info, not something the sale/invoice screen needs to surface.
fields = [
"book", "stock_location", "quantity", "unit_price", "discount_percent",
"gross_amount", "net_amount", "void_action",
]
readonly_fields = [
"gross_amount", "discount_amount", "net_amount", "void_action",
]
def has_add_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return False
@admin.display(description="Status")
def void_action(self, obj):
if not obj.pk:
return "—"
if obj.is_voided:
return format_html('Voided')
if obj.sale_id and obj.sale.is_voided:
return "—"
return format_html(
'Void line',
reverse("admin:inventory_saleitem_void", args=[obj.sale_id, obj.pk]),
)
@admin.register(Sale)
class SaleAdmin(admin.ModelAdmin):
change_list_template = "admin/inventory/sale/change_list.html"
list_display = [
"invoice_number", "sale_channel", "customer_name",
"sale_date", "total_net", "recorded_by", "voided_badge",
]
list_filter = ["sale_channel", "sale_date", "created_at", "bookstore", "event", "is_voided"]
search_fields = [
"invoice_number", "customer_name", "customer_phone", "customer_email",
"mpesa_code", "items__book__title",
]
date_hierarchy = "sale_date"
readonly_fields = [
"invoice_number", "total_gross", "total_discount", "total_net",
"created_at", "updated_at",
]
fieldsets = [
("Sale Info", {
"fields": [
"invoice_number", "sale_channel", "sale_date",
"customer", "customer_name", "customer_phone", "customer_email",
"bookstore", "event", "linked_order",
"notes", "recorded_by",
],
}),
("Payment", {
"fields": ["payment_method", "mpesa_code", "amount_tendered", "receipt_sent"],
}),
("Totals", {
# total_profit / total_author_payout deliberately omitted — see
# SaleItemInline comment above.
"fields": ["total_gross", "total_discount", "total_net"],
}),
("Timestamps", {"fields": ["created_at", "updated_at"], "classes": ["collapse"]}),
]
inlines = [SaleItemInline]
ordering = ["-sale_date", "-created_at"]
actions = ["export_sales_csv"]
def has_add_permission(self, request):
return False # use Record Sale form instead
def has_change_permission(self, request, obj=None):
return request.user.is_superuser
def has_delete_permission(self, request, obj=None):
# Sales are never hard-deleted — void instead (restores stock,
# reverses the ledger, and keeps the record for audit). Same
# discipline inventory_service already applies to InventoryTransaction.
return False
@admin.display(description="Status")
def voided_badge(self, obj):
if not obj.is_voided:
return "—"
return format_html('Voided')
def get_fieldsets(self, request, obj=None):
fieldsets = list(super().get_fieldsets(request, obj))
if obj is not None and obj.is_voided:
fieldsets.append(("Void", {
"fields": ["voided_at", "voided_by", "void_reason"],
}))
return fieldsets
def get_readonly_fields(self, request, obj=None):
return [*super().get_readonly_fields(request, obj), "voided_at", "voided_by", "void_reason"]
@admin.action(description="Export selected sales to CSV")
def export_sales_csv(self, request, queryset):
response = HttpResponse(content_type="text/csv")
response["Content-Disposition"] = 'attachment; filename="sales_export.csv"'
writer = csv.writer(response)
writer.writerow([
"Invoice", "Channel", "Date", "Customer",
"Gross", "Discount", "Net", "Profit", "Author Payout",
])
for sale in queryset:
writer.writerow([
sale.invoice_number, sale.get_sale_channel_display(),
sale.sale_date, sale.customer_name,
sale.total_gross, sale.total_discount, sale.total_net,
sale.total_profit, sale.total_author_payout,
])
return response
# ------------------------------------------------------------------
# Custom Record Sale view
# ------------------------------------------------------------------
def get_urls(self):
custom = [
# POS — specific paths first to avoid clash with /*
path(
"pos/",
self.admin_site.admin_view(self.pos_view),
name="inventory_sale_pos",
),
path(
"pos/book-search/",
self.admin_site.admin_view(self.pos_book_search_view),
name="inventory_sale_pos_book_search",
),
path(
"pos/book-by-isbn//",
self.admin_site.admin_view(self.pos_book_by_isbn_view),
name="inventory_sale_pos_book_by_isbn",
),
path(
"pos/mpesa/stk-push/",
self.admin_site.admin_view(self.pos_mpesa_stk_push_view),
name="inventory_sale_pos_mpesa_stk_push",
),
path(
"pos/mpesa/status//",
self.admin_site.admin_view(self.pos_mpesa_status_view),
name="inventory_sale_pos_mpesa_status",
),
# Record Sale (existing)
path(
"record/",
self.admin_site.admin_view(self.record_sale_view),
name="inventory_record_sale",
),
path(
"book-locations/",
self.admin_site.admin_view(self.book_locations_view),
name="inventory_book_locations",
),
# Per-sale receipt routes
path(
"/receipt/",
self.admin_site.admin_view(self.receipt_view),
name="inventory_sale_receipt",
),
path(
"/send-receipt/",
self.admin_site.admin_view(self.send_receipt_view),
name="inventory_sale_send_receipt",
),
path(
"/void/",
self.admin_site.admin_view(self.void_view),
name="inventory_sale_void",
),
path(
"/items//void/",
self.admin_site.admin_view(self.void_item_view),
name="inventory_saleitem_void",
),
]
return custom + super().get_urls()
def book_locations_view(self, request):
"""Return locations with available stock + price for a given book (used by Record Sale form)."""
book_id = request.GET.get("book_id")
if not book_id:
return JsonResponse({"price": "", "locations": []})
# Fetch locations using raw book_id (same as original — avoids any type mismatch)
balances = (
InventoryBalance.objects
.filter(book_id=book_id, quantity_on_hand__gt=0)
.select_related("location")
.order_by("location__name")
)
# Fetch price separately with full error guard
price = ""
try:
book = Book.objects.get(pk=book_id)
raw = book.selling_price_kes or book.price
price = str(int(raw)) if raw else ""
except Exception:
pass
data = {
"price": price,
"locations": [
{
"id": b.location_id,
"name": f"{b.location.name} ({b.location.code}) — {b.quantity_available} avail.",
}
for b in balances
],
}
return JsonResponse(data)
def record_sale_view(self, request):
if request.method == "POST":
header_form = SaleHeaderForm(request.POST)
formset = SaleItemFormSet(request.POST, prefix="items")
if header_form.is_valid() and formset.is_valid():
sale_channel = header_form.cleaned_data["sale_channel"]
bookstore = header_form.cleaned_data.get("bookstore")
# A Bookstore-channel sale always deducts from that bookstore's
# own location — never a manually-picked one — so the stock
# actually leaves where the sale says it happened. Anything
# else uses the single location chosen on the header.
if sale_channel == Sale.CHANNEL_BOOKSTORE:
resolved_location = bookstore.stock_location
if resolved_location is None:
messages.error(
request,
f'"{bookstore.name}" has no stock location linked — set one on the '
f"Bookstore record before recording sales for it.",
)
resolved_location = None
else:
resolved_location = header_form.cleaned_data.get("stock_location")
sale_mode = header_form.cleaned_data["sale_mode"]
same_payment = header_form.cleaned_data.get("same_payment_for_all")
header_payment_method = header_form.cleaned_data.get("payment_method", "")
header_mpesa_code = header_form.cleaned_data.get("mpesa_code", "").strip().upper()
header_customer_name = header_form.cleaned_data.get("customer_name", "")
header_customer_phone = header_form.cleaned_data.get("customer_phone", "")
valid_items = []
for row_number, f in enumerate(formset, start=1):
if f.cleaned_data.get("DELETE"):
continue
book = f.cleaned_data.get("book")
qty = f.cleaned_data.get("quantity")
price = f.cleaned_data.get("unit_price")
if book and qty and price:
valid_items.append({
"row": row_number,
"book": book,
"quantity": qty,
"unit_price": price,
"discount_percent": f.cleaned_data.get("discount_percent") or Decimal("0"),
"stock_location": resolved_location,
"payment_method": f.cleaned_data.get("payment_method", ""),
"mpesa_code": f.cleaned_data.get("mpesa_code", "").strip().upper(),
"customer_name": f.cleaned_data.get("customer_name") or header_customer_name,
"customer_phone": f.cleaned_data.get("customer_phone") or header_customer_phone,
})
if resolved_location is None and sale_channel == Sale.CHANNEL_BOOKSTORE:
pass # error already added above
elif not valid_items:
messages.error(request, "Add at least one item with a book, quantity, and price.")
else:
try:
common_kwargs = dict(
sale_channel=sale_channel,
sale_date=header_form.cleaned_data["sale_date"],
bookstore=bookstore,
event=header_form.cleaned_data.get("event"),
notes=header_form.cleaned_data.get("notes", ""),
user=request.user,
)
row_only_keys = ("row", "payment_method", "mpesa_code", "customer_name", "customer_phone")
if sale_mode == SaleHeaderForm.MODE_MULTIPLE:
# Bulk mode: each row is its own real-world sale
# (e.g. a month of a bookstore's individual sales
# being entered at once) — one Sale/invoice per row,
# each keeping its own customer and payment
# method/reference unless "same payment for all"
# was checked (customer always stays per-row).
# Partial success is deliberate here: one bad row
# (bad book, a transient DB error, etc.) shouldn't
# discard every other already-valid row in a
# month's worth of bulk-entered sales — each row
# gets its own atomic save, and failures are
# reported individually so staff know exactly
# which row(s) to re-enter.
sales = []
failures = [] # (row_number, book_title, error_message)
for item in valid_items:
pm, code = (
(header_payment_method, header_mpesa_code) if same_payment
else (item["payment_method"], item["mpesa_code"])
)
try:
with transaction.atomic():
sale = svc_record_sale(
items=[{k: v for k, v in item.items() if k not in row_only_keys}],
customer_name=item["customer_name"],
customer_phone=item["customer_phone"],
**common_kwargs,
)
sale.payment_method = pm
sale.mpesa_code = code
sale.save(update_fields=["payment_method", "mpesa_code"])
except Exception as exc:
_log_refusal(
request, SystemLog.MODULE_SALES, "sale.bulk_row.failed",
f'Bulk sale row {item["row"]} ({item["book"].title} × {item["quantity"]}) '
f"was not saved: {exc}",
level=SystemLog.LEVEL_WARNING if isinstance(exc, ValueError) else SystemLog.LEVEL_ERROR,
exc=None if isinstance(exc, ValueError) else exc,
phone_number=item["customer_phone"] or "", mpesa_receipt=code or "",
request_data=_sale_item_summary([item], sale_channel, resolved_location),
)
failures.append((item["row"], item["book"].title, str(exc)))
continue
sales.append(sale)
if sales:
total_net = sum(s.total_net for s in sales)
total_profit = sum(s.total_profit for s in sales)
messages.success(
request,
f"{len(sales)} of {len(valid_items)} sale(s) recorded "
f"({sales[0].invoice_number}–{sales[-1].invoice_number}) — "
f"KES {total_net} net, KES {total_profit} profit.",
)
if failures:
detail = " | ".join(
f"Row {row} ({book}): {err}" for row, book, err in failures
)
messages.error(
request,
f"{len(failures)} of {len(valid_items)} row(s) failed to save — {detail}",
)
if sales:
return redirect("..")
else:
items_for_service = [
{k: v for k, v in item.items() if k not in row_only_keys}
for item in valid_items
]
sale = svc_record_sale(
items=items_for_service,
customer_name=header_customer_name,
customer_phone=header_customer_phone,
**common_kwargs,
)
sale.payment_method = header_payment_method
sale.mpesa_code = header_mpesa_code
sale.save(update_fields=["payment_method", "mpesa_code"])
messages.success(
request,
f"Sale {sale.invoice_number} recorded — "
f"KES {sale.total_net} net, KES {sale.total_profit} profit.",
)
return redirect("..")
except ValueError as exc:
messages.error(request, str(exc))
_log_refusal(
request, SystemLog.MODULE_SALES, "sale.record.refused",
f"Record Sale refused: {exc}",
phone_number=header_customer_phone or "", mpesa_receipt=header_mpesa_code or "",
request_data=_sale_item_summary(valid_items, sale_channel, resolved_location),
)
else:
header_form = SaleHeaderForm()
formset = SaleItemFormSet(prefix="items")
return render(request, "inventory/admin/record_sale.html", {
**self.admin_site.each_context(request),
"title": "Record a Sale",
"header_form": header_form,
"formset": formset,
"media": self.media + header_form.media + formset.forms[0].media,
"opts": self.model._meta,
})
# ------------------------------------------------------------------
# POS view + supporting AJAX endpoints
# ------------------------------------------------------------------
def _get_paybill(self):
try:
from store.models import PaybillSettings
return PaybillSettings.get()
except Exception:
return None
# ------------------------------------------------------------------
# POS: M-Pesa STK Push
# ------------------------------------------------------------------
@staticmethod
def _pos_payment_json(payment: PosMpesaPayment) -> dict:
return {
"id": payment.pk,
"status": payment.status,
"status_display": payment.get_status_display(),
"amount": str(payment.amount),
"phone": payment.phone_number,
"receipt": payment.mpesa_receipt_number,
"message": payment.result_desc,
"used": payment.sale_id is not None,
}
def pos_mpesa_stk_push_view(self, request):
if request.method != "POST":
return JsonResponse({"error": "POST required."}, status=405)
if not settings.MPESA_STK_ENABLED:
_log_refusal(
request, SystemLog.MODULE_PAYMENTS, "mpesa.pos.disabled",
"POS STK prompt requested while M-Pesa STK is switched off (MPESA_STK_ENABLED) — refused.",
)
return JsonResponse(
{"error": "M-Pesa prompts aren't live yet. Take payment via Paybill and enter the code."},
status=403,
)
try:
body = json.loads(request.body or "{}")
amount = Decimal(str(body.get("amount", ""))).quantize(Decimal("0.01"))
except (json.JSONDecodeError, ArithmeticError, ValueError):
return JsonResponse({"error": "Invalid request."}, status=400)
if amount <= 0:
return JsonResponse({"error": "Add books to the cart before sending a prompt."}, status=400)
phone = str(body.get("phone", "")).strip()
if not phone:
return JsonResponse({"error": "Enter the customer's M-Pesa number."}, status=400)
previous = None
if body.get("previous_id"):
previous = PosMpesaPayment.objects.filter(pk=body["previous_id"], sale__isnull=True).first()
try:
payment = pos_mpesa_service.send_prompt(phone, amount, user=request.user, previous=previous)
except pos_mpesa_service.PromptStillOpen as exc:
_log_refusal(
request, SystemLog.MODULE_PAYMENTS, "mpesa.pos.resend_blocked", str(exc),
level=SystemLog.LEVEL_INFO, phone_number=phone, reference=f"POS-{previous.pk}" if previous else "",
)
return JsonResponse({"error": str(exc)}, status=409)
except pos_mpesa_service.MpesaError as exc:
# Daraja failures are logged in detail by the M-Pesa service; this
# records the cashier-side outcome (incl. invalid phone numbers).
_log_refusal(
request, SystemLog.MODULE_PAYMENTS, "mpesa.pos.prompt_refused",
f"POS prompt for KES {amount} not sent: {exc}", phone_number=phone,
)
return JsonResponse({"error": str(exc)}, status=400)
return JsonResponse(self._pos_payment_json(payment))
def pos_mpesa_status_view(self, request, payment_id):
"""Polled by the POS. ?check=1 also asks Daraja (sent sparingly — it rate-limits)."""
payment = get_object_or_404(PosMpesaPayment, pk=payment_id)
if request.GET.get("check"):
pos_mpesa_service.refresh_status(payment)
payment.refresh_from_db()
return JsonResponse(self._pos_payment_json(payment))
def pos_book_search_view(self, request):
q = request.GET.get("q", "").strip()
if len(q) < 2:
return JsonResponse({"results": []})
books = (
Book.objects
.filter(Q(title__icontains=q) | Q(isbn__icontains=q))
.filter(is_listed=True)
.select_related("author")
.order_by("title")[:15]
)
return JsonResponse({"results": [
{
"id": b.id,
"title": b.title,
"price": str(b.pos_price or b.selling_price_kes or b.price or "0"),
"stock": b.stock,
"isbn": b.isbn or "",
}
for b in books
]})
def pos_book_by_isbn_view(self, request, isbn):
book = Book.objects.filter(isbn=isbn).first()
if not book:
return JsonResponse({"error": "Book not found"}, status=404)
return JsonResponse({
"id": book.id,
"title": book.title,
"price": str(book.pos_price or book.selling_price_kes or book.price or "0"),
"stock": book.stock,
"isbn": book.isbn or "",
})
def pos_view(self, request):
paybill = self._get_paybill()
default_location = StockLocation.objects.filter(code="MAIN-WH").first()
locations = StockLocation.objects.filter(is_active=True).order_by("name")
if request.method == "POST":
return self._pos_handle_post(request, paybill, default_location, locations)
return render(request, "inventory/admin/pos.html", {
**self.admin_site.each_context(request),
"title": "Point of Sale",
"opts": self.model._meta,
"paybill": paybill,
"default_location": default_location,
"locations": locations,
"today": date.today().isoformat(),
"channels": Sale.CHANNELS,
"mpesa_stk_enabled": settings.MPESA_STK_ENABLED,
})
def _pos_handle_post(self, request, paybill, default_location, locations):
error = None
try:
cart_raw = request.POST.get("cart_data", "")
if not cart_raw:
raise ValueError("Cart is empty — add at least one book.")
cart_items = json.loads(cart_raw)
if not cart_items:
raise ValueError("Cart is empty — add at least one book.")
# Resolve stock location
loc_id = request.POST.get("stock_location_id")
try:
stock_location = StockLocation.objects.get(pk=int(loc_id)) if loc_id else default_location
except StockLocation.DoesNotExist:
stock_location = default_location
# Build items list + validate stock at the chosen location
items = []
for row in cart_items:
book = Book.objects.get(pk=int(row["bookId"]))
qty = int(row["qty"])
if book.track_inventory and stock_location:
balance = (
InventoryBalance.objects
.filter(book=book, location=stock_location)
.first()
)
available = balance.quantity_available if balance else 0
if available < qty:
raise ValueError(
f'"{book.title}" — only {available} unit{"s" if available != 1 else ""} '
f'available at {stock_location.name} '
f'({qty} requested). Adjust the quantity or choose a different location.'
)
elif book.track_inventory and book.stock < qty:
raise ValueError(
f'"{book.title}" — only {book.stock} unit{"s" if book.stock != 1 else ""} '
f'in stock ({qty} requested).'
)
items.append({
"book": book,
"quantity": qty,
"unit_price": Decimal(str(row["price"])),
"discount_percent": Decimal(str(row.get("discount", 0))),
"stock_location": stock_location,
})
# Customer: get-or-create by phone
customer_phone = request.POST.get("customer_phone", "").strip()
customer_name = request.POST.get("customer_name", "").strip()
customer_email = request.POST.get("customer_email", "").strip()
customer_obj = None
if customer_phone:
customer_obj, _ = Customer.objects.get_or_create(
phone=customer_phone,
defaults={"name": customer_name, "email": customer_email},
)
# Update name/email if provided and different
changed = False
if customer_name and customer_obj.name != customer_name:
customer_obj.name = customer_name
changed = True
if customer_email and customer_obj.email != customer_email:
customer_obj.email = customer_email
changed = True
if changed:
customer_obj.save(update_fields=["name", "email"])
sale_channel = request.POST.get("sale_channel", Sale.CHANNEL_WALK_IN)
sale_date_str = request.POST.get("sale_date", date.today().isoformat())
try:
from datetime import datetime
sale_date = datetime.strptime(sale_date_str, "%Y-%m-%d").date()
except ValueError:
sale_date = date.today()
payment_method = request.POST.get("payment_method", Sale.PAYMENT_MPESA)
mpesa_code = request.POST.get("mpesa_code", "").strip().upper()
stk_payment_id = request.POST.get("mpesa_payment_id", "").strip()
use_stk = payment_method == Sale.PAYMENT_MPESA and stk_payment_id.isdigit()
amount_tendered_raw = request.POST.get("amount_tendered", "")
try:
amount_tendered = Decimal(amount_tendered_raw) if amount_tendered_raw else None
except Exception:
amount_tendered = None
notes = request.POST.get("notes", "").strip()
with transaction.atomic():
sale = svc_record_sale(
items=items,
sale_channel=sale_channel,
sale_date=sale_date,
customer_name=customer_name or (customer_obj.name if customer_obj else ""),
customer_phone=customer_phone,
notes=notes,
user=request.user,
)
if use_stk:
# Raises ValueError (rolling back the sale) if the prompt
# isn't paid, is already used, or doesn't cover the total.
stk_payment = pos_mpesa_service.claim_for_sale(int(stk_payment_id), sale)
mpesa_code = (stk_payment.mpesa_receipt_number or mpesa_code)[:20]
sale.customer = customer_obj
sale.customer_email = customer_email
sale.payment_method = payment_method
sale.mpesa_code = mpesa_code
sale.amount_tendered = amount_tendered
sale.save(update_fields=[
"customer", "customer_email", "payment_method",
"mpesa_code", "amount_tendered",
])
# Send email receipt if email provided (outside transaction — non-critical)
if customer_email:
sent = send_receipt_email(sale)
if sent:
messages.success(request, f"Receipt emailed to {customer_email}.")
return redirect("admin:inventory_sale_receipt", sale.id)
except (ValueError, json.JSONDecodeError) as exc:
error = str(exc)
except Book.DoesNotExist:
error = "One or more books could not be found. Please refresh and try again."
try:
cart_summary = json.loads(request.POST.get("cart_data") or "[]")
except json.JSONDecodeError:
cart_summary = request.POST.get("cart_data", "")
_log_refusal(
request, SystemLog.MODULE_POS, "pos.sale.refused", f"POS sale not completed: {error}",
phone_number=request.POST.get("customer_phone", ""),
reference=f"POS-{request.POST['mpesa_payment_id']}" if request.POST.get("mpesa_payment_id") else "",
mpesa_receipt=request.POST.get("mpesa_code", ""),
request_data={
"cart": cart_summary,
"payment_method": request.POST.get("payment_method"),
"stock_location_id": request.POST.get("stock_location_id"),
},
)
paybill = self._get_paybill()
default_location = StockLocation.objects.filter(code="MAIN-WH").first()
locations = StockLocation.objects.filter(is_active=True).order_by("name")
return render(request, "inventory/admin/pos.html", {
**self.admin_site.each_context(request),
"title": "Point of Sale",
"opts": self.model._meta,
"paybill": paybill,
"default_location": default_location,
"locations": locations,
"today": date.today().isoformat(),
"channels": Sale.CHANNELS,
"mpesa_stk_enabled": settings.MPESA_STK_ENABLED,
"error": error,
})
def receipt_view(self, request, sale_id):
sale = get_object_or_404(Sale, pk=sale_id)
items = sale.items.select_related("book").all()
paybill = self._get_paybill()
change_due = None
if sale.payment_method == Sale.PAYMENT_CASH and sale.amount_tendered:
change_due = sale.amount_tendered - sale.total_net
return render(request, "inventory/admin/receipt.html", {
**self.admin_site.each_context(request),
"title": f"Receipt — {sale.invoice_number}",
"opts": self.model._meta,
"sale": sale,
"items": items,
"paybill": paybill,
"change_due": change_due,
})
def void_view(self, request, sale_id):
sale = get_object_or_404(Sale, pk=sale_id)
if sale.is_voided:
messages.error(request, f"{sale.invoice_number} is already voided.")
return redirect("admin:inventory_sale_change", sale_id)
if request.method == "POST":
reason = request.POST.get("reason", "").strip()
try:
svc_void_sale(sale, user=request.user, reason=reason)
except ValueError as exc:
messages.error(request, str(exc))
_log_refusal(request, SystemLog.MODULE_SALES, "sale.void.refused",
f"Voiding {sale.invoice_number} refused: {exc}",
reference=sale.invoice_number, related=sale)
else:
messages.success(request, f"{sale.invoice_number} voided — stock restored.")
return redirect("admin:inventory_sale_change", sale_id)
return render(request, "admin/inventory/sale/void_confirm.html", {
**self.admin_site.each_context(request),
"title": f"Void {sale.invoice_number}?",
"opts": self.model._meta,
"sale": sale,
})
def void_item_view(self, request, sale_id, item_id):
sale = get_object_or_404(Sale, pk=sale_id)
item = get_object_or_404(SaleItem, pk=item_id, sale=sale)
if item.is_voided:
messages.error(request, "This line is already voided.")
return redirect("admin:inventory_sale_change", sale_id)
if sale.is_voided:
messages.error(request, f"{sale.invoice_number} is already voided.")
return redirect("admin:inventory_sale_change", sale_id)
if request.method == "POST":
reason = request.POST.get("reason", "").strip()
try:
svc_void_sale_item(item, user=request.user, reason=reason)
except ValueError as exc:
messages.error(request, str(exc))
_log_refusal(request, SystemLog.MODULE_SALES, "sale.void_item.refused",
f"Voiding {item.book.title} on {sale.invoice_number} refused: {exc}",
reference=sale.invoice_number, related=item)
else:
messages.success(request, f"{item.book.title} voided on {sale.invoice_number} — stock restored.")
return redirect("admin:inventory_sale_change", sale_id)
return render(request, "admin/inventory/sale/void_item_confirm.html", {
**self.admin_site.each_context(request),
"title": f"Void line — {item.book.title}?",
"opts": self.model._meta,
"sale": sale,
"item": item,
})
def send_receipt_view(self, request, sale_id):
if request.method != "POST":
return redirect("admin:inventory_sale_receipt", sale_id)
sale = get_object_or_404(Sale, pk=sale_id)
if not sale.customer_email:
messages.error(request, "No email address on this sale.")
else:
ok = send_receipt_email(sale)
if ok:
messages.success(request, f"Receipt resent to {sale.customer_email}.")
else:
messages.error(request, "Failed to send email — details are in the System Log.")
return redirect("admin:inventory_sale_receipt", sale_id)
# ---------------------------------------------------------------------------
# Author Payout
# ---------------------------------------------------------------------------
class AuthorPayoutItemInline(admin.TabularInline):
model = AuthorPayoutItem
extra = 0
fields = ["book", "gross_earnings", "adjustment_amount", "adjustment_notes", "net_earnings"]
readonly_fields = ["book", "gross_earnings", "net_earnings"]
def has_add_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return False
@admin.register(AuthorPayout)
class AuthorPayoutAdmin(admin.ModelAdmin):
list_display = [
"id", "payee_display", "period_type", "period_start", "period_end",
"total_earnings", "total_paid", "outstanding_balance", "status_badge",
]
list_filter = ["status", "period_type", "author", "publisher"]
search_fields = ["author__name", "publisher__name", "payment_reference"]
date_hierarchy = "period_start"
readonly_fields = [
"total_earnings", "total_adjustments", "total_paid",
"outstanding_balance", "approved_at", "paid_at",
"created_at", "updated_at",
]
fieldsets = [
("Payout Details", {
"fields": ["author", "publisher", "period_type", "period_start", "period_end", "status", "notes"],
"description": "Set exactly one of author or publisher.",
}),
("Financials", {
"fields": ["total_earnings", "total_adjustments", "total_paid", "outstanding_balance"],
}),
("Payment", {
"fields": ["payment_method", "payment_reference", "payment_date", "paid_at"],
}),
("Approval", {
"fields": ["prepared_by", "approved_by", "approved_at"],
}),
("Timestamps", {"fields": ["created_at", "updated_at"], "classes": ["collapse"]}),
]
inlines = [AuthorPayoutItemInline]
ordering = ["-created_at"]
actions = ["action_approve", "export_payouts_csv"]
def get_urls(self):
custom = [
path(
"/send-statement/",
self.admin_site.admin_view(self.send_statement_view),
name="inventory_authorpayout_send_statement",
),
]
return custom + super().get_urls()
def send_statement_view(self, request, pk):
payout = get_object_or_404(AuthorPayout, pk=pk)
if request.method != "POST":
return redirect("admin:inventory_authorpayout_change", pk)
payee = payout.payee
if not payee.email:
messages.error(request, f"{payee.name} has no email address on file.")
else:
ok = send_author_statement_email(payee=payee, payout=payout)
if ok:
messages.success(request, f"Statement emailed to {payee.email}.")
else:
messages.error(request, "Failed to send email — details are in the System Log.")
return redirect("admin:inventory_authorpayout_change", pk)
@admin.display(description="Payee")
def payee_display(self, obj):
return f"{obj.payee.name} ({obj.payee_label})"
@admin.display(description="Status")
def status_badge(self, obj):
colors = {
AuthorPayout.STATUS_PENDING: "#e6ac00",
AuthorPayout.STATUS_APPROVED: "#1a7abf",
AuthorPayout.STATUS_PAID: "#2e7d32",
AuthorPayout.STATUS_CANCELLED: "#b00020",
}
color = colors.get(obj.status, "#666")
return format_html(
'{}',
color, obj.get_status_display(),
)
@admin.action(description="Approve selected payouts")
def action_approve(self, request, queryset):
approved = 0
for payout in queryset.filter(status=AuthorPayout.STATUS_PENDING):
try:
approve_payout(payout=payout, user=request.user)
approved += 1
except Exception as exc:
self.message_user(request, f"Payout #{payout.pk} error: {exc}", messages.ERROR)
_log_refusal(
request, SystemLog.MODULE_PAYOUTS, "payout.approve.failed",
f"Approving payout #{payout.pk} failed: {exc}",
level=SystemLog.LEVEL_WARNING if isinstance(exc, ValueError) else SystemLog.LEVEL_ERROR,
exc=None if isinstance(exc, ValueError) else exc,
reference=f"PAYOUT-{payout.pk}", related=payout,
)
if approved:
self.message_user(request, f"{approved} payout(s) approved.", messages.SUCCESS)
@admin.action(description="Export selected payouts to CSV")
def export_payouts_csv(self, request, queryset):
response = HttpResponse(content_type="text/csv")
response["Content-Disposition"] = 'attachment; filename="payouts_export.csv"'
writer = csv.writer(response)
writer.writerow(["ID", "Payee", "Payee Type", "Period", "Start", "End",
"Total Earnings", "Total Paid", "Outstanding", "Status"])
for p in queryset:
writer.writerow([
p.pk, p.payee.name, p.payee_label, p.get_period_type_display(),
p.period_start, p.period_end,
p.total_earnings, p.total_paid, p.outstanding_balance,
p.get_status_display(),
])
return response
# ---------------------------------------------------------------------------
# Expense Category
# ---------------------------------------------------------------------------
@admin.register(ExpenseCategory)
class ExpenseCategoryAdmin(admin.ModelAdmin):
list_display = ["name", "description"]
search_fields = ["name"]
# ---------------------------------------------------------------------------
# Expense
# ---------------------------------------------------------------------------
@admin.register(Expense)
class ExpenseAdmin(admin.ModelAdmin):
list_display = [
"category", "amount", "date", "description",
"linked_book", "linked_event", "recorded_by",
]
list_filter = ["category", "date"]
search_fields = ["description", "notes"]
date_hierarchy = "date"
readonly_fields = ["created_at", "recorded_by"]
ordering = ["-date"]
actions = ["export_expenses_csv"]
def save_model(self, request, obj, form, change):
if not obj.pk:
obj.recorded_by = request.user
super().save_model(request, obj, form, change)
@admin.action(description="Export selected expenses to CSV")
def export_expenses_csv(self, request, queryset):
response = HttpResponse(content_type="text/csv")
response["Content-Disposition"] = 'attachment; filename="expenses_export.csv"'
writer = csv.writer(response)
writer.writerow(["Category", "Amount", "Date", "Description",
"Linked Book", "Linked Event", "Notes"])
for e in queryset:
writer.writerow([
e.category.name, e.amount, e.date, e.description,
str(e.linked_book) if e.linked_book else "",
str(e.linked_event) if e.linked_event else "",
e.notes,
])
return response
# ---------------------------------------------------------------------------
# Audit Log
# ---------------------------------------------------------------------------
@admin.register(AuditLog)
class AuditLogAdmin(admin.ModelAdmin):
list_display = ["timestamp", "user", "action", "model_name", "object_repr", "ip_address"]
list_filter = ["action", "model_name"]
search_fields = ["user__username", "object_repr", "ip_address"]
readonly_fields = [
"user", "action", "model_name", "object_id", "object_repr",
"old_values", "new_values", "ip_address", "timestamp",
]
ordering = ["-timestamp"]
date_hierarchy = "timestamp"
def has_add_permission(self, request):
return False
def has_change_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return request.user.is_superuser
# ---------------------------------------------------------------------------
# POS M-Pesa payments (read-only — written by the POS and Safaricom)
# ---------------------------------------------------------------------------
@admin.register(PosMpesaPayment)
class PosMpesaPaymentAdmin(admin.ModelAdmin):
list_display = (
"created_at", "phone_number", "amount", "status", "mpesa_receipt_number",
"sale_link", "created_by",
)
list_filter = ("status", "created_at")
search_fields = ("phone_number", "mpesa_receipt_number", "checkout_request_id", "sale__invoice_number")
list_select_related = ("sale", "created_by")
readonly_fields = [f.name for f in PosMpesaPayment._meta.fields]
@admin.display(description="Sale")
def sale_link(self, obj):
if not obj.sale_id:
return "— not used" if obj.status == PosMpesaPayment.STATUS_PAID else "—"
url = reverse("admin:inventory_sale_change", args=[obj.sale_id])
return format_html('{}', url, obj.sale.invoice_number)
def has_add_permission(self, request):
return False
def has_change_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return False
# ---------------------------------------------------------------------------
# Reports & Dashboard
# ---------------------------------------------------------------------------
from . import admin_reports # noqa: E402,F401 (registers ReportsHubAdmin)