from django.contrib import admin from django.contrib.auth.admin import UserAdmin from django.contrib.auth.models import User from django.db.models import Sum from django.db.models import Sum, Count from django.utils.html import format_html from django.utils.safestring import mark_safe from django import forms from django.contrib import messages from django.utils import timezone from django.db import transaction as db_transaction from decimal import Decimal from django.shortcuts import render, redirect, get_object_or_404 from django.urls import path from django.http import HttpResponseRedirect from django.urls import reverse from io import StringIO from datetime import timedelta from django.core.management import call_command from django.contrib.admin.views.decorators import staff_member_required from .models import ( Product, Location, StockLevel, StockLevelSize, Size, Device, Transaction, TransactionItem, StockMovement, UserProfile, ProfitRecord, LocationCashBalance, CashTransaction, Purchase, CashTransfer,ExpenseCategory,Expense,SalesTarget, BonusRecord,CreditSale, MatatuIncome, MatatuExpense ) from django.db import transaction as db_transaction from .views import _perform_reversal from django.urls import path # --- User Profile Inline --- class UserProfileInline(admin.StackedInline): model = UserProfile can_delete = False verbose_name_plural = 'Profile' class CustomUserAdmin(UserAdmin): inlines = (UserProfileInline,) admin.site.unregister(User) admin.site.register(User, CustomUserAdmin) # --- Product Admin --- class StockLevelInline(admin.TabularInline): model = StockLevel extra = 0 @admin.register(Product) class ProductAdmin(admin.ModelAdmin): list_display = ('sku', 'name', 'unit_price', 'created_at', 'buying_price', 'image_preview_thumbnail') search_fields = ('sku', 'name') list_filter = ('created_at',) readonly_fields = ['image_preview'] inlines = [StockLevelInline] def image_preview_thumbnail(self, obj): """Small thumbnail for list view""" if obj.image: return format_html('

', obj.image.url) return "📷 No image" image_preview_thumbnail.short_description = 'Image' def image_preview(self, obj): """Larger preview for detail view""" if obj.image: return format_html('

', obj.image.url) return "No image uploaded" image_preview.short_description = 'Product Image Preview' # --- Location Admin --- class StockLevelInlineForLocation(admin.TabularInline): model = StockLevel extra = 0 readonly_fields = ('stock_value',) def stock_value(self, obj): """Live calculation of stock value for this product at this location""" if obj.product and obj.product.buying_price: value = obj.quantity * obj.product.buying_price return f"KES {value:,.2f}" return "KES 0.00" stock_value.short_description = "Stock Value" @admin.register(Location) class LocationAdmin(admin.ModelAdmin): list_display = ('name', 'is_truck', 'total_stock', 'stock_valuation') inlines = [StockLevelInlineForLocation] def total_stock(self, obj): """Total quantity of all products at this location""" return sum(level.quantity for level in obj.stock_levels.all()) total_stock.short_description = "Total Stock (Units)" def stock_valuation(self, obj): """ Live calculation of total stock value at this location. Calculates: SUM(quantity × buying_price) for all products. Updates automatically when stock changes. """ from decimal import Decimal total_value = Decimal('0.00') for level in obj.stock_levels.select_related('product'): if level.product and level.product.buying_price: total_value += level.quantity * level.product.buying_price return f"KES {total_value:,.2f}" stock_valuation.short_description = "Stock Valuation 💰" stock_valuation.admin_order_field = 'stock_levels__quantity' def changelist_view(self, request, extra_context=None): """Add purchase summary statistics""" response = super().changelist_view(request, extra_context) try: qs = response.context_data['cl'].queryset # Aggregate purchases for these locations through the relationship total_purchases = Purchase.objects.filter( destination_location__in=qs ).aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') total_paid = Purchase.objects.filter( destination_location__in=qs, payment_status='paid' ).aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') total_credit = Purchase.objects.filter( destination_location__in=qs, payment_status='credit' ).aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') # Format numbers FIRST total_purchases_str = f"{total_purchases:,.2f}" total_paid_str = f"{total_paid:,.2f}" total_credit_str = f"{total_credit:,.2f}" # Count by destination by_location = {} for location in Location.objects.all(): loc_total = Purchase.objects.filter( destination_location=location ).aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') if loc_total > 0: by_location[location.name] = f"{loc_total:,.2f}" loc_cards = "" for loc_name, amount_str in by_location.items(): loc_cards += f"""

📦 {loc_name}

KES {amount_str}

""" summary_html = f"""

🛒 Purchase Summary

💰 Total Purchases

KES {total_purchases_str}

✅ Paid

KES {total_paid_str}

⏳ On Credit

KES {total_credit_str}

Stock Delivered To:

{loc_cards}

{icon} {method_name}

KES {amount:,.2f}

""" # ✅ NEW: Check if there are balances to consolidate target_location = Location.objects.filter(name__iexact='Main Store').first() has_balances_to_consolidate = False if target_location: has_balances_to_consolidate = LocationCashBalance.objects.exclude( location=target_location, payment_method='bank_transfer' ).filter(balance__gt=0).exists() consolidate_url = reverse('admin:core_locationcashbalance_consolidate') # ✅ NEW: Consolidation button HTML consolidation_button = f"""

💰 Cash Consolidation

Move all cash from locations/methods to Main Store Bank

{'🏦 Consolidate to Bank' if has_balances_to_consolidate else '✅ All in Bank'}

{f'''

💡 Tip: Click the button to preview what will be transferred before confirming.

''' if has_balances_to_consolidate else ''}

""" summary_html = f""" {consolidation_button}

💰 Cash Balance Summary (The Bank)

