dodil/erp-brick-payments

DODIL ERP brick: payments

src/brick_payments/domain/settlement.py

"""Settlement is DERIVED, never stored: outstanding = total - SUM(allocations), as at a date. One table of
allocations settles every invoice - receipts, credit notes and write-offs - so there is one SUM and no second
mechanism to keep in step. A refund pays money back out of a CREDIT, so it consumes a receipt's (or credit
note's) remainder instead of an invoice's outstanding: its allocation names the credit in `invoice_id`
("receipt:<payment_id>" / "credit_note:<id>")."""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Iterable, Optional

from brick_payments.domain.money import ZERO
from brick_payments.domain.values import SourceKind

CREDIT_PREFIX = {SourceKind.RECEIPT.value: "receipt:", SourceKind.CREDIT_NOTE.value: "credit_note:"}


@dataclass(frozen=True)
class Allocation:
    """One allocation row. An undo is a NEGATIVE row naming the row it reverses (reverses_seq)."""
    source_kind: str
    source_id: str
    invoice_id: str
    seq: int
    amount: Decimal
    currency: str
    allocated_on: date
    allocated_by: str = ""
    reason: str = ""
    reverses_seq: Optional[int] = None
    suggestion_id: str = ""
    decided_how: str = "manual"
    origin: str = ""
    bp_id: Optional[int] = None
    source_amount: Optional[Decimal] = None     # in the source's own currency, when it differs from the invoice's

    @property
    def key(self) -> tuple:
        return (self.source_kind, self.source_id, self.invoice_id, self.seq)

    @property
    def drawn(self) -> Decimal:
        """What this row takes out of its SOURCE (in the source's currency)."""
        return self.source_amount if self.source_amount is not None else self.amount


def credit_ref(kind: str, source_id: str) -> str:
    return f"{CREDIT_PREFIX[kind]}{source_id}"


def on_or_before(rows: Iterable[Allocation], as_at: Optional[date]):
    return [r for r in rows if as_at is None or r.allocated_on <= as_at]


def outstanding(total: Decimal, rows: Iterable[Allocation], as_at: Optional[date] = None) -> Decimal:
    """total - SUM(allocations on or before as_at). The rows are this invoice's, refunds excluded."""
    return total - sum((r.amount for r in on_or_before(rows, as_at) if r.source_kind != SourceKind.REFUND.value), ZERO)


def state_of(total: Decimal, out: Decimal, rows: Iterable[Allocation], as_at: Optional[date] = None) -> str:
    """open | part_paid | paid | written_off - all follow from the arithmetic."""
    rows = [r for r in on_or_before(rows, as_at) if r.source_kind != SourceKind.REFUND.value]
    if out <= ZERO and total > ZERO:
        return "written_off" if sum((r.amount for r in rows if r.source_kind == SourceKind.WRITEOFF.value), ZERO) > ZERO else "paid"
    return "part_paid" if sum((r.amount for r in rows), ZERO) > ZERO else "open"


def source_remainder(amount: Decimal, source_kind: str, source_id: str, rows: Iterable[Allocation],
                     as_at: Optional[date] = None) -> Decimal:
    """What a receipt (or credit note) still holds: its amount, less what it was allocated to invoices and what
    was refunded out of it. Reversals are negative rows, so they give it back."""
    ref = credit_ref(source_kind, source_id)
    drawn = ZERO
    for r in on_or_before(rows, as_at):
        if r.source_kind == source_kind and r.source_id == source_id:
            drawn += r.drawn
        elif r.source_kind == SourceKind.REFUND.value and r.invoice_id == ref:
            drawn += r.amount
    return amount - drawn


# ── aging ────────────────────────────────────────────────────────────────────────────────────────────────

def parse_buckets(spec: str) -> list:
    """"0-30,31-60,61-90,90+" -> [(0, 30), (31, 60), (61, 90), (90, None)]"""
    out = []
    for part in spec.split(","):
        part = part.strip()
        if part.endswith("+"):
            out.append((int(part[:-1]), None))
        else:
            lo, hi = part.split("-")
            out.append((int(lo), int(hi)))
    return out


def bucket_labels(spec: str) -> list:
    return [f"{lo}+" if hi is None else f"{lo}-{hi}" for lo, hi in parse_buckets(spec)]


def bucket_label(days: int, spec: str) -> str:
    """The bucket an age in days falls in, labelled as configured ('0-30', '90+'). A day past due starts at 1."""
    buckets = parse_buckets(spec)
    for lo, hi in buckets:
        if days >= lo and (hi is None or days <= hi):
            return f"{lo}+" if hi is None else f"{lo}-{hi}"
    lo, _ = buckets[-1]
    return f"{lo}+"


def age_days(due_date: Optional[date], issued_on: Optional[date], today: date, age_from: str) -> int:
    base = due_date if age_from == "due_date" else issued_on
    return (today - base).days if base else 0