dodil/erp-brick-payments

DODIL ERP brick: payments

src/brick_payments/domain/allocation.py

"""Allocation: what settled what. The planner turns a person's (or the system's) lines into rows, refusing in
the readiness words of the spec. An undo is a reversing row; nothing is deleted or edited."""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional

from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO, money
from brick_payments.domain.readiness import Gate, fmt, met, refuse, unmet
from brick_payments.domain.settlement import Allocation


@dataclass(frozen=True)
class Line:
    """One thing to settle: an invoice, how much of it is outstanding, and how much to put against it."""
    invoice_id: str
    amount: Decimal
    outstanding: Decimal
    currency: str
    bp_id: Optional[int] = None


def within_remainder(amount: Decimal, remainder: Decimal, outstanding: Decimal, what: str = "receipt") -> Gate:
    if amount > remainder or amount > outstanding:
        return unmet("within_remainder",
                     f"Allocating {fmt(amount)}; the {what} has {fmt(remainder)} left and the invoice {fmt(outstanding)} outstanding",
                     "Allocate at most the smaller figure")
    return met("within_remainder")


def same_currency(source_currency: str, invoice_currency: str) -> Gate:
    if source_currency != invoice_currency:
        return unmet("same_currency", f"The receipt is {source_currency}, the invoice {invoice_currency}",
                     "Allocate a receipt in the invoice's currency")
    return met("same_currency")


def allocated_dated_ok(allocated_on: date, today: date) -> Gate:
    if allocated_on < today:
        return unmet("allocated_dated_today", f"An allocation dated {allocated_on} is before today ({today})",
                     "Allocations are dated the day they are recorded")
    return met("allocated_dated_today")


def plan(*, source_kind: str, source_id: str, source_currency: str, source_remainder: Decimal, lines: list,
         next_seq: dict, today: date, by: str, decided_how: str = "manual", suggestion_id: str = "",
         what: str = "receipt", allocated_on: Optional[date] = None, origin: str = "", reason: str = "") -> list:
    """The allocation rows for `lines`, or a refusal. `next_seq`: invoice_id -> the next seq of this (source, invoice)
    pair. Dated today (never earlier). The whole plan is checked before any row is returned."""
    if not lines:
        raise RuleBroken("Nothing to allocate. Name at least one invoice and an amount")
    seen = set()
    total = ZERO
    for ln in lines:
        if ln.invoice_id in seen:
            raise RuleBroken(f"Invoice {ln.invoice_id} appears twice. Put the whole amount for one invoice on one line")
        seen.add(ln.invoice_id)
        if money(ln.amount) <= ZERO:
            raise RuleBroken(f"An allocation to {ln.invoice_id} must be more than zero. Leave the invoice out instead")
    dated = allocated_on or today
    refuse(allocated_dated_ok(dated, today))
    rows = []
    for ln in lines:
        amount = money(ln.amount)
        refuse(same_currency(source_currency, ln.currency))
        # each line against what the source has left AFTER the lines before it
        refuse(within_remainder(amount, source_remainder - total, ln.outstanding, what))
        total += amount
        rows.append(Allocation(source_kind, source_id, ln.invoice_id, next_seq.get(ln.invoice_id, 1), amount,
                               ln.currency, dated, by, reason, None, suggestion_id, decided_how, origin, ln.bp_id))
    return rows


def reverse_row(original: Allocation, existing: list, *, next_seq: int, today: date, by: str, reason: str) -> Allocation:
    """The reversing row for `original`: negative, same source, dated today, naming the row it reverses."""
    if original.amount <= ZERO:
        raise RuleBroken("That row is itself a reversal. Allocate again instead")
    if any(r.reverses_seq == original.seq and (r.source_kind, r.source_id, r.invoice_id) ==
           (original.source_kind, original.source_id, original.invoice_id) for r in existing):
        raise RuleBroken(f"Allocation {original.seq} of {original.source_id} to {original.invoice_id} is already reversed. "
                         "Nothing to undo")
    if not reason.strip():
        raise RuleBroken("Undoing an allocation needs a reason. Say why, for example 'wrong invoice'")
    return Allocation(original.source_kind, original.source_id, original.invoice_id, next_seq, -original.amount,
                      original.currency, today, by, reason.strip(), original.seq, original.suggestion_id,
                      original.decided_how, "", original.bp_id,
                      -original.source_amount if original.source_amount is not None else None)