""" Extended accounting views: account types, GL, mappings, manual journals, trial balance, general ledger report, enhanced P&L and Balance Sheet. """ from collections import defaultdict, OrderedDict from datetime import date as date_cls from datetime import datetime from rest_framework.decorators import api_view, authentication_classes, permission_classes from rest_framework.authentication import TokenAuthentication from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from django.db import IntegrityError from django.db.models import Sum from system_administration.utils import check_if_system_administrator from system_administration.models import CompanyProfile from human_resource.models import StaffProfile from .models import ( AccountType, AccountTypeDetail, ChartOfAccount, AccountHistory, AccountMapping, ManualJournalEntry, ManualJournalLine, AccountingPeriod, AccountingSettings, ) from . import financial_reporting as fr from .journal_posting import ( PostingError, cancel_journal_draft, post_journal_entry_to_ledger, reverse_journal_entry, ) from .accounting_seed import ( ensure_account_types, ensure_branch_default_coa, seed_account_types, seed_default_chart_of_accounts, ) def _auth_error_response(payload, exc: ValueError): msg = str(exc) status = 401 if msg == "staff_profile required" else 400 body = {"message": "false", "payload": payload} if msg: body["payload"] = {**payload, "error": msg} if isinstance(payload, dict) else {"error": msg} return Response(body, status=status) def _finance_auth(request): """Return (company_profile, staff_profile) or raise ValueError on auth failure.""" serial = request.data.get("serial_number") if not serial: raise ValueError("serial_number required") company_profile = CompanyProfile.objects.get(company_serial_number=serial) staff_profile = StaffProfile.objects.filter(user=request.user).select_related( 'company_department', 'company_branch', 'staff_position', ).first() if not staff_profile: raise ValueError("staff_profile required") return company_profile, staff_profile def _department_name(staff_profile) -> str: dept = getattr(staff_profile, "company_department", None) return dept.department_name if dept else "" def _require_finance_staff(staff_profile): if check_if_system_administrator(staff_profile): return True return _department_name(staff_profile) == "finance_and_accounting" def _require_company_access(company_profile, staff_profile): if check_if_system_administrator(staff_profile): return True branch = getattr(staff_profile, "company_branch", None) if branch and branch.company_profile_id == company_profile.id: return True return False def _require_finance_write(staff_profile): if check_if_system_administrator(staff_profile): return True return ( _department_name(staff_profile) == "finance_and_accounting" and staff_profile.has_read_write_priviledges ) def _serialize_account_types(): ensure_account_types() account_types = AccountType.objects.all().order_by("name") return [ { "id": at.id, "name": at.name, "type_key": at.type_key, "financial_statement": at.financial_statement, "normal_balance": at.normal_balance, "details": [ {"id": d.id, "name": d.name, "description": d.description} for d in at.type_details.all().order_by("name") ], } for at in account_types ] def _parse_finance_date(value): return fr.parse_finance_date(value) def _filter_by_date_range(qs, request, field_name, from_key="date_from", to_key="date_to"): if request.data.get(from_key): qs = qs.filter(**{f"{field_name}__gte": _parse_finance_date(request.data[from_key])}) if request.data.get(to_key): qs = qs.filter(**{f"{field_name}__lte": _parse_finance_date(request.data[to_key])}) return qs def _ledger_amount(value): return float(value or 0) BANK_DETAIL_TYPE_NAMES = frozenset({ "bank account", "savings account", "foreign currency bank account", }) def _is_bank_gl_account(acc): """Bank register rows belong under Banking, not Chart of Accounts.""" if getattr(acc, "is_mpesa_account", False): return True if acc.account_type_id and acc.account_type.type_key == "bank": return True if acc.account_detail_type_id: detail_name = (acc.account_detail_type.name or "").strip().lower() if detail_name in BANK_DETAIL_TYPE_NAMES: return True if not acc.account_type_id and (acc.banking_institution_name or "").strip(): return True return False def _account_primary_balance(acc): agg = AccountHistory.objects.filter(account=acc).aggregate( dr=Sum("debit"), cr=Sum("credit"), ) debit = _ledger_amount(agg["dr"]) credit = _ledger_amount(agg["cr"]) normal = acc.account_type.normal_balance if acc.account_type_id else "debit" if normal == "credit": return round(credit - debit, 2) return round(debit - credit, 2) def _serialize_gl_account(acc): at = acc.account_type return { "id": acc.id, "account_name": acc.account_name, "account_number": acc.account_number, "account_descriptions": acc.account_descriptions, "account_type_id": at.id if at else None, "account_type_key": at.type_key if at else "", "account_type_name": at.name if at else "", "financial_statement": at.financial_statement if at else "", "normal_balance": at.normal_balance if at else "debit", "account_detail_type_id": acc.account_detail_type_id, "account_detail_type_name": acc.account_detail_type.name if acc.account_detail_type else "", "parent_account_id": acc.parent_account_id, "parent_account_name": acc.parent_account.account_name if acc.parent_account else "", "opening_balance": str(acc.opening_balance), "opening_balance_date": str(acc.opening_balance_date) if acc.opening_balance_date else "", "running_balance": acc.running_balance, "primary_balance": _account_primary_balance(acc), "account_currency": acc.account_currency, "is_default_account": acc.is_default_account, "is_active": acc.is_active, "is_mpesa_account": acc.is_mpesa_account, "banking_institution_name": acc.banking_institution_name, "bank_branch_name": acc.bank_branch_name, "bank_branch_code": acc.bank_branch_code, "bank_swift_code": acc.bank_swift_code, "bank_sort_code": acc.bank_sort_code, } # ============================================================================= # ACCOUNT TYPE MANAGEMENT # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_account_types(request): """Return all account types with their sub-type details (global reference data).""" payload = {} try: if not request.user.is_authenticated: return Response({"message": "false", "payload": payload}, status=401) bootstrap = ensure_account_types() if bootstrap.get("bootstrapped"): payload["bootstrapped_account_types"] = True payload["account_types"] = _serialize_account_types() return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def setup_default_account_types(request): """Seed the 16 standard account types. Safe to call multiple times.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) payload.update(seed_account_types()) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def setup_default_chart_of_accounts(request): """ Seed enterprise COA (1000–7999) and standard posting rules for the branch. Requires account types to exist. Safe to call multiple times (skips existing numbers). """ payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) branch = staff_profile.company_branch if not branch: return Response({ "message": "false", "payload": {"error": "Staff profile has no company branch."}, }, status=400) payload.update(seed_default_chart_of_accounts(branch, staff_profile)) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # ENHANCED CHART OF ACCOUNTS # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_chart_of_accounts_enhanced(request): """Return GL accounts with type/hierarchy data. Excludes bank register rows by default.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_company_access(company_profile, staff_profile): return Response({"message": "false", "payload": payload}, status=401) exclude_banks = request.data.get("exclude_banks", True) branch = staff_profile.company_branch ensure_account_types() payload["account_types"] = _serialize_account_types() if branch: coa_bootstrap = ensure_branch_default_coa(branch, staff_profile) if coa_bootstrap and coa_bootstrap.get("bootstrapped"): payload["bootstrapped_coa"] = True accounts = ChartOfAccount.objects.filter( recycle_bin=False, ) if branch: accounts = accounts.filter(company_branch=branch) accounts = accounts.select_related( "account_type", "account_detail_type", "parent_account" ).order_by("account_number") accounts_list = [] for acc in accounts: if exclude_banks and _is_bank_gl_account(acc): continue accounts_list.append(_serialize_gl_account(acc)) grouped = {} for acc in accounts_list: key = acc["account_type_name"] or "Uncategorised" if key not in grouped: grouped[key] = { "account_type_name": key, "account_type_key": acc["account_type_key"], "financial_statement": acc["financial_statement"], "normal_balance": acc["normal_balance"], "accounts": [], } grouped[key]["accounts"].append(acc) payload["accounts"] = accounts_list payload["grouped_accounts"] = list(grouped.values()) payload["total_accounts"] = len(accounts_list) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_account_enhanced(request): """Create a GL account with full type and hierarchy support.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data if not data.get("account_type_id"): return Response({"message": "false", "payload": {"error": "Account type is required."}}, status=400) if not data.get("account_name", "").strip(): return Response({"message": "false", "payload": {"error": "Account name is required."}}, status=400) if not data.get("account_number", "").strip(): return Response({"message": "false", "payload": {"error": "Account number is required."}}, status=400) account_type = AccountType.objects.get(id=data["account_type_id"]) if account_type and account_type.type_key == "bank": return Response({"message": "false", "payload": {"error": "Bank accounts belong under Banking."}}, status=400) account_detail_type = AccountTypeDetail.objects.get(id=data["account_detail_type_id"]) if data.get("account_detail_type_id") else None parent_account = ChartOfAccount.objects.get(id=data["parent_account_id"]) if data.get("parent_account_id") else None account = ChartOfAccount.objects.create( company_branch=staff_profile.company_branch, account_type=account_type, account_detail_type=account_detail_type, parent_account=parent_account, account_name=data["account_name"], account_number=data["account_number"], account_descriptions=data.get("account_descriptions", ""), opening_balance=data.get("opening_balance", 0), opening_balance_date=data.get("opening_balance_date") or None, account_currency=data.get("account_currency", "kes"), banking_institution_name=data.get("banking_institution_name", ""), bank_branch_name=data.get("bank_branch_name", ""), bank_branch_code=data.get("bank_branch_code", ""), bank_swift_code=data.get("bank_swift_code", ""), bank_sort_code=data.get("bank_sort_code", ""), is_mpesa_account=bool(data.get("is_mpesa_account", False)), created_by=staff_profile, last_updated_by=staff_profile, ) ob = float(data.get("opening_balance", 0)) if ob != 0: ob_date = date_cls.fromisoformat(data["opening_balance_date"]) if data.get("opening_balance_date") else date_cls.today() normal_bal = account_type.normal_balance if account_type else "debit" AccountHistory.objects.create( account=account, debit=ob if normal_bal == "debit" else 0, credit=ob if normal_bal == "credit" else 0, description="Opening Balance", transaction_date=ob_date, rel_type="opening_balance", company_branch=staff_profile.company_branch, created_by=staff_profile, ) payload["account_id"] = account.id payload["account_number"] = account.account_number return Response({"message": "true", "payload": payload}, status=200) except AccountType.DoesNotExist: return Response({"message": "false", "payload": {"error": "Invalid account type."}}, status=400) except IntegrityError: return Response({"message": "false", "payload": {"error": "Account number already exists."}}, status=400) except ValueError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": {"error": "Failed to create account."}}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_account_enhanced(request): """Update account type, detail type, name, descriptions etc.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data account = ChartOfAccount.objects.get(id=data["account_id"], recycle_bin=False) if data.get("account_type_id"): account.account_type = AccountType.objects.get(id=data["account_type_id"]) if account.account_type.type_key == "bank": return Response({"message": "false", "payload": {"error": "Bank accounts belong under Banking."}}, status=400) if data.get("account_detail_type_id"): account.account_detail_type = AccountTypeDetail.objects.get(id=data["account_detail_type_id"]) if "parent_account_id" in data: account.parent_account = ChartOfAccount.objects.get(id=data["parent_account_id"]) if data["parent_account_id"] else None if "is_active" in data: account.is_active = bool(data["is_active"]) for field in ["account_name", "account_descriptions", "account_currency"]: if field in data: setattr(account, field, data[field]) account.last_updated_by = staff_profile account.save() payload["account_id"] = account.id return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # BANK ACCOUNTS (operational register — ChartOfAccount bank rows) # ============================================================================= def _branch_chart_accounts(staff_profile): qs = ChartOfAccount.objects.filter(recycle_bin=False) branch = staff_profile.company_branch if branch: qs = qs.filter(company_branch=branch) return qs.select_related("account_type", "account_detail_type", "parent_account") def _serialize_gl_link_option(acc): return { "id": acc.id, "account_name": acc.account_name, "account_number": acc.account_number, "account_type_name": acc.account_type.name if acc.account_type_id else "", "primary_balance": _account_primary_balance(acc), } def _serialize_bank_account(acc): linked = acc.parent_account if acc.parent_account_id else acc balance_source = linked if acc.parent_account_id else acc return { "id": acc.id, "account_name": acc.account_name, "account_number": acc.account_number, "bank_name": acc.banking_institution_name or ("M-Pesa" if acc.is_mpesa_account else ""), "bank_branch": acc.bank_branch_name or "", "bank_branch_code": acc.bank_branch_code or "", "bank_swift_code": acc.bank_swift_code or "", "bank_sort_code": acc.bank_sort_code or "", "currency": acc.account_currency or "kes", "is_mpesa": bool(acc.is_mpesa_account), "is_default": bool(acc.is_default_account), "is_active": bool(acc.is_active), "gl_account_id": linked.id, "gl_account_number": linked.account_number, "gl_account_name": linked.account_name, "gl_balance": _account_primary_balance(balance_source), } def _bank_account_type(): ensure_account_types() return AccountType.objects.get(type_key="bank") def _bank_detail_type(account_type, prefer_mpesa=False): if prefer_mpesa: detail = AccountTypeDetail.objects.filter( account_type=account_type, name__icontains="m-pesa", ).first() if detail: return detail detail = AccountTypeDetail.objects.filter( account_type=account_type, name__iexact="Bank Account", ).first() if detail: return detail return AccountTypeDetail.objects.filter(account_type=account_type).first() @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_bank_accounts(request): """Return operational bank register rows and linkable GL cash/bank accounts.""" payload = {"bank_accounts": [], "gl_accounts": []} try: company_profile, staff_profile = _finance_auth(request) if not _require_company_access(company_profile, staff_profile): return Response({"message": "false", "payload": payload}, status=401) bank_accounts = [] gl_accounts = [] seen_gl_ids = set() for acc in _branch_chart_accounts(staff_profile).order_by("account_number"): if _is_bank_gl_account(acc): bank_accounts.append(_serialize_bank_account(acc)) if acc.id not in seen_gl_ids: gl_accounts.append(_serialize_gl_link_option(acc)) seen_gl_ids.add(acc.id) continue type_key = acc.account_type.type_key if acc.account_type_id else "" if type_key in ("cash_and_cash_equivalents", "bank"): if acc.id not in seen_gl_ids: gl_accounts.append(_serialize_gl_link_option(acc)) seen_gl_ids.add(acc.id) payload["bank_accounts"] = bank_accounts payload["gl_accounts"] = gl_accounts return Response({"message": "true", "payload": payload}, status=200) except ValueError as e: return _auth_error_response(payload, e) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_bank_account(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data account_name = (data.get("account_name") or "").strip() account_number = (data.get("account_number") or "").strip() if not account_name: return Response({"message": "false", "payload": {"error": "Account name is required."}}, status=400) if not account_number: return Response({"message": "false", "payload": {"error": "Account number is required."}}, status=400) is_mpesa = bool(data.get("is_mpesa", False)) bank_type = _bank_account_type() parent_account = None gl_account_id = data.get("gl_account_id") if gl_account_id: parent_account = ChartOfAccount.objects.filter( id=gl_account_id, recycle_bin=False, ).first() if not parent_account: return Response({"message": "false", "payload": {"error": "Linked GL account not found."}}, status=400) if data.get("is_default"): _branch_chart_accounts(staff_profile).filter(is_default_account=True).update(is_default_account=False) account = ChartOfAccount.objects.create( company_branch=staff_profile.company_branch, account_type=bank_type, account_detail_type=_bank_detail_type(bank_type, prefer_mpesa=is_mpesa), parent_account=parent_account, account_name=account_name, account_number=account_number, account_currency=(data.get("currency") or "kes").lower(), banking_institution_name="M-Pesa" if is_mpesa else (data.get("bank_name") or "").strip(), bank_branch_name=(data.get("bank_branch") or "").strip(), bank_branch_code=(data.get("bank_branch_code") or "").strip(), bank_swift_code=(data.get("bank_swift_code") or "").strip(), bank_sort_code=(data.get("bank_sort_code") or "").strip(), is_mpesa_account=is_mpesa, is_default_account=bool(data.get("is_default", False)), is_active=True, created_by=staff_profile, last_updated_by=staff_profile, ) payload["id"] = account.id return Response({"message": "true", "payload": payload}, status=200) except IntegrityError: return Response({"message": "false", "payload": {"error": "Account number already exists."}}, status=400) except AccountType.DoesNotExist: return Response({"message": "false", "payload": {"error": "Bank account type is not configured."}}, status=400) except ValueError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": {"error": "Failed to create bank account."}}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_bank_account(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data account_id = data.get("id") if not account_id: return Response({"message": "false", "payload": {"error": "Bank account id is required."}}, status=400) account = ChartOfAccount.objects.get(id=account_id, recycle_bin=False) if not _is_bank_gl_account(account): return Response({"message": "false", "payload": {"error": "Not a bank register account."}}, status=400) branch = staff_profile.company_branch if branch and account.company_branch_id != branch.id and not check_if_system_administrator(staff_profile): return Response({"message": "false", "payload": {"error": "Unauthorized."}}, status=401) is_mpesa = bool(data.get("is_mpesa", account.is_mpesa_account)) if "account_name" in data: account.account_name = (data.get("account_name") or "").strip() if "bank_name" in data or "is_mpesa" in data: account.banking_institution_name = "M-Pesa" if is_mpesa else (data.get("bank_name") or account.banking_institution_name or "").strip() if "bank_branch" in data: account.bank_branch_name = (data.get("bank_branch") or "").strip() if "bank_branch_code" in data: account.bank_branch_code = (data.get("bank_branch_code") or "").strip() if "bank_swift_code" in data: account.bank_swift_code = (data.get("bank_swift_code") or "").strip() if "bank_sort_code" in data: account.bank_sort_code = (data.get("bank_sort_code") or "").strip() if "currency" in data: account.account_currency = (data.get("currency") or account.account_currency or "kes").lower() if "is_mpesa" in data: account.is_mpesa_account = is_mpesa if "is_default" in data: if data.get("is_default"): _branch_chart_accounts(staff_profile).filter(is_default_account=True).exclude(id=account.id).update(is_default_account=False) account.is_default_account = bool(data.get("is_default")) if "gl_account_id" in data: gl_account_id = data.get("gl_account_id") if gl_account_id: parent_account = ChartOfAccount.objects.filter(id=gl_account_id, recycle_bin=False).first() if not parent_account: return Response({"message": "false", "payload": {"error": "Linked GL account not found."}}, status=400) account.parent_account = parent_account else: account.parent_account = None account.last_updated_by = staff_profile account.save() payload["id"] = account.id return Response({"message": "true", "payload": payload}, status=200) except ChartOfAccount.DoesNotExist: return Response({"message": "false", "payload": {"error": "Bank account not found."}}, status=404) except ValueError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": {"error": "Failed to update bank account."}}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_bank_account(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) account_id = request.data.get("id") if not account_id: return Response({"message": "false", "payload": {"error": "Bank account id is required."}}, status=400) account = ChartOfAccount.objects.get(id=account_id, recycle_bin=False) if not _is_bank_gl_account(account): return Response({"message": "false", "payload": {"error": "Not a bank register account."}}, status=400) branch = staff_profile.company_branch if branch and account.company_branch_id != branch.id and not check_if_system_administrator(staff_profile): return Response({"message": "false", "payload": {"error": "Unauthorized."}}, status=401) account.recycle_bin = True account.is_active = False account.last_updated_by = staff_profile account.save() payload["id"] = account.id return Response({"message": "true", "payload": payload}, status=200) except ChartOfAccount.DoesNotExist: return Response({"message": "false", "payload": {"error": "Bank account not found."}}, status=404) except ValueError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": {"error": "Failed to delete bank account."}}, status=500) # ============================================================================= # ACCOUNT MAPPINGS # ============================================================================= _PAYMENT_LABELS = { "cash": "Cash", "m_pesa": "M-Pesa", "bank_transfer": "Bank Transfer", "bankers_cheque": "Bankers Cheque", } _EXPENSE_LABELS = { "capital_expenditure": "Capital Expenditure", "purchase_expense": "Inventory Purchase Expense", "salary_expense": "Salary Expense", "loan_repayment_expense": "Loan Repayment Expense", "utility_expense": "Utility Expense", "rent_expense": "Rent Expense", "operations_expense": "Operations Expense", "refund_expense": "Refund Expense", "tax_and_license_expense": "Tax and License Expense", "entertainment_and_private_expense": "Entertainment & Private", "other": "Other", } _DEPOSIT_LABELS = { "sales_revenue": "Sales Revenue", "loans": "Loans", "vehicle_and_equipment_lease_revenue": "Vehicle & Equipment Lease Revenue", "share_capital": "Share Capital", "other": "Other", } _POSTING_EVENT_LABELS = { "sales_invoice": "Sales Invoice (DR AR · CR Revenue)", "customer_payment": "Customer Payment (DR Cash · CR AR)", "goods_delivery": "Delivery / Goods Issue (DR COGS · CR Inventory)", "goods_receipt": "Goods Receipt (DR Inventory · CR GRNI)", "vendor_invoice": "Supplier Invoice (DR GRNI · CR AP)", "supplier_payment": "Supplier Payment (DR AP · CR Cash)", "payroll_accrual": "Payroll Accrual (DR Salaries · CR Payable)", "payroll_payment": "Payroll Payment (DR Payable · CR Cash)", } _POSTING_EVENT_TYPES = { "sales_invoice": "invoice_posting", "customer_payment": "payment_posting_sale", "goods_delivery": "delivery_posting", "goods_receipt": "purchase_posting", "vendor_invoice": "purchase_posting", "supplier_payment": "payment_posting_expense", "payroll_accrual": "payroll_posting", "payroll_payment": "payroll_posting", } @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_account_mappings(request): """Return all account mappings with unmapped keys.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) mappings = AccountMapping.objects.filter( company_branch=staff_profile.company_branch ).select_related("debit_account", "credit_account").order_by("mapping_type", "source_key") result = [] for m in mappings: result.append({ "id": m.id, "mapping_type": m.mapping_type, "source_key": m.source_key, "source_label": m.source_label, "debit_account_id": m.debit_account_id, "debit_account_name": m.debit_account.account_name if m.debit_account else "", "debit_account_number": m.debit_account.account_number if m.debit_account else "", "credit_account_id": m.credit_account_id, "credit_account_name": m.credit_account.account_name if m.credit_account else "", "credit_account_number": m.credit_account.account_number if m.credit_account else "", "description": m.description, }) mapped_keys = {(m["mapping_type"], m["source_key"]) for m in result} payload["mappings"] = result payload["unmapped"] = { "payment_methods": [ {"key": k, "label": v} for k, v in _PAYMENT_LABELS.items() if ("payment_method", k) not in mapped_keys ], "expense_categories": [ {"key": k, "label": v} for k, v in _EXPENSE_LABELS.items() if ("expense_category", k) not in mapped_keys ], "deposit_sources": [ {"key": k, "label": v} for k, v in _DEPOSIT_LABELS.items() if ("deposit_source", k) not in mapped_keys ], } payload["mapping_keys"] = { "payment_methods": [{"key": k, "label": v} for k, v in _PAYMENT_LABELS.items()], "expense_categories": [{"key": k, "label": v} for k, v in _EXPENSE_LABELS.items()], "deposit_sources": [{"key": k, "label": v} for k, v in _DEPOSIT_LABELS.items()], "posting_events": [ { "key": k, "label": v, "mapping_type": _POSTING_EVENT_TYPES[k], } for k, v in _POSTING_EVENT_LABELS.items() ], } payload["unmapped"]["posting_events"] = [ { "key": k, "label": v, "mapping_type": _POSTING_EVENT_TYPES[k], } for k, v in _POSTING_EVENT_LABELS.items() if (_POSTING_EVENT_TYPES[k], k) not in mapped_keys ] return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def save_account_mapping(request): """Create or update a single account mapping.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data debit_account = ChartOfAccount.objects.get(id=data["debit_account_id"]) if data.get("debit_account_id") else None credit_account = ChartOfAccount.objects.get(id=data["credit_account_id"]) if data.get("credit_account_id") else None mapping, created = AccountMapping.objects.update_or_create( mapping_type=data["mapping_type"], source_key=data["source_key"], company_branch=staff_profile.company_branch, defaults={ "source_label": data.get("source_label", ""), "debit_account": debit_account, "credit_account": credit_account, "description": data.get("description", ""), "created_by": staff_profile, "last_updated_by": staff_profile, }, ) payload["mapping_id"] = mapping.id payload["created"] = created return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_account_mapping(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) AccountMapping.objects.filter( id=request.data["mapping_id"], company_branch=staff_profile.company_branch, ).delete() return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # MANUAL JOURNAL ENTRIES # ============================================================================= def _serialize_manual_journal(entry): lines = list(entry.journal_lines.all()) total_debit = sum(l.debit for l in lines) total_credit = sum(l.credit for l in lines) reversal = entry.reversal_entries.filter(recycle_bin=False, status="posted").first() return { "id": entry.id, "entry_number": entry.entry_number, "entry_date": str(entry.entry_date), "description": entry.description, "reference": entry.reference, "status": entry.status, "source_type": entry.source_type, "source_id": entry.source_id, "is_recurring": entry.is_recurring, "recurring_type": entry.recurring_type, "total_debit": str(total_debit), "total_credit": str(total_credit), "is_balanced": total_debit == total_credit, "posted_at": str(entry.posted_at) if entry.posted_at else "", "posted_by_name": ( f"{entry.posted_by.first_name} {entry.posted_by.last_name}" if entry.posted_by else "" ), "reversal_of_id": entry.reversal_of_id, "reversal_entry_number": reversal.entry_number if reversal else "", "created_by_name": ( f"{entry.created_by.first_name} {entry.created_by.last_name}" if entry.created_by else "" ), "created_on": str(entry.created_on.date()) if entry.created_on else "", "lines": [ { "id": line.id, "account_id": line.account_id, "account_number": line.account.account_number, "account_name": line.account.account_name, "account_type": line.account.account_type.type_key if line.account.account_type else "", "debit": str(line.debit), "credit": str(line.credit), "description": line.description, "line_order": line.line_order, } for line in lines ], } @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_manual_journal_entries(request): """Return manual journal entries with lines, optionally filtered by date.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) qs = ManualJournalEntry.objects.filter( recycle_bin=False, company_branch=staff_profile.company_branch, ).order_by("-entry_date", "-id") if request.data.get("date_from"): qs = qs.filter(entry_date__gte=_parse_finance_date(request.data["date_from"])) if request.data.get("date_to"): qs = qs.filter(entry_date__lte=_parse_finance_date(request.data["date_to"])) entries_list = [] for entry in qs.prefetch_related("journal_lines__account__account_type", "posted_by", "reversal_entries"): entries_list.append(_serialize_manual_journal(entry)) payload["journal_entries"] = entries_list payload["total"] = len(entries_list) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_manual_journal_entry(request): """ Create a manual journal entry. Draft by default; pass post_now=true to post to GL. Body: { serial_number, entry_date, description, reference, post_now, lines: [{account_id, debit, credit, description, line_order}] } """ payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data lines_data = data.get("lines", []) if len(lines_data) < 2: return Response({"message": "false", "payload": {"error": "Minimum 2 lines required"}}, status=400) total_debit = round(sum(float(l.get("debit", 0)) for l in lines_data), 2) total_credit = round(sum(float(l.get("credit", 0)) for l in lines_data), 2) if total_debit != total_credit: return Response({"message": "false", "payload": { "error": f"Entry not balanced: debits={total_debit} credits={total_credit}" }}, status=400) entry_date = date_cls.fromisoformat(data["entry_date"]) post_now = data.get("post_now", True) entry = ManualJournalEntry.objects.create( company_branch=staff_profile.company_branch, entry_date=entry_date, description=data.get("description", ""), reference=data.get("reference", ""), is_recurring=bool(data.get("is_recurring", False)), recurring_type=data.get("recurring_type", "none"), recurring_cycles=int(data.get("recurring_cycles", 0)), status="draft", source_type="manual", source_id="", created_by=staff_profile, last_updated_by=staff_profile, ) for idx, line in enumerate(lines_data): account = ChartOfAccount.objects.get(id=line["account_id"]) ManualJournalLine.objects.create( journal_entry=entry, account=account, debit=round(float(line.get("debit", 0)), 2), credit=round(float(line.get("credit", 0)), 2), description=line.get("description", ""), line_order=int(line.get("line_order", idx)), ) if post_now: post_journal_entry_to_ledger(entry, staff_profile) payload["entry_id"] = entry.id payload["entry_number"] = entry.entry_number payload["status"] = entry.status payload["journal_entry"] = _serialize_manual_journal(entry) return Response({"message": "true", "payload": payload}, status=200) except PostingError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def post_manual_journal_entry(request): """Post a draft journal entry into the General Ledger.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) entry = ManualJournalEntry.objects.get( id=int(request.data["entry_id"]), company_branch=staff_profile.company_branch, recycle_bin=False, ) post_journal_entry_to_ledger(entry, staff_profile) payload["journal_entry"] = _serialize_manual_journal(entry) return Response({"message": "true", "payload": payload}, status=200) except PostingError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def reverse_manual_journal_entry(request): """Reverse a posted journal entry via a new reversing entry.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) entry = ManualJournalEntry.objects.get( id=int(request.data["entry_id"]), company_branch=staff_profile.company_branch, recycle_bin=False, ) reversal = reverse_journal_entry( entry, staff_profile, description=request.data.get("description", ""), ) payload["reversal_entry"] = _serialize_manual_journal(reversal) payload["original_entry"] = _serialize_manual_journal(entry) return Response({"message": "true", "payload": payload}, status=200) except PostingError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_manual_journal_entry(request): """Cancel/delete draft journals. Posted entries must be reversed, not deleted.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) entry_id = int(request.data["entry_id"]) entry = ManualJournalEntry.objects.get( id=entry_id, company_branch=staff_profile.company_branch, recycle_bin=False) if entry.status == "posted": return Response({"message": "false", "payload": { "error": "Posted journal entries cannot be deleted. Use Reverse instead.", }}, status=400) if entry.status == "reversed": return Response({"message": "false", "payload": { "error": "Reversed journal entries cannot be deleted.", }}, status=400) if entry.status == "draft": AccountHistory.objects.filter(journal_entry=entry).delete() cancel_journal_draft(entry, staff_profile) else: entry.recycle_bin = True entry.save(update_fields=["recycle_bin", "last_updated_on"]) return Response({"message": "true", "payload": payload}, status=200) except PostingError as e: return Response({"message": "false", "payload": {"error": str(e)}}, status=400) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # GENERAL LEDGER (AccountHistory) # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_account_history(request): """All GL postings for one account with running balance.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) account = ChartOfAccount.objects.get(id=request.data["account_id"], recycle_bin=False) qs = AccountHistory.objects.filter(account=account).select_related( "journal_entry", ).order_by("transaction_date", "id") qs = _filter_by_date_range(qs, request, "transaction_date") normal_bal = account.account_type.normal_balance if account.account_type else "debit" running = 0 transactions = [] for hist in qs: dr = float(hist.debit) cr = float(hist.credit) running += (dr - cr) if normal_bal == "debit" else (cr - dr) transactions.append({ "id": hist.id, "transaction_date": str(hist.transaction_date), "description": hist.description, "rel_type": hist.rel_type, "rel_id": hist.rel_id, "journal_entry_id": hist.journal_entry_id, "journal_entry_number": hist.journal_entry.entry_number if hist.journal_entry else "", "debit": str(dr), "credit": str(cr), "running_balance": str(round(running, 2)), "reconciled": hist.reconciled, }) payload["account"] = { "id": account.id, "account_number": account.account_number, "account_name": account.account_name, "account_type": account.account_type.name if account.account_type else "", "normal_balance": normal_bal, } payload["transactions"] = transactions payload["total_debit"] = str(round(sum(float(t["debit"]) for t in transactions), 2)) payload["total_credit"] = str(round(sum(float(t["credit"]) for t in transactions), 2)) payload["closing_balance"] = str(round(running, 2)) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_general_ledger(request): """Full GL report: all accounts, chronological with running balance.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) qs = AccountHistory.objects.filter( company_branch=staff_profile.company_branch ).select_related("account", "account__account_type").order_by( "account__account_number", "transaction_date", "id") qs = _filter_by_date_range(qs, request, "transaction_date") if request.data.get("account_id"): qs = qs.filter(account_id=request.data["account_id"]) accounts_data = OrderedDict() for hist in qs: aid = hist.account_id if aid not in accounts_data: acct = hist.account normal_bal = acct.account_type.normal_balance if acct.account_type else "debit" accounts_data[aid] = { "account_id": aid, "account_number": acct.account_number, "account_name": acct.account_name, "account_type": acct.account_type.name if acct.account_type else "", "account_type_key": acct.account_type.type_key if acct.account_type else "", "normal_balance": normal_bal, "transactions": [], "total_debit": 0.0, "total_credit": 0.0, "running_balance": 0.0, } d = accounts_data[aid] dr = float(hist.debit) cr = float(hist.credit) d["total_debit"] += dr d["total_credit"] += cr d["running_balance"] += (dr - cr) if d["normal_balance"] == "debit" else (cr - dr) d["transactions"].append({ "id": hist.id, "transaction_date": str(hist.transaction_date), "description": hist.description, "rel_type": hist.rel_type, "rel_id": hist.rel_id, "debit": str(dr), "credit": str(cr), "running_balance": str(round(d["running_balance"], 2)), }) result = [] for d in accounts_data.values(): d["total_debit"] = str(round(d["total_debit"], 2)) d["total_credit"] = str(round(d["total_credit"], 2)) d["running_balance"] = str(round(d["running_balance"], 2)) result.append(d) payload["ledger"] = result payload["total_accounts"] = len(result) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # TRIAL BALANCE # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_trial_balance(request): """ Compute trial balance by aggregating AccountHistory by account. Shows debit/credit totals and net balance per account. """ payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) qs = fr.branch_ledger( staff_profile.company_branch, request.data.get("date_from"), request.data.get("date_to"), ) payload.update(fr.build_trial_balance(qs)) payload["date_from"] = request.data.get("date_from", "") payload["date_to"] = request.data.get("date_to", "") return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # ENHANCED P&L FROM LEDGER # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_pl_from_ledger(request): """P&L driven entirely by account type classification in AccountHistory.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) qs = fr.branch_ledger( staff_profile.company_branch, request.data.get("date_from"), request.data.get("date_to"), ) payload.update(fr.build_profit_and_loss(qs)) payload["date_from"] = request.data.get("date_from", "") payload["date_to"] = request.data.get("date_to", "") return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # ENHANCED BALANCE SHEET FROM LEDGER # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_balance_sheet_from_ledger(request): """ Balance Sheet from AccountHistory + P&L retained earnings. Verifies Assets = Liabilities + Equity (accounting equation). """ payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) date_to = request.data.get("date_to") or request.data.get("date_from") qs = fr.branch_ledger(staff_profile.company_branch, date_to=date_to) payload.update(fr.build_balance_sheet(qs)) payload["date_to"] = request.data.get("date_to", "") return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # CASH FLOW FROM LEDGER # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_cash_flow_from_ledger(request): """Statement of cash flows derived from AccountHistory (Ocean-style indirect method).""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) qs = fr.branch_ledger( staff_profile.company_branch, request.data.get("date_from"), request.data.get("date_to"), ) payload.update(fr.build_cash_flow_statement(qs)) payload["date_from"] = request.data.get("date_from", "") payload["date_to"] = request.data.get("date_to", "") return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # GL JOURNAL (all ledger postings) # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_journal_from_ledger(request): """Chronological journal of all AccountHistory postings (Ocean journal report).""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) qs = fr.branch_ledger( staff_profile.company_branch, request.data.get("date_from"), request.data.get("date_to"), ) payload.update(fr.build_ledger_journal(qs)) payload["date_from"] = request.data.get("date_from", "") payload["date_to"] = request.data.get("date_to", "") return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # ACCOUNTING PERIODS (fiscal period management) # ============================================================================= def _serialize_period(p): return { "period_id": p.id, "period_name": p.period_name, "period_start_date": str(p.period_start_date), "period_end_date": str(p.period_end_date), "period_locked": p.period_locked, "created_by": p.created_by.staff_number if p.created_by else "", "created_on": str(p.created_on)[:10], "last_updated_on": str(p.last_updated_on)[:10], } @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_accounting_periods(request): """List all accounting periods for the branch.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) periods = AccountingPeriod.objects.filter( company_branch=staff_profile.company_branch, recycle_bin=False, ) payload["accounting_periods_list"] = [_serialize_period(p) for p in periods] return Response({"message": "true", "payload": payload}, status=200) except ValueError as e: return _auth_error_response(payload, e) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_accounting_period(request): """Create a new accounting period.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data period = AccountingPeriod.objects.create( company_branch=staff_profile.company_branch, period_name=data["period_name"], period_start_date=_parse_finance_date(data["period_start_date"]), period_end_date=_parse_finance_date(data["period_end_date"]), created_by=staff_profile, last_updated_by=staff_profile, ) payload["period"] = _serialize_period(period) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_accounting_period(request): """Edit an existing accounting period (name and dates only — not locked periods).""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data period = AccountingPeriod.objects.get( id=data["period_id"], company_branch=staff_profile.company_branch, recycle_bin=False, ) if period.period_locked: return Response({"message": "false", "payload": {"error": "Period is locked"}}, status=400) if "period_name" in data: period.period_name = data["period_name"] if "period_start_date" in data: period.period_start_date = _parse_finance_date(data["period_start_date"]) if "period_end_date" in data: period.period_end_date = _parse_finance_date(data["period_end_date"]) period.last_updated_by = staff_profile period.save() payload["period"] = _serialize_period(period) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def lock_unlock_accounting_period(request): """Toggle the locked status of an accounting period.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data period = AccountingPeriod.objects.get( id=data["period_id"], company_branch=staff_profile.company_branch, recycle_bin=False, ) period.period_locked = not period.period_locked period.last_updated_by = staff_profile period.save() payload["period"] = _serialize_period(period) payload["period_locked"] = period.period_locked return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_accounting_period(request): """Soft-delete an accounting period.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) period = AccountingPeriod.objects.get( id=request.data["period_id"], company_branch=staff_profile.company_branch, recycle_bin=False, ) if period.period_locked: return Response({"message": "false", "payload": {"error": "Cannot delete a locked period"}}, status=400) period.recycle_bin = True period.save() return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # A/R AGEING REPORT # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_ar_ageing(request): """Accounts Receivable Ageing: outstanding customer order balances by age bucket.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) as_of = request.data.get("as_of_date") or None payload.update(fr.build_ar_ageing(staff_profile.company_branch, as_of)) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # A/P AGEING REPORT # ============================================================================= @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_ap_ageing(request): """Accounts Payable Ageing: outstanding purchase order balances by age bucket.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) as_of = request.data.get("as_of_date") or None payload.update(fr.build_ap_ageing(staff_profile.company_branch, as_of)) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # ACCOUNTING SETTINGS # ============================================================================= def _serialize_accounting_settings(s): return { "fiscal_year_start_month": s.fiscal_year_start_month, "tax_year_start_month": s.tax_year_start_month, "accounting_method": s.accounting_method, "close_books": s.close_books, "closing_date": str(s.closing_date) if s.closing_date else "", "enable_account_numbers": s.enable_account_numbers, "show_account_numbers": s.show_account_numbers, } @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_accounting_settings(request): """Return (or auto-create) accounting settings for this branch.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) settings, _ = AccountingSettings.objects.get_or_create( company_branch=staff_profile.company_branch) payload["settings"] = _serialize_accounting_settings(settings) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def save_accounting_settings(request): """Upsert accounting settings for this branch.""" payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) data = request.data settings, _ = AccountingSettings.objects.get_or_create( company_branch=staff_profile.company_branch) bool_fields = ("close_books", "enable_account_numbers", "show_account_numbers") for field in ("fiscal_year_start_month", "tax_year_start_month", "accounting_method"): if field in data: setattr(settings, field, data[field]) for field in bool_fields: if field in data: setattr(settings, field, data[field] in (True, "true", "True", 1, "1")) if "closing_date" in data: settings.closing_date = data["closing_date"] or None settings.updated_by = staff_profile settings.save() payload["settings"] = _serialize_accounting_settings(settings) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # FINANCE CUSTOMER PROFILES & CREDIT MANAGEMENT # ============================================================================= def _customer_display_name(customer): parts = [customer.customer_first_name, customer.customer_last_name] name = " ".join(p for p in parts if p).strip() if customer.company_name: return customer.company_name if not name else f"{name} ({customer.company_name})" return name or "Unnamed Customer" def _customer_outstanding_balance(customer): from sales_and_marketing.models import CustomerOrder total = 0.0 orders = CustomerOrder.objects.filter( customer_profile=customer, customer_order_approved=True, recycle_bin=False, ) for order in orders: net = float(order.customer_order_total_net_value or 0) paid = float(order.customer_order_total_amount_paid or 0) total += max(0.0, round(net - paid, 2)) return round(total, 2) def _serialize_finance_customer(customer): from sales_and_marketing.models import CustomerOrder outstanding = _customer_outstanding_balance(customer) order_count = CustomerOrder.objects.filter( customer_profile=customer, recycle_bin=False, ).count() account_manager = {"staff_id": "", "staff_name": ""} return { "customer_id": str(customer.id), "customer_profile_id": str(customer.id), "customer_name": _customer_display_name(customer), "customer_first_name": customer.customer_first_name, "customer_last_name": customer.customer_last_name, "customer_type": customer.customer_type, "customer_phone": customer.phone_number, "customer_email": customer.email_address, "email_address": customer.email_address, "phone_number": customer.phone_number, "company_name": customer.company_name, "kra_pin": customer.kra_pin, "customer_credit_limit": customer.customer_credit_limit or "0.00", "customer_payment_terms_days": customer.customer_payment_terms_days or "30", "customer_outstanding_balance": str(outstanding), "total_customer_orders": str(order_count), "customer_account_manager": account_manager, } @api_view(["POST"]) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def finance_customer_profiles(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) from sales_and_marketing.models import CustomerProfile customers = CustomerProfile.objects.filter( company_profile=company_profile, recycle_bin=False, ).order_by("-id") customer_profiles_list = [ _serialize_finance_customer(c) for c in customers ] payload["customer_profiles_list"] = customer_profiles_list payload["customer_count"] = str(len(customer_profiles_list)) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(["POST"]) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def finance_view_single_customer(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) from sales_and_marketing.models import CustomerProfile customer_id = request.data.get("customer_id") if not customer_id: return Response({"message": "false", "payload": payload}, status=400) customer = CustomerProfile.objects.get( id=int(customer_id), company_profile=company_profile, recycle_bin=False, ) payload["customer_profile"] = _serialize_finance_customer(customer) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(["POST"]) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def finance_edit_customer_credit(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_write(staff_profile): return Response({"message": "false", "payload": payload}, status=401) from sales_and_marketing.models import CustomerProfile customer_id = request.data.get("customer_id") if not customer_id: return Response({"message": "false", "payload": payload}, status=400) customer = CustomerProfile.objects.get( id=int(customer_id), company_profile=company_profile, recycle_bin=False, ) if "customer_credit_limit" in request.data: customer.customer_credit_limit = str(request.data["customer_credit_limit"] or "0.00") if "customer_payment_terms_days" in request.data: customer.customer_payment_terms_days = str( request.data["customer_payment_terms_days"] or "30" ) customer.save() payload["customer_profile"] = _serialize_finance_customer(customer) return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) # ============================================================================= # FINANCE SUPPLIER LIST (Accounts Payable) # ============================================================================= def _supplier_outstanding_balance(supplier, branch): from procurement.models import ProductPurchaseInstance, PurchaseOrder total = 0.0 paid_map = {} ah_qs = AccountHistory.objects.filter( company_branch=branch, rel_type="purchase_order", ).values("rel_id", "credit") for h in ah_qs: rid = str(h["rel_id"]) paid_map[rid] = paid_map.get(rid, 0.0) + _ledger_amount(h["credit"]) po_ids = ProductPurchaseInstance.objects.filter( supplier=supplier, recycle_bin=False, ).values_list("purchase_order_id", flat=True).distinct() orders = PurchaseOrder.objects.filter( id__in=list(po_ids), company_profile=branch.company_profile, purchase_order_approved=True, recycle_bin=False, ) for order in orders: total_val = float(order.purchase_value_overall or 0) paid = paid_map.get(str(order.id), 0.0) total += max(0.0, round(total_val - paid, 2)) return round(total, 2) @api_view(["POST"]) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def finance_supplier_list(request): payload = {} try: company_profile, staff_profile = _finance_auth(request) if not _require_finance_staff(staff_profile): return Response({"message": "false", "payload": payload}, status=401) from procurement.models import Supplier branch = staff_profile.company_branch suppliers = Supplier.objects.filter( company_profile=company_profile, recycle_bin=False, ).order_by("-id") supplier_list = [] for supplier in suppliers: outstanding = _supplier_outstanding_balance(supplier, branch) supplier_list.append({ "supplier_id": str(supplier.id), "supplier_name": supplier.supplier_name, "supplier_phone": supplier.supplier_phone, "supplier_email": supplier.supplier_email, "supplier_address": supplier.supplier_address, "supplier_description": supplier.supplier_description, "supplier_category": supplier.supplier_description or "General", "supplier_outstanding_balance": str(outstanding), "preferred_currency": company_profile.company_preferred_currency, "created_on": supplier.created_on.isoformat() if supplier.created_on else "", "last_updated_on": supplier.last_updated_on.isoformat() if supplier.last_updated_on else "", }) payload["supplier_list"] = supplier_list payload["company_profile_map"] = { "company_id": str(company_profile.id), "company_name": company_profile.company_name, "preferred_currency": company_profile.company_preferred_currency, } return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500)