💰 Total Balance

KES {total_balance:,.2f}

📈 All-Time Sales

KES {total_sales_all:,.2f}

📉 All-Time Purchases

KES {total_purchases_all:,.2f}

Breakdown by Payment Method:

{breakdown_cards}

ℹ️ Updates automatically with each transaction | Locations: {qs.values('location').distinct().count()}

""" extra_context = extra_context or {} extra_context['summary_stats'] = mark_safe(summary_html) response.context_data.update(extra_context) except (AttributeError, KeyError): pass return response # --- Cash Transaction Log Admin --- @admin.register(CashTransaction) class CashTransactionAdmin(admin.ModelAdmin): list_display = [ 'created_at', 'location', 'transaction_type_display', 'payment_method_display', 'amount_colored', 'balance_after_display', 'created_by' ] list_filter = [ 'location', 'transaction_type', 'payment_method', 'created_at' ] search_fields = [ 'location__name', 'description', 'reference_number', 'sale_transaction__id' ] readonly_fields = [ 'balance_after', 'created_at' ] date_hierarchy = 'created_at' def transaction_type_display(self, obj): icons = { 'sale': '💰', 'purchase': '📦', 'withdrawal': '🏧', 'deposit': '💵', 'transfer': '🔄', 'adjustment': '⚖️', 'transfer_out': '📤', 'transfer_in': '📥', } icon = icons.get(obj.transaction_type, '📝') return f"{icon} {obj.get_transaction_type_display()}" transaction_type_display.short_description = 'Type' def payment_method_display(self, obj): icons = { 'cash': '💵', 'mpesa_paybill': '📱', 'mpesa_till': '🏪', 'bank_transfer': '🏦', } icon = icons.get(obj.payment_method, '💳') return f"{icon} {obj.get_payment_method_display()}" payment_method_display.short_description = 'Method' def amount_colored(self, obj): color = '#10b981' if obj.transaction_type in ['sale', 'deposit', 'transfer_in'] else '#ef4444' sign = '+' if obj.transaction_type in ['sale', 'deposit', 'transfer_in'] else '-' return format_html( '{} KES {}', color, sign, f'{abs(obj.amount):,.2f}' ) amount_colored.short_description = 'Amount' amount_colored.admin_order_field = 'amount' def balance_after_display(self, obj): balance_str = f"{obj.balance_after:,.2f}" return format_html( 'KES {}', balance_str ) balance_after_display.short_description = 'Balance After' balance_after_display.admin_order_field = 'balance_after' # --- Transaction Admin --- class TransactionItemInline(admin.TabularInline): model = TransactionItem readonly_fields = ('product', 'size', 'quantity', 'unit_price', 'line_total') extra = 0 # ============================================================ # Replace your entire TransactionAdmin class in admin.py with this. # Make sure these imports are at the top of admin.py: # # from django.db import transaction as db_transaction # from django.urls import path # from .views import _perform_reversal # ============================================================ class TransactionItemInline(admin.TabularInline): model = TransactionItem readonly_fields = ('product', 'size', 'quantity', 'unit_price', 'line_total') extra = 0 @admin.register(Transaction) class TransactionAdmin(admin.ModelAdmin): # ── Config ──────────────────────────────────────────────────────────── actions = ['action_reverse_transaction'] list_display = ( 'receipt_display', 'created_at', 'location_display', 'payment_method_display', 'total_amount_display', 'mpesa_reference', 'synced', 'reverse_button', ) list_filter = ( 'payment_method', 'synced', 'created_at', 'device__location', ) date_hierarchy = 'created_at' search_fields = ( 'server_receipt_number', 'id', 'mpesa_reference', 'payment_phone', ) readonly_fields = ( 'id', 'created_at', 'server_receipt_number', 'mpesa_reference', 'payment_phone', ) inlines = [TransactionItemInline] class Media: css = {'all': ('admin/css/transaction_summary.css',)} # ── Column helpers ──────────────────────────────────────────────────── def location_display(self, obj): if obj.device and obj.device.location: return obj.device.location.name return 'N/A' location_display.short_description = 'Location' def receipt_display(self, obj): short_id = str(obj.id)[:8].upper() return format_html( '#{}', short_id, ) receipt_display.short_description = 'Receipt #' receipt_display.admin_order_field = 'id' def payment_method_display(self, obj): icons = { 'cash': '💵', 'mpesa_paybill': '📱', 'mpesa_till': '🏪', 'bank_transfer': '🏦', 'on_credit': '📋', } icon = icons.get(obj.payment_method, '💳') method_name = obj.get_payment_method_display() if obj.mpesa_reference and obj.payment_method in ['mpesa_paybill', 'mpesa_till']: return format_html( '{} {} ({})', icon, method_name, obj.mpesa_reference, ) if obj.payment_method == 'on_credit': try: customer = obj.credit_sale.customer_name return format_html( '{} {} ({})', icon, method_name, customer, ) except Exception: pass return f"{icon} {method_name}" payment_method_display.short_description = 'Payment' def total_amount_display(self, obj): return format_html( 'KES {}', f'{obj.total_amount:,.2f}', ) total_amount_display.short_description = 'Amount' total_amount_display.admin_order_field = 'total_amount' # ── Reverse button column ───────────────────────────────────────────── def reverse_button(self, obj): time_limit = timezone.now() - timedelta(hours=24) if obj.created_at < time_limit: return mark_safe( '' '⏰ Expired' ) try: if obj.credit_sale.is_paid: return mark_safe( '' '💚 Credit Paid' ) except Exception: pass receipt = obj.server_receipt_number or str(obj.id)[:8] url = f'/admin/core/transaction/reverse-transaction/{obj.id}/' confirm_msg = ( f'Reverse transaction #{receipt}?\\n\\n' f'Amount: KES {obj.total_amount:,.2f}\\n' f'This will:\\n' f' • Restock all items\\n' f' • Reverse the cash balance\\n' f' • Delete all related records\\n\\n' f'This CANNOT be undone. Continue?' ) return format_html( '' '🔴 Reverse' '', url, confirm_msg, ) reverse_button.short_description = 'Action' # ── Custom URL for reverse button ───────────────────────────────────── def get_urls(self): urls = super().get_urls() custom = [ path( 'reverse-transaction//', self.admin_site.admin_view(self.reverse_transaction_view), name='core_reverse_transaction', ), ] return custom + urls def reverse_transaction_view(self, request, transaction_id): from django.shortcuts import redirect try: with db_transaction.atomic(): tx = Transaction.objects.select_for_update().get(id=transaction_id) time_limit = timezone.now() - timedelta(hours=24) if tx.created_at < time_limit: self.message_user( request, '❌ Cannot reverse — transaction is older than 24 hours.', level='error', ) return redirect('admin:core_transaction_changelist') try: if tx.credit_sale.is_paid: self.message_user( request, '❌ Cannot reverse — this credit sale has already been paid.', level='error', ) return redirect('admin:core_transaction_changelist') except Exception: pass result = _perform_reversal( tx, request.user, reason=f'Reversed by admin: {request.user.username}', ) if result['success']: details = result['reversal_details'] restocked = details['items_restocked'] items_text = ', '.join( f"{i['product']} sz{i['size']} (+{i['quantity_restored']})" for i in restocked ) self.message_user( request, f'✅ Transaction #{details["receipt_number"]} reversed. ' f'Restocked: {items_text}.', ) else: self.message_user( request, f'❌ Reversal failed: {result.get("error", "Unknown error")}', level='error', ) except Transaction.DoesNotExist: self.message_user(request, '❌ Transaction not found.', level='error') except Exception as e: import traceback; traceback.print_exc() self.message_user(request, f'❌ Error: {str(e)}', level='error') return redirect('admin:core_transaction_changelist') # ── Bulk reverse action (dropdown) ──────────────────────────────────── @admin.action(description='🔴 Reverse selected transaction') def action_reverse_transaction(self, request, queryset): if queryset.count() > 1: self.message_user( request, '❌ Select only ONE transaction at a time.', level='error', ) return tx = queryset.first() time_limit = timezone.now() - timedelta(hours=24) if tx.created_at < time_limit: self.message_user( request, f'❌ Cannot reverse — transaction is older than 24 hours ' f'(created {tx.created_at.strftime("%d %b %Y %H:%M")}).', level='error', ) return try: if tx.credit_sale.is_paid: self.message_user( request, '❌ Cannot reverse — this credit sale has already been paid.', level='error', ) return except Exception: pass try: with db_transaction.atomic(): result = _perform_reversal( tx, request.user, reason=f'Reversed by admin: {request.user.username}', ) if result['success']: details = result['reversal_details'] self.message_user( request, f'✅ Transaction #{details["receipt_number"]} reversed. ' f'{len(details["items_restocked"])} item(s) restocked.', ) else: self.message_user( request, f'❌ Reversal failed: {result.get("error", "Unknown error")}', level='error', ) except Exception as e: self.message_user(request, f'❌ Error: {str(e)}', level='error') # ── Summary banner ──────────────────────────────────────────────────── def changelist_view(self, request, extra_context=None): from django.db.models import Sum from decimal import Decimal from django.utils.safestring import mark_safe response = super().changelist_view(request, extra_context) try: qs = response.context_data['cl'].queryset total_sales = qs.aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') transaction_ids = qs.values_list('id', flat=True) stock_movements = StockMovement.objects.filter( transaction_id__in=transaction_ids, delta__lt=0 ) total_profit = ProfitRecord.objects.filter( stock_movement__in=stock_movements ).aggregate(total=Sum('profit'))['total'] or Decimal('0.00') total_purchase_cost = Decimal('0.00') for movement in stock_movements.select_related('product'): total_purchase_cost += ( abs(movement.delta) * (movement.product.buying_price or Decimal('0.00')) ) profit_margin = ( (total_profit / total_sales * 100) if total_sales > 0 else Decimal('0.00') ) # ── Payment method breakdown ────────────────────────────────── payment_icons = { 'Cash': '💵', 'M-Pesa Paybill': '📱', 'M-Pesa Till': '🏪', 'Bank Transfer': '🏦', 'On Credit': '📋', } breakdown_cards = '' for method, name in Transaction.PAYMENT_METHODS: method_sales = qs.filter(payment_method=method).aggregate( total=Sum('total_amount') )['total'] or Decimal('0.00') if method_sales <= 0: continue icon = payment_icons.get(name, '💳') pct = float(method_sales / total_sales * 100) if total_sales > 0 else 0 breakdown_cards += f"""

