from dataclasses import dataclass from datetime import date from decimal import Decimal import re from uuid import UUID from django.db.models import DecimalField, F, Q, Sum, Value from django.db.models.functions import Coalesce from finance.models import Account, Transaction _CURRENCY_RE = re.compile(r"[A-Z]{3}", re.ASCII) _MONEY_FIELD = DecimalField(max_digits=24, decimal_places=2) _ZERO = Decimal("0.00") class ReportInputError(ValueError): def __init__(self, errors: dict[str, str]): super().__init__("Invalid report inputs") self.errors = errors @dataclass(frozen=True) class MonthlySummary: month: date currency: str inflow: Decimal expense: Decimal refund: Decimal net_spending: Decimal net_activity: Decimal @dataclass(frozen=True) class SpendingCategory: category_id: UUID | None group_name: str | None category_name: str is_group_total: bool expense: Decimal refund: Decimal net_spending: Decimal @dataclass(frozen=True) class AccountBalance: account_id: UUID account_name: str balance: Decimal def available_currencies() -> tuple[str, ...]: return tuple( Account.objects.order_by("currency") .values_list("currency", flat=True) .distinct() ) def monthly_summary(month: date, currency: str) -> MonthlySummary: month, currency, month_after = _validated_period(month, currency) totals = _monthly_rows(month, month_after, currency).aggregate( inflow=_sum("amount", Q(kind=Transaction.Kind.INCOME)), expense=_sum(-F("amount"), Q(kind=Transaction.Kind.EXPENSE)), refund=_sum("amount", Q(kind=Transaction.Kind.REFUND)), ) inflow = totals["inflow"] expense = totals["expense"] refund = totals["refund"] net_spending = expense - refund return MonthlySummary( month=month, currency=currency, inflow=inflow, expense=expense, refund=refund, net_spending=net_spending, net_activity=inflow - net_spending, ) def spending_by_category(month: date, currency: str) -> tuple[SpendingCategory, ...]: month, currency, month_after = _validated_period(month, currency) grouped = ( _monthly_rows(month, month_after, currency) .filter(kind__in=[Transaction.Kind.EXPENSE, Transaction.Kind.REFUND]) .values("category_id", "category__name", "category__group__name") .annotate( expense=_sum(-F("amount"), Q(kind=Transaction.Kind.EXPENSE)), refund=_sum("amount", Q(kind=Transaction.Kind.REFUND)), ) ) leaves = tuple( SpendingCategory( category_id=row["category_id"], group_name=row["category__group__name"], category_name=row["category__name"] or "Uncategorized", is_group_total=False, expense=row["expense"], refund=row["refund"], net_spending=row["expense"] - row["refund"], ) for row in grouped ) grouped_rows = [] for group_name in sorted({row.group_name for row in leaves if row.group_name}): children = [row for row in leaves if row.group_name == group_name] grouped_rows.append(SpendingCategory(None, group_name, group_name, True, sum((row.expense for row in children), _ZERO), sum((row.refund for row in children), _ZERO), sum((row.net_spending for row in children), _ZERO))) grouped_rows.extend(sorted(children, key=lambda row: (-row.net_spending, row.category_name))) grouped_rows.extend(sorted((row for row in leaves if not row.group_name), key=lambda row: (-row.net_spending, row.category_name))) return tuple(grouped_rows) def account_balances(month: date, currency: str) -> tuple[AccountBalance, ...]: month, currency, month_after = _validated_period(month, currency) del month rows = ( Account.objects.filter(currency=currency) .annotate( report_balance=Coalesce( Sum("transactions__amount", filter=Q(transactions__posting_date__lt=month_after)), Value(_ZERO), output_field=_MONEY_FIELD, ) ) .order_by("name", "id") ) return tuple( AccountBalance( account_id=row.pk, account_name=row.name, balance=row.report_balance, ) for row in rows ) def _monthly_rows(month: date, month_after: date, currency: str): return Transaction.objects.filter( posting_date__gte=month, posting_date__lt=month_after, account__currency=currency, ) def _sum(expression, condition: Q): return Coalesce( Sum(expression, filter=condition), Value(_ZERO), output_field=_MONEY_FIELD, ) def _validated_period(month: date, currency: str) -> tuple[date, str, date]: errors: dict[str, str] = {} if not isinstance(month, date): errors["month"] = "invalid_date" elif month.day != 1: errors["month"] = "must_be_first_day" elif month.year == date.max.year and month.month == 12: errors["month"] = "invalid_month" if not isinstance(currency, str) or _CURRENCY_RE.fullmatch(currency) is None: errors["currency"] = "invalid_currency" if errors: raise ReportInputError(errors) if month.month == 12: month_after = date(month.year + 1, 1, 1) else: month_after = date(month.year, month.month + 1, 1) return month, currency, month_after