dodil/erp-brick-ar

DODIL ERP brick: accounts receivable

src/brick_ar/domain/provision.py

"""The doubtful-debt provision (AR-45): a month's allowance for debts that may not be paid, proposed by the biller from payments'
aging as at the period end and approved by someone else. The provision is on the outstanding EXCLUDING VAT (VAT is recovered by bad
debt relief, never provided). The books carry the allowance of the last approved run; approval posts only the MOVEMENT since then."""
import json
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from brick_ar.domain.dates import end_of_month
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.money import ZERO, money

SPECIFIC = "specific"          # the rate row for an invoice in dispute


def period_end(period_id: str) -> date:
    try:
        y, m = period_id.split("-")
        return end_of_month(date(int(y), int(m), 1))
    except (ValueError, TypeError):
        raise RuleBroken(f"{period_id!r} is not a period such as 2026-11") from None


def run_id(company_code: str, period_id: str) -> str:
    return f"prov:{company_code or 'default'}:{period_id}"


def compute(aging: dict, invoices: dict, rate_of) -> tuple:
    """(lines, total). `aging` is payments' {buckets: [{bucket, invoices: [{invoice_id, outstanding, disputed}]}]}; `invoices` maps an
    invoice_id to (net, gross, fx_rate) from the frozen invoice; `rate_of(bucket)` is the dated rate in force (None = 0%). An invoice in
    dispute is its own line at the specific rate (the bucket's when no specific rate exists) and is not in its bucket's sum."""
    lines, specific = [], []
    for b in aging.get("buckets", []):
        name, net_sum, n = b["bucket"], ZERO, 0
        for inv in b.get("invoices", []):
            known = invoices.get(inv["invoice_id"])
            if known is None:
                continue
            net, gross, fx = known
            outstanding_net = money(Decimal(str(inv["outstanding"])) * net / gross * (fx or 1)) if gross else ZERO
            if inv.get("disputed"):
                rate = rate_of(SPECIFIC)
                rate = rate_of(name) if rate is None else rate
                rate = rate or Decimal(0)
                specific.append({"kind": "specific", "invoice_id": inv["invoice_id"], "bucket": name, "outstanding_net": outstanding_net,
                                 "rate_pct": rate, "provision": money(outstanding_net * rate / 100)})
            else:
                net_sum += outstanding_net
                n += 1
        rate = rate_of(name) or Decimal(0)
        lines.append({"kind": "bucket", "bucket": name, "invoices": n, "outstanding_net": net_sum, "rate_pct": rate,
                      "provision": money(net_sum * rate / 100)})
    lines += sorted(specific, key=lambda x: x["invoice_id"])
    return lines, sum((x["provision"] for x in lines), ZERO)


@dataclass
class ProvisionRun:
    run_id: str
    company_code: str = ""
    period_id: str = ""
    lifecycle: str = "proposed"
    as_at: Optional[date] = None
    lines_json: str = "[]"
    total: Decimal = ZERO
    previous_total: Decimal = ZERO
    movement: Decimal = ZERO
    proposed_by: str = ""
    proposed_by_sub: str = ""
    approved_by: str = ""
    approved_at: Optional[datetime] = None
    rejected_reason: str = ""
    journal_id: Optional[int] = None
    events: list = field(default_factory=list, repr=False, compare=False)

    @property
    def lines(self) -> list:
        return json.loads(self.lines_json or "[]")

    def propose(self, *, lines: list, total: Decimal, previous_total: Decimal, by: str, by_sub: str) -> None:
        if self.lifecycle == "approved":
            raise RuleBroken(f"provision {self.run_id} is approved and final: the next month's run carries it on")
        self.lines_json = json.dumps(lines, default=str, sort_keys=True)
        self.total, self.previous_total, self.movement = total, previous_total, total - previous_total
        self.lifecycle, self.proposed_by, self.proposed_by_sub, self.rejected_reason = "proposed", by, by_sub, ""

    def approve(self, *, by: str, identities: tuple, second_approver: bool, at: datetime, previous_total: Decimal) -> None:
        if self.lifecycle == "approved":
            return
        if self.lifecycle != "proposed":
            raise RuleBroken(f"provision {self.run_id} is {self.lifecycle}: propose it again")
        if second_approver and (set(identities) & {(self.proposed_by or "").lower(), (self.proposed_by_sub or "").lower()} - {""}):
            raise RuleBroken("the proposer of a provision cannot approve it: another person must (second_approver is on)")
        if previous_total != self.previous_total:
            raise RuleBroken(f"the allowance carried in changed from {self.previous_total} to {previous_total} since this run was proposed: "
                             "propose it again")
        self.lifecycle, self.approved_by, self.approved_at = "approved", by, at

    def reject(self, *, reason: str) -> None:
        if self.lifecycle != "proposed":
            raise RuleBroken(f"provision {self.run_id} is {self.lifecycle}: only a proposed run is rejected")
        if not (reason or "").strip():
            raise RuleBroken("a rejection needs a reason")
        self.lifecycle, self.rejected_reason = "rejected", reason.strip()