{icon} {name}

KES {method_sales:,.2f}

{pct:.1f}% of total

""" # ── Location breakdown ──────────────────────────────────────── location_cards = '' for location in Location.objects.all(): loc_txns = qs.filter(device__location=location) if not loc_txns.exists(): continue loc_sales = ( loc_txns.aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') ) loc_mvts = StockMovement.objects.filter( transaction_id__in=loc_txns.values_list('id', flat=True), delta__lt=0 ) loc_profit = ( ProfitRecord.objects.filter(stock_movement__in=loc_mvts) .aggregate(total=Sum('profit'))['total'] or Decimal('0.00') ) pct = float(loc_sales / total_sales * 100) if total_sales > 0 else 0 margin = float(loc_profit / loc_sales * 100) if loc_sales > 0 else 0 location_cards += f"""

📍 {location.name}

KES {loc_sales:,.2f}

💰 Profit: KES {loc_profit:,.2f}

{loc_txns.count()} transactions • {pct:.1f}% of total

📊 {margin:.1f}% margin

""" location_section = f"""

📍 Sales by Location:

{location_cards}

""" if location_cards else '' summary_html = f"""

📊 Transaction Summary

💰 Total Sales

KES {total_sales:,.2f}

🛒 Purchase Cost

KES {total_purchase_cost:,.2f}

