"""Unified financial overview for dashboard and reports.""" from decimal import Decimal from django.db.models import Sum from api.models.data.currency import BASE_CURRENCY_ID, CURRENCY_DEFINITIONS, CURRENCY_BY_CODE, get_currency from api.models.data.expenses import Expense from api.models.data.journal import JournalLine from api.models.data.loan import Loan, LoanPayment from api.models.data.payment import Payment from api.models.data.payroll import Payroll from api.models.data.returns import Return from api.models.data.raw_materials import RawMaterialPurchase from api.models.data.sales import Sales from api.utils.currency import convert_currency SKIP_SALE_STATUSES = {'draft', 'cancelled'} SKIP_PURCHASE_STATUSES = {'draft', 'cancelled', 'pending'} SALE_CASH_STATUSES = {'confirmed', 'shipped', 'delivered', 'invoiced'} PURCHASE_CASH_STATUSES = {'received'} def _d(value): return Decimal(str(value or 0)) def _float(value): return float(_d(value)) def _convert(amount, from_currency_id, to_currency_id, date=None): if not amount or not from_currency_id or not to_currency_id: return _d(amount) if int(from_currency_id) == int(to_currency_id): return _d(amount) return convert_currency(_d(amount), from_currency_id, to_currency_id, date) def _gl_account_balance(base_code, currency_id, *, end_date=None, category='asset'): """ Net balance for a GL account by bare code + currency id. Currency-specific accounts (Cash/Bank/Sarafi) use native line amounts. Shared chart accounts use AFN base_* when viewing AFN, native lines when viewing a foreign currency book. """ from api.constants.gl_roles import is_currency_specific_gl_code from api.models.data.currency import BASE_CURRENCY_ID, CURRENCY_DEFINITIONS from api.models.data.journal import GLAccount currency_id = int( currency_id) if currency_id is not None else BASE_CURRENCY_ID qs = GLAccount.objects.filter(code=base_code, is_active=True) if is_currency_specific_gl_code(base_code): qs = qs.filter(currency=currency_id) account = qs.first() else: account = qs.filter(currency=BASE_CURRENCY_ID).first() or qs.first() if not account: legacy_code = f'{base_code}-{CURRENCY_DEFINITIONS.get(currency_id, {}).get("code", "")}' account = GLAccount.objects.filter( code=legacy_code, is_active=True).first() if account: category = account.category lines = JournalLine.objects.filter( gl_account=account) if account else JournalLine.objects.none() if end_date: lines = lines.filter(journal_entry__entry_date__lte=end_date) if account and is_currency_specific_gl_code(account.code): lines = lines.filter(currency=currency_id) agg = lines.aggregate(total_debit=Sum('debit'), total_credit=Sum('credit')) elif currency_id == BASE_CURRENCY_ID: agg = lines.aggregate(total_debit=Sum( 'base_debit'), total_credit=Sum('base_credit')) else: lines = lines.filter(currency=currency_id) agg = lines.aggregate(total_debit=Sum('debit'), total_credit=Sum('credit')) debit = _d(agg['total_debit']) credit = _d(agg['total_credit']) if category in ('liability', 'equity', 'revenue'): return credit - debit return debit - credit def _consolidated_gl_balance(base_code, target_currency_id, *, end_date=None, category='asset'): """Sum GL balances across currencies, converted to target.""" total = Decimal('0') for cid in CURRENCY_DEFINITIONS: balance = _gl_account_balance( base_code, cid, end_date=end_date, category=category ) if balance: total += _convert(balance, cid, target_currency_id, end_date) return total def _consolidated_contact_balances(target_currency_id, *, contact_type): """Sum customer receivables or vendor payables with per-currency conversion.""" from api.models.data.customers import Customer from api.models.data.vendors import Vendor from api.services.contact_ledger import get_customer_finance, get_vendor_finance total = Decimal('0') if contact_type == 'customer': contacts = Customer.objects.all() get_finance = get_customer_finance else: contacts = Vendor.objects.all() get_finance = get_vendor_finance for contact in contacts: finance = get_finance(contact) for code, bucket in finance.items(): if code == 'overall': continue c = CURRENCY_BY_CODE.get(code.upper()) if not c: continue balance = _d(bucket['balance']) if balance <= 0: continue total += _convert(balance, c['id'], target_currency_id) return total def _filter_by_date_range(qs, field_name, start_date, end_date): """Apply optional inclusive datetime bounds (None = no bound).""" if start_date and end_date: return qs.filter(**{f'{field_name}__range': [start_date, end_date]}) if start_date: return qs.filter(**{f'{field_name}__gte': start_date}) if end_date: return qs.filter(**{f'{field_name}__lte': end_date}) return qs def _period_sales(start_date, end_date): return _filter_by_date_range( Sales.objects.exclude(status__in=SKIP_SALE_STATUSES), 'sale_date', start_date, end_date, ) def _period_purchases(start_date, end_date): """Received raw-material purchases (AP / inventory intake).""" return _filter_by_date_range( RawMaterialPurchase.objects.filter(status='received'), 'purchase_date', start_date, end_date, ) def _period_returns(start_date, end_date): return _filter_by_date_range(Return.objects.all(), 'return_date', start_date, end_date) def _period_expenses(start_date, end_date): return _filter_by_date_range(Expense.objects.all(), 'expense_date', start_date, end_date) def _period_payroll(start_date, end_date): return _filter_by_date_range( Payroll.objects.filter(payroll_type='salary'), 'payment_date', start_date, end_date, ) def _period_advances(start_date, end_date): return _filter_by_date_range( Payroll.objects.filter(payroll_type='advance'), 'payment_date', start_date, end_date, ) def _period_sale_payments(start_date, end_date): return _filter_by_date_range( Payment.objects.filter(payment_type='sale'), 'payment_date', start_date, end_date, ) def _period_purchase_payments(start_date, end_date): return _filter_by_date_range( Payment.objects.filter(payment_type='purchase'), 'payment_date', start_date, end_date, ) def _period_loans(start_date, end_date): return _filter_by_date_range(Loan.objects.all(), 'loan_date', start_date, end_date) def _period_loan_payments(start_date, end_date): return _filter_by_date_range(LoanPayment.objects.all(), 'payment_date', start_date, end_date) def _period_withdrawals(start_date, end_date): from api.models.data.withdrawal import Withdrawal return _filter_by_date_range( Withdrawal.objects.all(), 'withdrawal_date', start_date, end_date ) def _sum_remaining_base(qs, total_field, paid_field): """Sum unpaid remainder in base currency for sales/purchases in a period.""" total = Decimal('0') for row in qs.only(total_field, paid_field): remaining = max(_d(getattr(row, total_field)) - _d(getattr(row, paid_field)), Decimal('0')) if remaining > 0: total += remaining return total def _sum_remaining_native(qs, total_field, paid_field): total = Decimal('0') for row in qs.only(total_field, paid_field): remaining = max(_d(getattr(row, total_field)) - _d(getattr(row, paid_field)), Decimal('0')) if remaining > 0: total += remaining return total def _period_receivable_base(start_date, end_date): """Unpaid sales balance for transactions in the selected period.""" return _sum_remaining_base( _period_sales(start_date, end_date), 'total_amount_base', 'paid_amount_base', ) def _period_payable_base(start_date, end_date): """Unpaid purchase balance for transactions in the selected period.""" return _sum_remaining_base( _period_purchases(start_date, end_date), 'total_amount_base', 'paid_amount_base', ) def _period_loans_outstanding_base(start_date, end_date, loan_type): """Outstanding principal on loans originated in the selected period.""" return _sum_remaining_base( _period_loans(start_date, end_date).filter(loan_type=loan_type), 'amount_base', 'amount_paid_base', ) def _period_receivable_native(start_date, end_date, currency): return _sum_remaining_native( _period_sales(start_date, end_date).filter(currency=currency), 'total_amount', 'paid_amount', ) def _period_payable_native(start_date, end_date, currency): return _sum_remaining_native( _period_purchases(start_date, end_date).filter(currency=currency), 'total_amount', 'paid_amount', ) def _period_loans_outstanding_native(start_date, end_date, currency, loan_type): return _sum_remaining_native( _period_loans(start_date, end_date).filter( currency=currency, loan_type=loan_type), 'amount', 'amount_paid', ) def _loan_outstanding_at(currency, end_date, loan_type): """Outstanding principal by loan type, using amount_paid on each loan.""" qs = Loan.objects.filter(currency=currency, loan_type=loan_type) if end_date: qs = qs.filter(loan_date__lte=end_date) outstanding = Decimal('0') for loan in qs: outstanding += max(_d(loan.amount) - _d(loan.amount_paid), Decimal('0')) return outstanding def _breakdown_entry(label, amount, count): return {'type': label, 'amount': _float(amount), 'count': count} def _contact_balance_for_currency(currency_code, *, contact_type): """Sum receivable/payable for one currency from contact ledger.""" from api.models.data.customers import Customer from api.models.data.vendors import Vendor from api.services.contact_ledger import get_customer_finance, get_vendor_finance total = Decimal('0') key = currency_code.lower() contacts = Customer.objects.all() if contact_type == 'customer' else Vendor.objects.all() get_finance = get_customer_finance if contact_type == 'customer' else get_vendor_finance for contact in contacts: finance = get_finance(contact) if key not in finance: continue balance = _d(finance[key]['balance']) if balance > 0: total += balance return total def _sum_base(qs, field): """Sum a *_base amount column (already in reporting base currency).""" return _d(qs.aggregate(t=Sum(field))['t']) def _build_base_summary(start_date, end_date): """Consolidated report in reporting base currency — uses stored *_base fields.""" sales = _period_sales(start_date, end_date) purchases = _period_purchases(start_date, end_date) returns = _period_returns(start_date, end_date) expenses = _period_expenses(start_date, end_date) payroll = _period_payroll(start_date, end_date) advances = _period_advances(start_date, end_date) sale_payments = _period_sale_payments(start_date, end_date) purchase_payments = _period_purchase_payments(start_date, end_date) period_loans = _period_loans(start_date, end_date) period_loan_payments = _period_loan_payments(start_date, end_date) loans_given = period_loans.filter(loan_type='loan_out') loans_received = period_loans.filter(loan_type='loan_in') collections_on_given = period_loan_payments.filter( loan__loan_type='loan_out') repayments_on_received = period_loan_payments.filter( loan__loan_type='loan_in') loans_given_total = _sum_base(loans_given, 'amount_base') loans_received_total = _sum_base(loans_received, 'amount_base') loan_collections_total = _sum_base(collections_on_given, 'amount_base') loan_repayments_total = _sum_base(repayments_on_received, 'amount_base') revenue = _sum_base(sales, 'total_amount_base') purchase_total = _sum_base(purchases, 'total_amount_base') returns_total = _sum_base(returns, 'total_amount_base') expense_total = _sum_base(expenses, 'amount_base') payroll_total = _sum_base(payroll, 'amount_base') advance_total = _sum_base(advances, 'amount_base') sales_payments_total = _sum_base(sale_payments, 'amount_base') purchase_payments_total = _sum_base(purchase_payments, 'amount_base') return_refunds_total = _sum_base(returns, 'refund_amount_base') withdrawals = _period_withdrawals(start_date, end_date) withdrawal_total = _sum_base(withdrawals, 'amount_base') net_revenue = revenue - returns_total gross_profit = net_revenue - purchase_total cash_from_sales = sales_payments_total credit_sales = _period_receivable_base(start_date, end_date) total_cash_in = cash_from_sales + loans_received_total + loan_collections_total total_cash_out = ( purchase_payments_total + expense_total + payroll_total + advance_total + return_refunds_total + loans_given_total + loan_repayments_total ) net_cash_flow = total_cash_in - total_cash_out base_code = CURRENCY_DEFINITIONS[BASE_CURRENCY_ID]['code'] accounts_receivable = _period_receivable_base(start_date, end_date) accounts_payable = _period_payable_base(start_date, end_date) loans_receivable = _period_loans_outstanding_base( start_date, end_date, 'loan_out') loans_payable = _period_loans_outstanding_base( start_date, end_date, 'loan_in') cash_balance = net_cash_flow from api.services.accounting.reports import ( REPORT_MODE_BASE, get_account_code_balance, get_profit_and_loss, ) ledger_pl = get_profit_and_loss( start_date=start_date, end_date=end_date, mode=REPORT_MODE_BASE, ) ledger_ar = get_account_code_balance( '1200', start_date=None, end_date=end_date, mode=REPORT_MODE_BASE, ) ledger_ap = get_account_code_balance( '2000', start_date=None, end_date=end_date, mode=REPORT_MODE_BASE, ) ledger_cash = ( get_account_code_balance( '1000', start_date=None, end_date=end_date, mode=REPORT_MODE_BASE, ) + get_account_code_balance( '1050', start_date=None, end_date=end_date, mode=REPORT_MODE_BASE, ) + get_account_code_balance( '1060', start_date=None, end_date=end_date, mode=REPORT_MODE_BASE, ) ) return { 'total_revenue': _float(ledger_pl['revenue']), 'total_returns': _float(returns_total), 'total_income': _float(ledger_pl['revenue']), 'cash_from_sales': _float(cash_from_sales), 'credit_sales': _float(credit_sales), 'total_purchases': _float(purchase_total), 'gross_profit': _float(ledger_pl['gross_profit']), 'operating_expenses': _float(ledger_pl['expenses']), 'total_payroll': _float(payroll_total), 'total_advances': _float(advance_total), 'total_expenses': _float(ledger_pl['expenses'] + ledger_pl['cogs']), 'net_profit': _float(ledger_pl['net_income']), 'total_cash_in': _float(total_cash_in), 'total_cash_out': _float(total_cash_out), 'net_cash_flow': _float(net_cash_flow), 'cash_balance': _float(ledger_cash), 'accounts_receivable': _float(ledger_ar), 'accounts_payable': _float(ledger_ap), 'loans_receivable': _float(loans_receivable), 'loans_payable': _float(loans_payable), 'loans_given': _float(loans_given_total), 'loans_received': _float(loans_received_total), 'loan_collections': _float(loan_collections_total), 'loan_repayments': _float(loan_repayments_total), 'base_currency_code': base_code, 'cash_in_breakdown': [ _breakdown_entry('Sales Payments', sales_payments_total, sale_payments.count()), _breakdown_entry('Loans Received', loans_received_total, loans_received.count()), _breakdown_entry( 'Loan Collections', loan_collections_total, collections_on_given.count()), ], 'cash_out_breakdown': [ _breakdown_entry( 'Purchase Payments', purchase_payments_total, purchase_payments.count()), _breakdown_entry('Expenses', expense_total, expenses.count()), _breakdown_entry('Payroll', payroll_total, payroll.count()), _breakdown_entry('Advances', advance_total, advances.count()), _breakdown_entry('Return Refunds', return_refunds_total, returns.filter(refund_amount__gt=0).count()), _breakdown_entry('Withdrawals', withdrawal_total, withdrawals.count()), _breakdown_entry('Loans Given', loans_given_total, loans_given.count()), _breakdown_entry( 'Loan Repayments', loan_repayments_total, repayments_on_received.count()), ], 'expense_breakdown': [ _breakdown_entry('COGS', ledger_pl['cogs'], 0), _breakdown_entry('Operating Expenses', expense_total, expenses.count()), _breakdown_entry('Payroll', payroll_total, payroll.count()), _breakdown_entry('Advances (cash, not P&L)', advance_total, advances.count()), ], 'loan_breakdown': [ _breakdown_entry('Loans Given', loans_given_total, loans_given.count()), _breakdown_entry('Loans Received', loans_received_total, loans_received.count()), _breakdown_entry('Collections (Loan Out)', loan_collections_total, collections_on_given.count()), _breakdown_entry( 'Repayments (Loan In)', loan_repayments_total, repayments_on_received.count()), ], 'sales_total': _float(revenue), 'cogs': _float(ledger_pl['cogs']), } def _loan_outstanding_base(end_date, loan_type): """Outstanding loan principal in base currency.""" qs = Loan.objects.filter(loan_type=loan_type) if end_date: qs = qs.filter(loan_date__lte=end_date) outstanding = Decimal('0') for loan in qs: outstanding += max(_d(loan.amount_base) - _d(loan.amount_paid_base), Decimal('0')) return outstanding def _build_currencies_report(start_date, end_date): """Full per-currency report — no cross-currency conversion.""" snapshot = {} for cid, cdef in CURRENCY_DEFINITIONS.items(): code = cdef['code'] sales = _period_sales(start_date, end_date).filter(currency=cid) purchases = _period_purchases( start_date, end_date).filter(currency=cid) returns = _period_returns(start_date, end_date).filter(currency=cid) expenses = _period_expenses(start_date, end_date).filter(currency=cid) payroll = _period_payroll(start_date, end_date).filter(currency=cid) advances = _period_advances(start_date, end_date).filter(currency=cid) sale_payments = _period_sale_payments( start_date, end_date).filter(currency=cid) purchase_payments = _period_purchase_payments( start_date, end_date).filter(currency=cid) period_loans = _period_loans(start_date, end_date).filter(currency=cid) period_loan_payments = _period_loan_payments( start_date, end_date).filter(loan__currency=cid) loans_given = period_loans.filter(loan_type='loan_out') loans_received = period_loans.filter(loan_type='loan_in') collections_on_given = period_loan_payments.filter( loan__loan_type='loan_out') repayments_on_received = period_loan_payments.filter( loan__loan_type='loan_in') loans_given_total = _d(loans_given.aggregate(t=Sum('amount'))['t']) loans_received_total = _d( loans_received.aggregate(t=Sum('amount'))['t']) loan_collections_total = _d( collections_on_given.aggregate(t=Sum('amount'))['t']) loan_repayments_total = _d( repayments_on_received.aggregate(t=Sum('amount'))['t']) revenue = _d(sales.aggregate(t=Sum('total_amount'))['t']) purchase_total = _d(purchases.aggregate(t=Sum('total_amount'))['t']) returns_total = _d(returns.aggregate(t=Sum('total_amount'))['t']) expense_total = _d(expenses.aggregate(t=Sum('amount'))['t']) payroll_total = _d(payroll.aggregate(t=Sum('amount'))['t']) advance_total = _d(advances.aggregate(t=Sum('amount'))['t']) sales_payments_total = _d( sale_payments.aggregate(t=Sum('amount'))['t']) purchase_payments_total = _d( purchase_payments.aggregate(t=Sum('amount'))['t']) return_refunds_total = _d( returns.aggregate(t=Sum('refund_amount'))['t'] ) withdrawals = _period_withdrawals( start_date, end_date).filter(currency=cid) withdrawal_total = _d(withdrawals.aggregate(t=Sum('amount'))['t']) net_revenue = revenue - returns_total gross_profit = net_revenue - purchase_total total_expenses = purchase_total + expense_total net_profit = net_revenue - total_expenses cash_from_sales = sales_payments_total credit_sales = _period_receivable_native(start_date, end_date, cid) total_cash_in = cash_from_sales + loans_received_total + loan_collections_total total_cash_out = ( purchase_payments_total + expense_total + payroll_total + advance_total + return_refunds_total + loans_given_total + loan_repayments_total ) net_cash_flow = total_cash_in - total_cash_out accounts_receivable = _period_receivable_native( start_date, end_date, cid) accounts_payable = _period_payable_native(start_date, end_date, cid) loans_receivable = _period_loans_outstanding_native( start_date, end_date, cid, 'loan_out') loans_payable = _period_loans_outstanding_native( start_date, end_date, cid, 'loan_in') cash_balance = net_cash_flow snapshot[code] = { 'currency_id': cid, 'symbol': cdef['symbol'], 'name': cdef['name'], 'total_revenue': _float(revenue), 'total_returns': _float(returns_total), 'total_income': _float(net_revenue), 'cash_from_sales': _float(cash_from_sales), 'credit_sales': _float(credit_sales), 'total_purchases': _float(purchase_total), 'gross_profit': _float(gross_profit), 'operating_expenses': _float(expense_total), 'total_payroll': _float(payroll_total), 'total_advances': _float(advance_total), 'total_expenses': _float(total_expenses), 'net_profit': _float(net_profit), 'total_cash_in': _float(total_cash_in), 'total_cash_out': _float(total_cash_out), 'net_cash_flow': _float(net_cash_flow), 'cash_balance': _float(cash_balance), 'accounts_receivable': _float(accounts_receivable), 'accounts_payable': _float(accounts_payable), 'loans_receivable': _float(loans_receivable), 'loans_payable': _float(loans_payable), 'loans_given': _float(loans_given_total), 'loans_received': _float(loans_received_total), 'loan_collections': _float(loan_collections_total), 'loan_repayments': _float(loan_repayments_total), 'cash_in_breakdown': [ _breakdown_entry('Sales Payments', sales_payments_total, sale_payments.count()), _breakdown_entry('Loans Received', loans_received_total, loans_received.count()), _breakdown_entry( 'Loan Collections', loan_collections_total, collections_on_given.count()), ], 'cash_out_breakdown': [ _breakdown_entry( 'Purchase Payments', purchase_payments_total, purchase_payments.count()), _breakdown_entry('Expenses', expense_total, expenses.count()), _breakdown_entry('Payroll', payroll_total, payroll.count()), _breakdown_entry('Advances', advance_total, advances.count()), _breakdown_entry('Return Refunds', return_refunds_total, returns.filter( refund_amount__gt=0).count()), _breakdown_entry( 'Withdrawals', withdrawal_total, withdrawals.count()), _breakdown_entry( 'Loans Given', loans_given_total, loans_given.count()), _breakdown_entry( 'Loan Repayments', loan_repayments_total, repayments_on_received.count()), ], 'loan_breakdown': [ _breakdown_entry( 'Loans Given', loans_given_total, loans_given.count()), _breakdown_entry('Loans Received', loans_received_total, loans_received.count()), _breakdown_entry( 'Collections (Loan Out)', loan_collections_total, collections_on_given.count()), _breakdown_entry( 'Repayments (Loan In)', loan_repayments_total, repayments_on_received.count()), ], 'expense_breakdown': [ _breakdown_entry('Purchases', purchase_total, purchases.count()), _breakdown_entry('Operating Expenses', expense_total, expenses.count()), _breakdown_entry('Payroll', payroll_total, payroll.count()), _breakdown_entry('Advances', advance_total, advances.count()), ], } return snapshot def _currency_row_has_activity(row): """True if a per-currency row has non-zero activity in the selected period.""" keys = ( 'total_revenue', 'total_returns', 'total_income', 'cash_from_sales', 'credit_sales', 'total_purchases', 'operating_expenses', 'total_payroll', 'total_advances', 'total_expenses', 'net_profit', 'total_cash_in', 'total_cash_out', 'net_cash_flow', 'accounts_receivable', 'accounts_payable', 'loans_receivable', 'loans_payable', 'loans_given', 'loans_received', 'loan_collections', 'loan_repayments', ) return any(abs(row.get(k, 0) or 0) > 0.001 for k in keys) def get_base_currency_report(*, start_date, end_date): """ Consolidated report in the system base currency (AFN). All transactional amounts use stored *_base fields — no runtime conversion. """ from api.services.accounting.reports import ( REPORT_MODE_BASE, get_account_code_balance, get_profit_and_loss, ) base = CURRENCY_DEFINITIONS[BASE_CURRENCY_ID] summary = _build_base_summary(start_date, end_date) return { 'report_type': 'base', 'base_currency': { 'id': BASE_CURRENCY_ID, 'code': base['code'], 'symbol': base['symbol'], 'name': base['name'], }, 'currency': base['code'], 'currency_symbol': base['symbol'], 'summary': summary, 'cash_in_breakdown': summary.get('cash_in_breakdown', []), 'cash_out_breakdown': summary.get('cash_out_breakdown', []), 'expense_breakdown': summary.get('expense_breakdown', []), 'loan_breakdown': summary.get('loan_breakdown', []), } def get_by_currency_report(*, start_date, end_date): """ Per-currency report — native amounts only, no cross-currency conversion. Only currencies with activity in the period are included. """ raw = _build_currencies_report(start_date, end_date) by_currency = { code: row for code, row in raw.items() if _currency_row_has_activity(row) } # Sort by net income descending for clearer presentation sorted_codes = sorted( by_currency.keys(), key=lambda c: by_currency[c].get('total_income', 0), reverse=True, ) return { 'report_type': 'by_currency', 'by_currency': by_currency, 'currency_codes': sorted_codes, 'currency_count': len(by_currency), } def _item_revenue_base(item) -> Decimal: """Convert a sale line total to AFN using the parent sale's stored base ratio.""" from api.utils.packaging import sales_item_line_total sale = item.sales line_native = sales_item_line_total(item) if line_native <= 0: return Decimal('0') sale_total = _d(getattr(sale, 'total_amount', 0)) sale_base = _d(getattr(sale, 'total_amount_base', 0)) if sale_total > 0 and sale_base > 0: return (line_native * (sale_base / sale_total)).quantize(Decimal('0.0001')) rate = _d(getattr(sale, 'exchange_rate', 1) or 1) return (line_native * rate).quantize(Decimal('0.0001')) def _item_cogs_base(item) -> Decimal: from api.utils.packaging import sales_item_pieces pieces = sales_item_pieces(item) unit_cost = _d(getattr(item, 'unit_cost', 0)) if pieces <= 0 or unit_cost <= 0: return Decimal('0') # unit_cost is snapshotted in AFN (base); do not apply the sale FX rate again. return (pieces * unit_cost).quantize(Decimal('0.0001')) def get_top_products(*, start_date=None, end_date=None, limit=10): """Top sold products in period by quantity (pieces), with AFN revenue.""" from api.models.data.sales import SalesItems from api.utils.packaging import sales_item_pieces qs = SalesItems.objects.filter( sales__in=_period_sales(start_date, end_date), ).select_related('stock__product', 'sales') buckets = {} for item in qs.iterator(chunk_size=500): product = getattr(getattr(item, 'stock', None), 'product', None) if not product: continue pid = product.id bucket = buckets.get(pid) if bucket is None: bucket = { 'product_id': pid, 'name': product.name or '—', 'barcode': getattr(product, 'barcode', '') or '', 'quantity_pieces': Decimal('0'), 'revenue_base': Decimal('0'), 'cogs_base': Decimal('0'), 'line_count': 0, } buckets[pid] = bucket bucket['quantity_pieces'] += sales_item_pieces(item) bucket['revenue_base'] += _item_revenue_base(item) bucket['cogs_base'] += _item_cogs_base(item) bucket['line_count'] += 1 ranked = sorted( buckets.values(), key=lambda row: (row['quantity_pieces'], row['revenue_base']), reverse=True, )[: max(1, min(int(limit or 10), 50))] return [ { 'product_id': row['product_id'], 'name': row['name'], 'barcode': row['barcode'], 'quantity_pieces': _float(row['quantity_pieces']), 'revenue_base': _float(row['revenue_base']), 'cogs_base': _float(row['cogs_base']), 'gross_profit_base': _float(row['revenue_base'] - row['cogs_base']), 'line_count': row['line_count'], } for row in ranked ] def get_products_sold_by_month(*, start_date=None, end_date=None, limit_per_month=10): """ Calendar months → top-sold products. For a single calendar year in the period, always returns all 12 months (empty months included). Quantity is pieces (1 carton = 8); revenue is AFN. """ from calendar import month_name from api.models.data.sales import SalesItems from api.utils.packaging import PIECES_PER_CARTON, sales_item_pieces limit = max(1, min(int(limit_per_month or 10), 50)) def _month_slots(): if not start_date or not end_date: return [] start = start_date.date() if hasattr(start_date, 'date') else start_date end = end_date.date() if hasattr(end_date, 'date') else end_date if start.year == end.year: return [(start.year, m) for m in range(1, 13)] slots = [] y, m = start.year, start.month while (y, m) <= (end.year, end.month): slots.append((y, m)) if m == 12: y, m = y + 1, 1 else: m += 1 return slots def _rank_products(products): ranked = sorted( products, key=lambda p: (_d(p['quantity_pieces']), _d(p['revenue_base'])), reverse=True, ) top = ranked[:limit] total_cartons = sum((_d(p['cartons']) for p in ranked), Decimal('0')) total_pieces = sum((_d(p['quantity_pieces']) for p in ranked), Decimal('0')) total_revenue = sum((_d(p['revenue_base']) for p in ranked), Decimal('0')) return { 'product_count': len(ranked), 'top_product': top[0] if top else None, 'total_cartons': _float(total_cartons), 'total_pieces': _float(total_pieces), 'total_revenue_base': _float(total_revenue), 'products': top, '_all_products': ranked, } def _merge_products(product_lists, *, limit_n): merged = {} for products in product_lists: for p in products or []: pid = p.get('product_id') if pid is None: continue bucket = merged.get(pid) if bucket is None: bucket = { 'product_id': pid, 'name': p.get('name') or '—', 'barcode': p.get('barcode') or '', 'quantity_pieces': Decimal('0'), 'revenue_base': Decimal('0'), 'line_count': 0, } merged[pid] = bucket bucket['quantity_pieces'] += _d(p.get('quantity_pieces')) bucket['revenue_base'] += _d(p.get('revenue_base')) bucket['line_count'] += int(p.get('line_count') or 0) ranked = sorted( merged.values(), key=lambda row: (row['quantity_pieces'], row['revenue_base']), reverse=True, )[:limit_n] out = [] for row in ranked: cartons = ( (row['quantity_pieces'] / PIECES_PER_CARTON).quantize(Decimal('0.0001')) if row['quantity_pieces'] > 0 else Decimal('0') ) out.append( { 'product_id': row['product_id'], 'name': row['name'], 'barcode': row['barcode'], 'quantity_pieces': _float(row['quantity_pieces']), 'cartons': _float(cartons), 'revenue_base': _float(row['revenue_base']), 'line_count': row['line_count'], } ) return out def _month_block(month_product_map, slots): months = [] all_products = [] for year, month in slots: products = month_product_map.get((year, month), []) ranked = _rank_products(products) all_products.append(ranked.pop('_all_products')) months.append( { 'year': year, 'month': month, 'month_label': month_name[month].lower(), **ranked, } ) top_products = _merge_products(all_products, limit_n=limit) return { 'months': months, 'totals': { 'month_count': len(slots), 'months_with_sales': sum(1 for m in months if m['product_count'] > 0), 'total_cartons': _float( sum((_d(m['total_cartons']) for m in months), Decimal('0')) ), 'total_pieces': _float( sum((_d(m['total_pieces']) for m in months), Decimal('0')) ), 'total_revenue_base': _float( sum((_d(m['total_revenue_base']) for m in months), Decimal('0')) ), 'top_product': top_products[0] if top_products else None, }, 'top_products': top_products, } slots = _month_slots() if not slots: return { 'currency': 'AFN', 'year': None, 'months': [], 'totals': { 'month_count': 0, 'total_cartons': 0.0, 'total_pieces': 0.0, 'total_revenue_base': 0.0, }, 'limit_per_month': limit, } qs = SalesItems.objects.filter( sales__in=_period_sales(start_date, end_date), ).select_related('stock__product', 'sales') month_product_map = {} slot_set = set(slots) for item in qs.iterator(chunk_size=500): sale = getattr(item, 'sales', None) product = getattr(getattr(item, 'stock', None), 'product', None) if not sale or not product or not getattr(sale, 'sale_date', None): continue dt = sale.sale_date year = dt.year month = dt.month if (year, month) not in slot_set: continue plist = month_product_map.setdefault((year, month), {}) pid = product.id row = plist.get(pid) if row is None: row = { 'product_id': pid, 'name': product.name or '—', 'barcode': getattr(product, 'barcode', '') or '', 'quantity_pieces': Decimal('0'), 'revenue_base': Decimal('0'), 'line_count': 0, } plist[pid] = row row['quantity_pieces'] += sales_item_pieces(item) row['revenue_base'] += _item_revenue_base(item) row['line_count'] += 1 # Convert product dicts to float rows with cartons converted_map = {} for key, products in month_product_map.items(): rows = [] for row in products.values(): cartons = ( (row['quantity_pieces'] / PIECES_PER_CARTON).quantize(Decimal('0.0001')) if row['quantity_pieces'] > 0 else Decimal('0') ) rows.append( { 'product_id': row['product_id'], 'name': row['name'], 'barcode': row['barcode'], 'quantity_pieces': _float(row['quantity_pieces']), 'cartons': _float(cartons), 'revenue_base': _float(row['revenue_base']), 'line_count': row['line_count'], } ) converted_map[key] = rows months_block = _month_block(converted_map, slots) grand_cartons = _d(months_block['totals']['total_cartons']) grand_pieces = _d(months_block['totals']['total_pieces']) grand_revenue = _d(months_block['totals']['total_revenue_base']) year = slots[0][0] if len({y for y, _ in slots}) == 1 else None return { 'currency': 'AFN', 'year': year, 'month_slots': [ {'year': y, 'month': m, 'month_label': month_name[m].lower()} for y, m in slots ], 'months': months_block, 'totals': { 'month_count': len(slots), 'total_cartons': _float(grand_cartons), 'total_pieces': _float(grand_pieces), 'total_revenue_base': _float(grand_revenue), }, 'limit_per_month': limit, } def get_expense_by_category(*, start_date=None, end_date=None): """Operating expenses rolled up by ExpenseCategory (AFN base).""" from django.db.models import Count expenses = _period_expenses(start_date, end_date) rows = ( expenses.values('category_id', 'category__name') .annotate(total=Sum('amount_base'), count=Count('id')) .order_by('-total') ) return [ { 'category_id': row['category_id'], 'name': row['category__name'] or 'Uncategorized', 'amount': _float(row['total']), 'count': row['count'], } for row in rows if _d(row['total']) > 0 ] def get_management_report(*, start_date=None, end_date=None, top_limit=10): """ Professional management report (AFN): - Accrual income & profit from GL P&L - Cash vs credit collections from operational data - Financing (loans) kept separate from P&L income - Top selling products """ from api.services.accounting.reports import REPORT_MODE_BASE, get_profit_and_loss base = _build_base_summary(start_date, end_date) ledger_pl = get_profit_and_loss( start_date=start_date, end_date=end_date, mode=REPORT_MODE_BASE, ) sales_total = _d(base.get('sales_total', 0)) returns_total = _d(base.get('total_returns', 0)) cash_from_sales = _d(base.get('cash_from_sales', 0)) credit_sales = _d(base.get('credit_sales', 0)) payroll_total = _d(base.get('total_payroll', 0)) cogs = _d(ledger_pl['cogs']) operating_exp = _d(ledger_pl['expenses']) revenue_gl = _d(ledger_pl['revenue']) gross_profit = _d(ledger_pl['gross_profit']) net_profit = _d(ledger_pl['net_income']) expense_by_category = get_expense_by_category( start_date=start_date, end_date=end_date) top_products = get_top_products( start_date=start_date, end_date=end_date, limit=top_limit, ) # Full expense story for charts (exclude advances from P&L expense chart) expense_chart = [ _breakdown_entry('COGS', cogs, 0), *[ {'type': row['name'], 'amount': row['amount'], 'count': row['count']} for row in expense_by_category ], ] if payroll_total > 0: expense_chart.append(_breakdown_entry('Payroll', payroll_total, 0)) # If category rollup is empty but GL has operating expenses, show GL residual category_sum = sum((_d(row.get('amount', 0)) for row in expense_by_category), Decimal('0')) residual_op = operating_exp - category_sum if residual_op > Decimal('0.5'): expense_chart.append(_breakdown_entry( 'Other Operating (GL)', residual_op, 0)) expense_chart = [e for e in expense_chart if _d(e['amount']) > 0] income_breakdown = [ _breakdown_entry('Sales (invoices)', sales_total, 0), _breakdown_entry( 'Returns', -returns_total if returns_total else Decimal('0'), 0), ] income_breakdown = [ e for e in income_breakdown if _d(e['amount']) != 0 ] collection_breakdown = [ _breakdown_entry('Cash Collected', cash_from_sales, 0), _breakdown_entry('On Credit (Unpaid)', credit_sales, 0), ] gl_expense_accounts = [ { 'code': row.get('code'), 'name': row.get('name'), 'amount': _float(row.get('amount', 0)), } for row in (ledger_pl.get('expense_accounts') or []) if _d(row.get('amount', 0)) > 0 ] gl_revenue_accounts = [ { 'code': row.get('code'), 'name': row.get('name'), 'amount': _float(row.get('amount', 0)), } for row in (ledger_pl.get('revenue_accounts') or []) if _d(row.get('amount', 0)) != 0 ] profit_bridge = [ {'label': 'Revenue', 'amount': _float(revenue_gl)}, {'label': 'COGS', 'amount': _float(-cogs)}, {'label': 'Gross Profit', 'amount': _float(gross_profit)}, {'label': 'Operating Expenses', 'amount': _float(-operating_exp)}, {'label': 'Net Profit', 'amount': _float(net_profit)}, ] base_meta = get_currency( BASE_CURRENCY_ID) or CURRENCY_DEFINITIONS[BASE_CURRENCY_ID] return { 'report_type': 'management', 'currency': base_meta.get('code', 'AFN'), 'currency_symbol': base_meta.get('symbol', '؋'), 'base_currency': { 'id': BASE_CURRENCY_ID, 'code': base_meta.get('code', 'AFN'), 'symbol': base_meta.get('symbol', '؋'), 'name': base_meta.get('name', 'Afghani'), }, 'kpis': { 'accrual_revenue': _float(revenue_gl), 'sales_invoices': _float(sales_total), 'returns': _float(returns_total), 'cash_from_sales': _float(cash_from_sales), 'credit_sales': _float(credit_sales), 'total_purchases': base.get('total_purchases', 0), 'cogs': _float(cogs), 'gross_profit': _float(gross_profit), 'operating_expenses': _float(operating_exp), 'total_expenses': _float(operating_exp + cogs), 'net_profit': _float(net_profit), 'total_cash_in': base.get('total_cash_in', 0), 'total_cash_out': base.get('total_cash_out', 0), 'net_cash_flow': base.get('net_cash_flow', 0), 'cash_balance': base.get('cash_balance', 0), 'accounts_receivable': base.get('accounts_receivable', 0), 'accounts_payable': base.get('accounts_payable', 0), }, 'income_breakdown': income_breakdown, 'collection_breakdown': collection_breakdown, 'expense_breakdown': expense_chart, 'expense_by_category': expense_by_category, 'gl_revenue_accounts': gl_revenue_accounts, 'gl_expense_accounts': gl_expense_accounts, 'profit_bridge': profit_bridge, 'cash_in_breakdown': base.get('cash_in_breakdown', []), 'cash_out_breakdown': base.get('cash_out_breakdown', []), 'financing': { 'loans_received': base.get('loans_received', 0), 'loans_given': base.get('loans_given', 0), 'loan_collections': base.get('loan_collections', 0), 'loan_repayments': base.get('loan_repayments', 0), 'loans_receivable': base.get('loans_receivable', 0), 'loans_payable': base.get('loans_payable', 0), 'loan_breakdown': base.get('loan_breakdown', []), 'note': 'Loans are financing/cash movement — not sales income or P&L profit.', }, 'top_products': top_products, } def _company_ops_counts(*, start_date=None, end_date=None): """Customer count + period sales count for the company.""" from api.models.data.customers import Customer customers_qs = Customer.objects.filter(is_active=True) sales_qs = _period_sales(start_date, end_date) return { 'customer_count': customers_qs.count(), 'sales_count': sales_qs.count(), } def _sum_kpis(rows: list[dict]) -> dict: """Sum numeric KPI fields across rows.""" keys = [ 'accrual_revenue', 'sales_invoices', 'returns', 'cash_from_sales', 'credit_sales', 'cogs', 'gross_profit', 'operating_expenses', 'total_expenses', 'net_profit', 'total_cash_in', 'total_cash_out', 'net_cash_flow', 'cash_balance', 'accounts_receivable', 'accounts_payable', 'customer_count', 'sales_count', ] totals = {k: 0.0 for k in keys} for row in rows: kpis = row.get('kpis') or {} for k in keys: totals[k] += float(kpis.get(k) or 0) return totals def _merge_breakdowns(rows: list[dict], field: str) -> list[dict]: """Merge named breakdown rows by type/name.""" buckets: dict[str, dict] = {} for row in rows: for item in row.get(field) or []: key = item.get('type') or item.get('name') or '—' bucket = buckets.get(key) if bucket is None: bucket = {'type': key, 'amount': 0.0, 'count': 0} buckets[key] = bucket bucket['amount'] += float(item.get('amount') or 0) bucket['count'] += int(item.get('count') or 0) return sorted(buckets.values(), key=lambda x: x['amount'], reverse=True) def _merge_top_products(rows: list[dict], limit: int = 10) -> list[dict]: buckets: dict[int, dict] = {} for row in rows: for item in row.get('top_products') or []: pid = item.get('product_id') if pid is None: continue bucket = buckets.get(pid) if bucket is None: bucket = { 'product_id': pid, 'name': item.get('name') or '—', 'barcode': item.get('barcode') or '', 'quantity_pieces': 0.0, 'revenue_base': 0.0, 'cogs_base': 0.0, 'gross_profit_base': 0.0, 'line_count': 0, } buckets[pid] = bucket bucket['quantity_pieces'] += float( item.get('quantity_pieces') or 0) bucket['revenue_base'] += float(item.get('revenue_base') or 0) bucket['cogs_base'] += float(item.get('cogs_base') or 0) bucket['gross_profit_base'] += float( item.get('gross_profit_base') or 0) bucket['line_count'] += int(item.get('line_count') or 0) ranked = sorted( buckets.values(), key=lambda r: (r['quantity_pieces'], r['revenue_base']), reverse=True, ) return ranked[: max(1, min(int(limit or 10), 50))] def _build_owner_row(*, start_date=None, end_date=None, top_limit=8): """Full owner KPI + breakdowns for the company.""" report = get_management_report( start_date=start_date, end_date=end_date, top_limit=top_limit, ) counts = _company_ops_counts( start_date=start_date, end_date=end_date, ) kpis = dict(report.get('kpis') or {}) kpis.update(counts) return { 'kpis': kpis, 'expense_breakdown': report.get('expense_breakdown') or [], 'expense_by_category': report.get('expense_by_category') or [], 'collection_breakdown': report.get('collection_breakdown') or [], 'income_breakdown': report.get('income_breakdown') or [], 'profit_bridge': report.get('profit_bridge') or [], 'cash_in_breakdown': report.get('cash_in_breakdown') or [], 'cash_out_breakdown': report.get('cash_out_breakdown') or [], 'financing': report.get('financing') or {}, 'top_products': report.get('top_products') or [], } def get_owner_detail(*, start_date=None, end_date=None, top_limit=10): """Owner detail report for the company.""" from api.models.data.currency import get_currency row = _build_owner_row( start_date=start_date, end_date=end_date, top_limit=top_limit, ) base_meta = get_currency( BASE_CURRENCY_ID) or CURRENCY_DEFINITIONS[BASE_CURRENCY_ID] return { 'report_type': 'owner_detail', 'currency': base_meta.get('code', 'AFN'), 'currency_symbol': base_meta.get('symbol', '؋'), 'base_currency': { 'id': BASE_CURRENCY_ID, 'code': base_meta.get('code', 'AFN'), 'symbol': base_meta.get('symbol', '؋'), 'name': base_meta.get('name', 'Afghani'), }, 'company': row, } def get_owner_report(*, start_date=None, end_date=None, top_limit=5): """ Owner / super-admin report: company-wide KPIs + totals. """ from api.models.data.currency import get_currency row = _build_owner_row( start_date=start_date, end_date=end_date, top_limit=top_limit, ) base_meta = get_currency( BASE_CURRENCY_ID) or CURRENCY_DEFINITIONS[BASE_CURRENCY_ID] return { 'report_type': 'owner_report', 'currency': base_meta.get('code', 'AFN'), 'currency_symbol': base_meta.get('symbol', '؋'), 'base_currency': { 'id': BASE_CURRENCY_ID, 'code': base_meta.get('code', 'AFN'), 'symbol': base_meta.get('symbol', '؋'), 'name': base_meta.get('name', 'Afghani'), }, 'company_totals': { 'kpis': row['kpis'], 'expense_breakdown': row['expense_breakdown'], 'collection_breakdown': row['collection_breakdown'], 'income_breakdown': row['income_breakdown'], 'top_products': row['top_products'], }, }