dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

src/brick_purchasing/domain/payment.py

"""Payment runs and supplier payments (PUR-14 to PUR-16, PUR-19, PUR-29, PUR-30, PUR-60, PUR-66).

DODIL never moves the money: a run is APPROVED by a second person, written as a file a person uploads to the bank, and each payment is
'paid' only when the bank's reference and date are confirmed (a fact, today or earlier). A correction is a reversing row, never an edit."""
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional

from brick_purchasing.domain.approval import identities, same_person
from brick_purchasing.domain.errors import GateUnmet, RuleBroken
from brick_purchasing.domain.readiness import no, ok, require
from brick_purchasing.domain.values import ZERO


@dataclass
class RunItem:
    run_id: str
    payment_id: str
    bill_id: str
    amount: Decimal = ZERO                 # the gross taken from the bill's outstanding (negative for a credit netted in)
    kind: str = "bill"                     # bill | credit | down_payment
    labour_paid: Decimal = ZERO            # CIS: the labour share of this amount
    materials_paid: Decimal = ZERO
    recorded_by: str = ""                  # who recorded the bill: identities joined by '|' (email|sub)


@dataclass
class Allocation:
    source_kind: str                       # payment | credit | down_payment
    source_id: str
    bill_id: str
    seq: int
    amount: Decimal = ZERO                 # negative on a reversal
    allocated_on: Optional[date] = None
    allocated_by_email: str = ""
    reverses_seq: Optional[int] = None


@dataclass
class SupplierPayment:
    payment_id: str
    run_id: str = ""
    bp_id: int = 0
    currency: str = "GBP"
    amount: Decimal = ZERO                 # what the supplier is paid (gross less any CIS deduction)
    gross: Decimal = ZERO
    cis_deduction: Decimal = ZERO
    cis_rate_pct: Decimal = ZERO
    cis_status: str = ""
    cis_ref: str = ""
    cis_tax_month: str = ""
    bank_account_id: str = ""
    bank_reference: str = ""
    reference: str = ""                    # our reference on the bank file (R-.../118)
    paid_on: Optional[date] = None
    status: str = "planned"                # planned | paid | rejected
    kind: str = "run"                      # run | outside | down_payment | odoo
    payee_bank_valid_from: Optional[date] = None
    confirmed_by: str = ""
    reject_reason: str = ""
    posted_journal_id: str = ""
    post_note: str = ""
    source_kind: str = "person"            # person | odoo | bank
    version_note: str = ""
    events: list = field(default_factory=list, repr=False)

    def confirm(self, bank_reference: str, paid_on: date, today: date, by: str) -> bool:
        """Confirm what the bank paid. Returns True when this call confirmed it, False when it was already confirmed (a repeat
        answers the same and changes nothing)."""
        if self.status == "paid":
            return False
        if self.status != "planned":
            raise RuleBroken(f"payment_open: the payment is {self.status}")
        if not (bank_reference or "").strip():
            raise RuleBroken("confirming a payment needs the bank's reference")
        if paid_on > today:
            raise GateUnmet([no("not_future", "a payment cannot be dated after today", "confirm it on the day the bank shows it")])
        self.status, self.paid_on, self.bank_reference, self.confirmed_by = "paid", paid_on, bank_reference.strip(), by
        return True

    def reject(self, reason: str) -> None:
        if self.status != "paid" and self.status != "planned":
            raise RuleBroken(f"payment_open: the payment is {self.status}")
        if not (reason or "").strip():
            raise RuleBroken("rejecting a payment needs a reason")
        self.status, self.reject_reason = "rejected", reason.strip()


@dataclass
class PaymentRun:
    run_id: str
    bank_account_id: str = ""
    gl_account_id: str = ""
    currency: str = ""
    pay_date: Optional[date] = None
    file_format: str = ""
    status: str = "draft"                  # draft | approved | exported | done | cancelled
    proposed_by_email: str = ""
    proposed_by_sub: str = ""
    approved_by_email: str = ""
    approved_by_sub: str = ""
    approved_on: Optional[date] = None
    file_key: str = ""
    file_sha256: str = ""
    exported_on: Optional[date] = None
    cancel_reason: str = ""
    left_out: str = "[]"                   # JSON [{bill_id, reason}]
    items: list = field(default_factory=list, repr=False)
    events: list = field(default_factory=list, repr=False)

    @property
    def proposer_ids(self) -> frozenset:
        return identities(self.proposed_by_email, self.proposed_by_sub)

    def recorder_ids(self) -> frozenset:
        out = set()
        for i in self.items:
            out |= set(x for x in (i.recorded_by or "").split("|") if x)
        return frozenset(out)

    def approve_gates(self, approver: frozenset) -> list:
        sp = not approver or same_person(approver, self.proposer_ids) or same_person(approver, self.recorder_ids())
        return [no("second_person", "you proposed this run, or recorded a bill in it (same email or same sign-in)", "another approver")
                if sp else ok("second_person")]

    def approve(self, email: str, sub: str, today: date, extra_gates: list) -> None:
        if self.status != "draft":
            raise RuleBroken(f"the run is {self.status}: only a draft run is approved")
        require(self.approve_gates(identities(email, sub)) + list(extra_gates))
        self.status, self.approved_on = "approved", today
        self.approved_by_email, self.approved_by_sub = (email or "").lower(), (sub or "").lower()

    def cancel(self, reason: str) -> None:
        if self.status not in ("draft", "approved"):
            raise RuleBroken(f"the run is {self.status}: a run is cancelled before its file is written")
        if self.file_key:
            raise RuleBroken("a file was written for this run: it cannot be cancelled")
        if not (reason or "").strip():
            raise RuleBroken("cancelling a run needs a reason")
        self.status, self.cancel_reason = "cancelled", reason.strip()

    def mark_exported(self, key: str, sha: str, today: date) -> None:
        if self.status == "exported" and self.file_key:
            return
        if self.status != "approved":
            raise RuleBroken(f"the run is {self.status}: only an approved run is written as a file")
        self.status, self.file_key, self.file_sha256, self.exported_on = "exported", key, sha, today

    def finish_if_done(self, payments: list) -> bool:
        if self.status == "exported" and payments and all(p.status in ("paid", "rejected") for p in payments):
            self.status = "done"
            return True
        return False


@dataclass
class SupplierBankRow:
    bp_id: int
    valid_from: date
    iban_or_account: str = ""              # stored ENCRYPTED, shown masked
    bic_or_sort_code: str = ""
    account_name: str = ""
    status: str = "proposed"
    entered_by_email: str = ""
    entered_by_sub: str = ""
    approved_by_email: str = ""
    approved_by_sub: str = ""
    call_back_note: str = ""
    reject_reason: str = ""
    country: str = ""
    requested_by: str = ""                 # how the change was asked for (email, phone, letter): shown to the approver
    approved_on: Optional[date] = None


def mask(value: str) -> str:
    v = (value or "").replace(" ", "")
    return ("*" * max(len(v) - 4, 0) + v[-4:]) if v else ""