📈 Total Profit

KES {total_profit:,.2f}

📊 Profit Margin

{profit_margin:.1f}%

Sales by Payment Method:

{breakdown_cards}

{location_section}

ℹ️ Statistics reflect current filtered results | Transactions: {qs.count()}

' '

📦 No items added yet.

' '➕ Add Products to Purchase' '

', add_url ) html = f'''

➕ Add More Products

''' grand_total = Decimal('0.00') for item in items: qty = abs(item.delta) unit_cost = item.product.buying_price or Decimal('0.00') line_total = qty * unit_cost grand_total += line_total html += f''' ''' html += f'''

Product	Quantity	Unit Cost	Line Total	Date Added
{item.product.name} SKU: {item.product.sku}	{qty}	KES {unit_cost:,.2f}	KES {line_total:,.2f}	{item.created_at.strftime('%Y-%m-%d %H:%M')}
TOTAL:			KES {grand_total:,.2f}

''' return mark_safe(html) items_table_display.short_description = 'Purchased Items' # ... (keep all the existing display methods: destination_display, paid_from_display, etc.) def destination_display(self, obj): return format_html( '📦 {}', obj.destination_location.name ) destination_display.short_description = 'Stock To' destination_display.admin_order_field = 'destination_location' def paid_from_display(self, obj): if obj.paid_from_location: return format_html( '💸 {}', obj.paid_from_location.name ) return '-' paid_from_display.short_description = 'Paid From' def total_amount_display(self, obj): amount = obj.total_amount or Decimal('0.00') amount_str = f"{amount:,.2f}" return format_html( 'KES {}', amount_str ) total_amount_display.short_description = 'Total' total_amount_display.admin_order_field = 'total_amount' def payment_method_display(self, obj): icons = { 'cash': '💵', 'mpesa_paybill': '📱', 'mpesa_till': '🏪', 'bank_transfer': '🏦', } icon = icons.get(obj.payment_method, '💳') return f"{icon} {obj.get_payment_method_display()}" payment_method_display.short_description = 'Payment' def payment_status_colored(self, obj): colors = {'paid': '#10b981', 'credit': '#f59e0b'} return format_html( '{}', colors.get(obj.payment_status, '#666'), obj.get_payment_status_display() ) payment_status_colored.short_description = 'Status' payment_status_colored.admin_order_field = 'payment_status' def items_summary(self, obj): count = obj.item_count() units = obj.total_units() if count > 0: return f"{count} products ({units} units)" return "No items" items_summary.short_description = 'Items' def get_urls(self): """Add custom URL for adding purchase items""" urls = super().get_urls() custom_urls = [ path( '/add-items/', self.admin_site.admin_view(self.add_items_view), name='core_purchase_additems', ), ] return custom_urls + urls def add_items_view(self, request, purchase_id): """Custom view to add items to purchase""" purchase = get_object_or_404(Purchase, pk=purchase_id) if request.method == 'POST': product_ids = request.POST.getlist('product_id') size_ids = request.POST.getlist('size_id') quantities = request.POST.getlist('quantity') with db_transaction.atomic(): total_cost = Decimal('0.00') for i in range(len(product_ids)): if product_ids[i]: product = Product.objects.get(id=product_ids[i]) size = Size.objects.get(id=size_ids[i]) if size_ids[i] else None quantity = int(quantities[i]) unit_cost = product.buying_price line_total = quantity * unit_cost total_cost += line_total StockMovement.objects.create( purchase=purchase, product=product, delta=+quantity, reason='purchase', location=purchase.destination_location, created_at=purchase.purchase_date ) stock_level, _ = StockLevel.objects.get_or_create( product=product, location=purchase.destination_location, defaults={'quantity': 0} ) if size: size_stock, _ = StockLevelSize.objects.get_or_create( stock_level=stock_level, size=size, defaults={'quantity': 0} ) size_stock.quantity += quantity size_stock.save() stock_level.update_total_quantity() # Update purchase total current_total = purchase.total_amount or Decimal('0.00') purchase.total_amount = current_total + total_cost purchase.save() if purchase.payment_status == 'paid': self._process_payment(request, purchase, total_cost) messages.success(request, f'Added items to purchase. New items total: KES {total_cost:,.2f}') return redirect('admin:core_purchase_change', purchase.pk) products = Product.objects.all().order_by('name') sizes = Size.objects.all().order_by('name') context = { 'purchase': purchase, 'products': products, 'sizes': sizes, 'opts': self.model._meta, 'has_view_permission': self.has_view_permission(request), 'site_title': 'Django admin', 'site_header': 'Django administration', } return render(request, 'admin/core/purchase_add_items.html', context) def _process_payment(self, request, purchase, amount): """Deduct cash from global bank for new items""" paid_from = purchase.paid_from_location or purchase.destination_location cash_balance, _ = LocationCashBalance.objects.get_or_create( location=paid_from, payment_method=purchase.payment_method, defaults={ 'balance': Decimal('0.00'), 'total_sales': Decimal('0.00'), 'total_purchases': Decimal('0.00') } ) if cash_balance.balance < amount: messages.warning( request, f"Warning: Insufficient balance in {paid_from.name}. Balance will go negative." ) cash_balance.balance -= amount cash_balance.total_purchases += amount cash_balance.save() CashTransaction.objects.create( location=paid_from, payment_method=purchase.payment_method, transaction_type='purchase', amount=amount, description=f"Purchase items added from {purchase.supplier_name} (Stock to: {purchase.destination_location.name})", reference_number=purchase.reference_number or '', balance_after=cash_balance.balance, created_by=request.user, created_at=purchase.purchase_date ) def save_model(self, request, obj, form, change): """Save purchase with initial total_amount of 0""" if not change: obj.created_by = request.user if not obj.purchase_date: obj.purchase_date = timezone.now() obj.total_amount = Decimal('0.00') super().save_model(request, obj, form, change) def response_add(self, request, obj, post_url_continue=None): """After creating purchase, redirect to add items""" messages.success( request, f'Purchase "{obj.supplier_name}" created successfully. ' f'Now add items to this purchase.' ) return HttpResponseRedirect( reverse('admin:core_purchase_additems', args=[obj.pk]) ) # ... (keep the changelist_view method with summary stats) def changelist_view(self, request, extra_context=None): """Add purchase summary statistics""" response = super().changelist_view(request, extra_context) try: qs = response.context_data['cl'].queryset total_purchases = qs.aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') total_paid = qs.filter(payment_status='paid').aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') total_credit = qs.filter(payment_status='credit').aggregate(total=Sum('total_amount'))['total'] or Decimal('0.00') by_location = {} for location in Location.objects.all(): loc_total = qs.filter(destination_location=location).aggregate( total=Sum('total_amount') )['total'] or Decimal('0.00') if loc_total > 0: by_location[location.name] = loc_total loc_cards = "" for loc_name, amount in by_location.items(): loc_cards += f"""

📦 {loc_name}

KES {amount:,.2f}

""" summary_html = f"""

🛒 Purchase Summary

💰 Total Purchases

KES {total_purchases:,.2f}

✅ Paid

KES {total_paid:,.2f}

⏳ On Credit

KES {total_credit:,.2f}

Stock Delivered To:

{loc_cards}

' '{}
' '{} {}' '

', obj.from_location.name, icon, obj.get_from_payment_method_display() ) from_display.short_description = 'From' def arrow_display(self, obj): return mark_safe( '→' ) arrow_display.short_description = '' def to_display(self, obj): icons = { 'cash': '💵', 'mpesa_paybill': '📱', 'mpesa_till': '🏪', 'bank_transfer': '🏦', } icon = icons.get(obj.to_payment_method, '💳') return format_html( '

' '{}
' '{} {}' '

', obj.to_location.name, icon, obj.get_to_payment_method_display() ) to_display.short_description = 'To' def amount_display(self, obj): amount_str = f'{obj.amount:,.2f}' return format_html( 'KES {}', amount_str ) amount_display.short_description = 'Amount' amount_display.admin_order_field = 'amount' def transfer_type_display(self, obj): icons = { 'method_change': '🔄', 'location_transfer': '🚚', 'consolidation': '🏦', 'float': '💸', 'other': '📝', } icon = icons.get(obj.transfer_type, '📝') return f"{icon} {obj.get_transfer_type_display()}" transfer_type_display.short_description = 'Type' def save_model(self, request, obj, form, change): """Process cash transfer""" if not change: obj.created_by = request.user if obj.from_location == obj.to_location and obj.from_payment_method == obj.to_payment_method: messages.error(request, "Cannot transfer to the same account!") return source_balance = LocationCashBalance.objects.filter( location=obj.from_location, payment_method=obj.from_payment_method ).first() if source_balance and source_balance.balance < obj.amount: messages.warning( request, f"Warning: Insufficient balance. Available: KES {source_balance.balance:,.2f}, " f"Transfer: KES {obj.amount:,.2f}. Balance will go negative." ) super().save_model(request, obj, form, change) if not change: self._process_transfer(request, obj) def _process_transfer(self, request, transfer): """Execute the cash transfer""" with db_transaction.atomic(): source_balance, _ = LocationCashBalance.objects.get_or_create( location=transfer.from_location, payment_method=transfer.from_payment_method, defaults={'balance': Decimal('0.00'), 'total_sales': Decimal('0.00'), 'total_purchases': Decimal('0.00')} ) source_balance.balance -= transfer.amount source_balance.save() CashTransaction.objects.create( location=transfer.from_location, payment_method=transfer.from_payment_method, transaction_type='transfer_out', amount=transfer.amount, description=f"Transfer to {transfer.to_location.name} ({transfer.get_to_payment_method_display()})", reference_number=transfer.reference_number, balance_after=source_balance.balance, created_by=request.user ) dest_balance, _ = LocationCashBalance.objects.get_or_create( location=transfer.to_location, payment_method=transfer.to_payment_method, defaults={'balance': Decimal('0.00'), 'total_sales': Decimal('0.00'), 'total_purchases': Decimal('0.00')} ) dest_balance.balance += transfer.amount dest_balance.save() CashTransaction.objects.create( location=transfer.to_location, payment_method=transfer.to_payment_method, transaction_type='transfer_in', amount=transfer.amount, description=f"Transfer from {transfer.from_location.name} ({transfer.get_from_payment_method_display()})", reference_number=transfer.reference_number, balance_after=dest_balance.balance, created_by=request.user ) messages.success( request, f"Transfer completed: KES {transfer.amount:,.2f} moved from " f"{transfer.from_location.name} to {transfer.to_location.name}" ) def changelist_view(self, request, extra_context=None): """Add transfer summary""" response = super().changelist_view(request, extra_context) try: qs = response.context_data['cl'].queryset # Total amount transferred total_transferred = qs.aggregate(total=Sum('amount'))['total'] or Decimal('0.00') # Count transfers by type type_breakdown = {} for type_code, type_name in CashTransfer.TRANSFER_TYPES: type_count = qs.filter(transfer_type=type_code).count() if type_count > 0: type_breakdown[type_name] = type_count # Breakdown cards breakdown_cards = "" type_icons = { 'Payment Method Change': '🔄', 'Location Transfer': '🚚', 'Consolidation/Banking': '🏦', 'Float Allocation': '💸', 'Other': '📝', } for type_name, count in type_breakdown.items(): icon = type_icons.get(type_name, '📝') breakdown_cards += f"""

{icon} {type_name}

{count} transfers

""" summary_html = f"""

💸 Cash Transfer Summary

Total Transfers

{qs.count()}

Amount Moved

KES {total_transferred:,.2f}

{f'''

Breakdown by Transfer Type:

{breakdown_cards}

''' if breakdown_cards else ''}

ℹ️ Cash movement between locations and payment methods

{icon} {cat['category__name']}

KES {cat['total']:,.2f}

""" summary_html = f"""

💸 Expense Summary

💸 Total Expenses

KES {total_expenses:,.2f}

💼 Business

KES {business_total:,.2f}

👤 Personal

KES {personal_total:,.2f}

{f'''

Top Expense Categories:

{category_cards}

''' if category_cards else ''}

ℹ️ All expenses deducted from MainStore Bank | Count: {qs.count()}

""" extra_context = extra_context or {} extra_context['summary_stats'] = mark_safe(summary_html) response.context_data.update(extra_context) except (AttributeError, KeyError): pass return response @admin.register(SalesTarget) class SalesTargetAdmin(admin.ModelAdmin): list_display = [ 'location', 'period', 'target_display', 'current_sales_display', 'progress_bar', 'bonus_display', 'is_active' ] list_filter = ['location', 'period', 'is_active'] search_fields = ['location__name'] readonly_fields = ['created_at', 'updated_at'] fieldsets = ( ('Target Information', { 'fields': ('location', 'target_amount', 'bonus_percentage', 'period') }), ('Date Range (Optional)', { 'fields': ('start_date', 'end_date'), 'classes': ('collapse',), 'description': 'Leave empty to use period-based calculation' }), ('Status', { 'fields': ('is_active', 'created_at', 'updated_at') }) ) def target_display(self, obj): return format_html( 'KES {}', f'{obj.target_amount:,.0f}' ) target_display.short_description = 'Target' def current_sales_display(self, obj): sales = obj.get_current_sales() color = '#10b981' if sales >= obj.target_amount else '#f59e0b' return format_html( 'KES {}', color, f'{sales:,.0f}' ) current_sales_display.short_description = 'Current Sales' def progress_bar(self, obj): percentage = float(obj.get_progress_percentage() or 0) # ensure float color = '#10b981' if percentage >= 100 else '#f59e0b' return format_html( '''

{}%

''', min(percentage, 100), # width color, # bar color f"{percentage:.1f}" # formatted percentage as string ) def bonus_display(self, obj): bonus = obj.calculate_bonus() if bonus > 0: return format_html( '💰 KES {}', f'{bonus:,.0f}' ) return mark_safe('Not yet') bonus_display.short_description = 'Bonus Earned' @admin.register(BonusRecord) class BonusRecordAdmin(admin.ModelAdmin): list_display = [ 'location', 'period_display', 'target_amount', 'actual_sales', 'bonus_amount_display', 'paid_status', 'created_at' ] list_filter = ['location', 'paid', 'period_start'] readonly_fields = ['created_at'] def period_display(self, obj): return f"{obj.period_start} to {obj.period_end}" period_display.short_description = 'Period' def bonus_amount_display(self, obj): return format_html( 'KES {}', f'{obj.bonus_amount:,.0f}' ) bonus_amount_display.short_description = 'Bonus' def paid_status(self, obj): if obj.paid: return format_html( '✅ Paid' ) return format_html( '⏳ Pending' ) paid_status.short_description = 'Status' # ✅ Custom admin view for transaction reversal @staff_member_required def transaction_reversal_admin_view(request): """ Admin-only page for reversing transactions """ context = { 'title': 'Transaction Reversal', 'site_header': 'Seemwork Traders Admin', 'has_permission': True, } return render(request, 'admin/transaction_reversal.html', context) # ✅ Custom AdminSite to add reversal link to admin menu class CustomAdminSite(admin.AdminSite): site_header = 'Seemwork Traders Administration' site_title = 'Seemwork Traders Admin' index_title = 'Welcome to Seemwork Traders Admin' def get_app_list(self, request): """ Add custom links to the admin index """ app_list = super().get_app_list(request) # Add our custom "Transaction Reversal" link app_list += [ { 'name': 'Operations', 'app_label': 'operations', 'models': [ { 'name': 'Transaction Reversal', 'object_name': 'reversal', 'admin_url': '/admin/reversal/', 'view_only': True, 'perms': {'view': True}, } ], } ] return app_list # ✅ Replace default admin site admin_site = CustomAdminSite(name='custom_admin') @admin.register(CreditSale) class CreditSaleAdmin(admin.ModelAdmin): list_display = [ 'customer_name', 'customer_phone', 'amount_display', 'due_date', 'status_badge', 'due_info', 'location_display', 'paid_via', 'confirmed_by', 'created_at', ] list_filter = ['is_paid', 'paid_payment_method', 'due_date'] search_fields = [ 'customer_name', 'customer_phone', 'transaction__id', 'notes', ] readonly_fields = [ 'transaction', 'is_overdue_display', 'created_at', 'confirmed_by', 'paid_at', ] date_hierarchy = 'due_date' actions = [ 'action_mark_paid_cash', 'action_mark_paid_mpesa_till', 'action_mark_paid_mpesa_paybill', 'action_mark_paid_bank', ] fieldsets = ( ('Customer', { 'fields': ('customer_name', 'customer_phone'), }), ('Credit Details', { 'fields': ('transaction', 'due_date', 'is_overdue_display', 'notes'), }), ('Payment Confirmation', { 'fields': ( 'is_paid', 'paid_at', 'paid_payment_method', 'confirmed_by', ), 'description': ( 'Use the bulk actions below the list to mark as paid. ' 'Editing "is_paid" directly here does NOT update cash balances.' ), }), ('Metadata', { 'fields': ('created_at',), 'classes': ('collapse',), }), ) def get_queryset(self, request): return super().get_queryset(request).select_related( 'transaction', 'transaction__device', 'transaction__device__location', 'confirmed_by', ) # ── Column display methods ──────────────────────────────────────────────── def amount_display(self, obj): return format_html( 'KES {}', f'{obj.transaction.total_amount:,.2f}' ) amount_display.short_description = 'Amount' amount_display.admin_order_field = 'transaction__total_amount' def status_badge(self, obj): if obj.is_paid: return mark_safe( '' '✅ PAID' ) if obj.is_overdue: return mark_safe( '' '🚨 OVERDUE' ) return mark_safe( '' '⏳ PENDING' ) status_badge.short_description = 'Status' def due_info(self, obj): if obj.is_paid: paid_str = obj.paid_at.strftime('%d %b %Y') if obj.paid_at else '—' return format_html( 'Paid {}', paid_str ) if obj.is_overdue: return format_html( '' '{} day{} overdue', obj.days_overdue, 's' if obj.days_overdue != 1 else '', ) days = obj.days_until_due color = '#f59e0b' if days <= 3 else '#2d3748' return format_html( 'Due in {} day{}', color, days, 's' if days != 1 else '', ) due_info.short_description = 'Due Info' def location_display(self, obj): tx = obj.transaction if tx.device and tx.device.location: return tx.device.location.name return '—' location_display.short_description = 'Location' def paid_via(self, obj): if not obj.is_paid or not obj.paid_payment_method: return '—' icons = { 'cash': '💵', 'mpesa_paybill': '📱', 'mpesa_till': '🏪', 'bank_transfer': '🏦', } icon = icons.get(obj.paid_payment_method, '💳') return f"{icon} {obj.get_paid_payment_method_display()}" paid_via.short_description = 'Paid Via' def is_overdue_display(self, obj): if obj.is_overdue: return format_html( '🚨 Yes — {} day{} overdue', obj.days_overdue, 's' if obj.days_overdue != 1 else '', ) if obj.is_paid: return mark_safe('✅ Paid') return mark_safe('No') is_overdue_display.short_description = 'Overdue?' # ── Bulk mark-paid actions ──────────────────────────────────────────────── def _bulk_mark_paid(self, request, queryset, payment_method): from .models import LocationCashBalance, CashTransaction updated = 0 skipped = 0 for credit in queryset: if credit.is_paid: skipped += 1 continue with db_transaction.atomic(): credit.is_paid = True credit.paid_at = timezone.now() credit.paid_payment_method = payment_method credit.confirmed_by = request.user credit.save() location = credit.transaction.get_location() if location: amount = credit.transaction.total_amount cash_balance, _ = LocationCashBalance.objects.get_or_create( location=location, payment_method=payment_method, defaults={ 'balance': Decimal('0.00'), 'total_sales': Decimal('0.00'), 'total_purchases': Decimal('0.00'), } ) cash_balance.balance += amount cash_balance.total_sales += amount cash_balance.save() CashTransaction.objects.create( location=location, payment_method=payment_method, transaction_type='sale', amount=amount, sale_transaction=credit.transaction, balance_after=cash_balance.balance, created_by=request.user, description=( f"Credit payment received — " f"{credit.customer_name} ({credit.customer_phone}) " f"| Receipt #{credit.transaction.server_receipt_number}" ), ) updated += 1 method_label = payment_method.replace('_', ' ').title() msg = f'✅ {updated} credit(s) marked as paid via {method_label}.' if skipped: msg += f' {skipped} already-paid record(s) skipped.' self.message_user(request, msg) @admin.action(description='✅ Mark as Paid — Cash') def action_mark_paid_cash(self, request, queryset): self._bulk_mark_paid(request, queryset, 'cash') @admin.action(description='✅ Mark as Paid — M-Pesa Till') def action_mark_paid_mpesa_till(self, request, queryset): self._bulk_mark_paid(request, queryset, 'mpesa_till') @admin.action(description='✅ Mark as Paid — M-Pesa Paybill') def action_mark_paid_mpesa_paybill(self, request, queryset): self._bulk_mark_paid(request, queryset, 'mpesa_paybill') @admin.action(description='✅ Mark as Paid — Bank Transfer') def action_mark_paid_bank(self, request, queryset): self._bulk_mark_paid(request, queryset, 'bank_transfer') # ── Summary banner ──────────────────────────────────────────────────────── def changelist_view(self, request, extra_context=None): response = super().changelist_view(request, extra_context) try: from .models import CreditSale from django.db.models import Sum all_credits = CreditSale.objects.select_related('transaction') unpaid_credits = all_credits.filter(is_paid=False) total_outstanding = ( unpaid_credits .aggregate(t=Sum('transaction__total_amount'))['t'] or Decimal('0.00') ) total_collected = ( all_credits.filter(is_paid=True) .aggregate(t=Sum('transaction__total_amount'))['t'] or Decimal('0.00') ) # Python-side overdue calculation (no DB annotation for properties) overdue_credits = [c for c in unpaid_credits if c.is_overdue] overdue_count = len(overdue_credits) overdue_amount = sum(c.transaction.total_amount for c in overdue_credits) due_soon = [ c for c in unpaid_credits if not c.is_overdue and 0 <= c.days_until_due <= 3 ] due_soon_count = len(due_soon) overdue_banner = '' if overdue_count: overdue_banner = ( f'

' f'🚨 {overdue_count} overdue customer(s) — ' f'KES {overdue_amount:,.2f} — call them today!

' ) due_soon_banner = '' if due_soon_count: due_soon_banner = ( f'

' f'⚠️ {due_soon_count} payment(s) due within 3 days

' ) summary_html = f"""

📋 Credit Sales Overview

⏳ Outstanding

KES {total_outstanding:,.2f}

{unpaid_credits.count()} customer(s)

🚨 Overdue

KES {overdue_amount:,.2f}

{overdue_count} customer(s)

✅ Collected

KES {total_collected:,.2f}

{overdue_banner} {due_soon_banner}

{cat_labels.get(row['category'], row['category'])} KES {row['total']:,.0f}

""" summary_html = f"""

🚐 Matatu Income Summary

💰 Income (filtered)

KES {total_income:,.2f}

{count} entr{'y' if count==1 else 'ies'}

📊 Avg per Entry

KES {avg:,.2f}

📉 All-time Expenses

KES {all_expenses:,.2f}

{net_label}

KES {abs(net):,.2f}

{f'''

📋 All-time Expense Breakdown:

{cat_rows}

''' if cat_rows else ''}

ℹ️ KPI cards show all-time totals • Income table shows filtered results

""" extra_context = extra_context or {} extra_context['summary_stats'] = mark_safe(summary_html) response.context_data.update(extra_context) except (AttributeError, KeyError): pass return response @admin.register(MatatuExpense) class MatatuExpenseAdmin(admin.ModelAdmin): list_display = ['date', 'category_display', 'amount_display', 'description_short', 'created_by'] list_filter = ['category', 'date'] search_fields = ['description'] date_hierarchy = 'date' readonly_fields = ['created_at', 'created_by'] CATEGORY_ICONS = { 'fuel': '⛽', 'repair': '🔧', 'police': '👮', 'parking': '🅿️', 'insurance': '📋', 'sacco': '🏦', 'driver': '👷', 'service': '🛠️', 'other': '📝', } def category_display(self, obj): icon = self.CATEGORY_ICONS.get(obj.category, '📝') return f"{icon} {obj.get_category_display()}" category_display.short_description = 'Category' category_display.admin_order_field = 'category' def amount_display(self, obj): return format_html( 'KES {}', f'{obj.amount:,.2f}' ) amount_display.short_description = 'Amount' amount_display.admin_order_field = 'amount' def description_short(self, obj): if len(obj.description) > 60: return obj.description[:60] + '...' return obj.description or '—' description_short.short_description = 'Description' def save_model(self, request, obj, form, change): if not change: obj.created_by = request.user super().save_model(request, obj, form, change) def changelist_view(self, request, extra_context=None): from django.db.models import Sum from decimal import Decimal response = super().changelist_view(request, extra_context) try: qs = response.context_data['cl'].queryset # ── Totals for current filter ───────────────────────────────── total_expenses = qs.aggregate(t=Sum('amount'))['t'] or Decimal('0.00') count = qs.count() # ── All-time for net ────────────────────────────────────────── all_income = MatatuIncome.objects.aggregate(t=Sum('amount'))['t'] or Decimal('0.00') all_expenses = MatatuExpense.objects.aggregate(t=Sum('amount'))['t'] or Decimal('0.00') net = all_income - all_expenses net_color = '#10b981' if net >= 0 else '#ef4444' net_label = 'NET PROFIT' if net >= 0 else 'NET LOSS' # ── Category breakdown for FILTERED queryset ────────────────── cat_breakdown = ( qs.values('category') .annotate(total=Sum('amount')) .order_by('-total') ) cat_labels = dict(MatatuExpense.CATEGORY_CHOICES) cat_cards = '' for row in cat_breakdown: icon = self.CATEGORY_ICONS.get(row['category'], '📝') pct = float(row['total'] / total_expenses * 100) if total_expenses > 0 else 0 cat_cards += f"""

{icon} {cat_labels.get(row['category'], row['category'])}

KES {row['total']:,.2f}

{pct:.1f}% of filtered total

""" summary_html = f"""

🚐 Matatu Expense Summary

📉 Expenses (filtered)

KES {total_expenses:,.2f}

{count} entr{'y' if count==1 else 'ies'}

💰 All-time Income

KES {all_income:,.2f}

📊 All-time Expenses

KES {all_expenses:,.2f}

{net_label}

KES {abs(net):,.2f}

{f'''

Breakdown by Category (filtered):

{cat_cards}

''' if cat_cards else ''}

ℹ️ Use the date hierarchy and filters on the right to drill down by month/week

""" extra_context = extra_context or {} extra_context['summary_stats'] = mark_safe(summary_html) response.context_data.update(extra_context) except (AttributeError, KeyError): pass return response