dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

src/brick_purchasing/domain/bill.py

"""A supplier's bill or credit note: THEIR claim on us, not our document (PUR-10, PUR-11, PUR-12, PUR-17, PUR-18, PUR-27).

The natural key is (bp_id, supplier_invoice_no, invoice_date): the supplier's number is theirs, reused across years, never ours to
generate. An approved bill is never edited; a correction is a credit. Outstanding is derived (total minus allocations)."""
from dataclasses import dataclass, field
from datetime import date, datetime
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
from brick_purchasing.domain.values import ZERO


def bill_id_of(bp_id: int, supplier_invoice_no: str, invoice_date) -> str:
    return f"bill:{int(bp_id)}:{supplier_invoice_no.strip()}:{invoice_date}"


@dataclass
class BillLine:
    bill_id: str
    line_no: int
    order_line_no: Optional[int] = None
    product_id: str = ""
    description: str = ""
    quantity: Decimal = ZERO
    unit_code: str = ""
    unit_numerator: int = 1
    unit_denominator: int = 1
    quantity_base: Decimal = ZERO
    pricing_unit_code: str = ""
    pricing_numerator: int = 1
    pricing_denominator: int = 1
    pricing_quantity: int = 1
    unit_price: Decimal = ZERO
    net_amount: Decimal = ZERO
    tax_category: str = ""
    tax_code: str = ""
    tax_rate: Decimal = ZERO
    tax_amount: Decimal = ZERO
    vat_reclaimable: Decimal = ZERO
    expense_category: str = "general"
    cis_part: str = "none"
    withdrawal_id: str = ""
    delivery_cost_ref: str = ""
    fa_transaction_id: str = ""
    asset_id: str = ""
    cost_centre: str = ""
    project_id: str = ""
    account_assignment: str = ""
    item_category: str = ""
    ic_reference: str = ""
    posting_kind: str = "expense"             # expense | grni | asset | consignment | delivery_cost (how the ledger is debited)


@dataclass
class VarianceAcceptance:
    bill_id: str
    line_no: int
    kind: str
    key: str = ""
    difference: str = ""
    tolerance_in_force: str = ""
    accepted_by_email: str = ""
    accepted_by_sub: str = ""
    accepted_on: Optional[date] = None
    reason: str = ""


@dataclass
class BilledClaim:
    purchase_order_id: str
    line_no: int
    bill_id: str
    quantity_base: Decimal = ZERO
    claimed_at: Optional[datetime] = None


@dataclass
class SupplierBill:
    bill_id: str
    bp_id: int = 0
    supplier_invoice_no: str = ""
    invoice_date: Optional[date] = None
    kind: str = "bill"                         # bill | credit | down_payment_invoice
    self_billed: bool = False
    corrects_bill_id: str = ""
    purchase_order_id: str = ""
    company_code: str = ""
    currency: str = "GBP"
    subtotal: Decimal = ZERO
    tax_total: Decimal = ZERO
    total: Decimal = ZERO
    supplier_tax_total: Optional[Decimal] = None
    due_date: Optional[date] = None
    terms_id: str = ""
    status: str = "recorded"
    bill_no: str = ""
    recorded_by_email: str = ""
    recorded_by_sub: str = ""
    recorded_on: Optional[date] = None
    approved_by_email: str = ""
    approved_by_sub: str = ""
    approved_on: Optional[date] = None
    source_kind: str = "person"                # person | odoo | intercompany | ers
    source_id: str = ""
    dispute_reason: str = ""
    dispute_by: str = ""
    resolve_outcome: str = ""
    cancel_reason: str = ""
    tax_ref: str = ""                         # the tax document_ref kept with the frozen answer
    tax_frozen: bool = False
    posted_journal_id: str = ""
    post_note: str = ""
    partner_company: str = ""
    ic_reference: str = ""
    down_payment_id: str = ""
    pre_dispute: str = ""
    supplier_vat_number: str = ""             # v8 (PUR-28): the particulars as the supplier printed them (the clerk may edit what was defaulted)
    supplier_address: str = ""                # the address object (shared.yaml conventions.address) as JSON, "" when none is known
    lines: list = field(default_factory=list, repr=False)
    acceptances: list = field(default_factory=list, repr=False)
    claims: list = field(default_factory=list, repr=False)
    events: list = field(default_factory=list, repr=False)

    @property
    def recorder_ids(self) -> frozenset:
        return identities(self.recorded_by_email, self.recorded_by_sub)

    def line(self, n: int) -> Optional[BillLine]:
        return next((x for x in self.lines if x.line_no == int(n)), None)

    def accepted(self, line_no: int, kind: str) -> bool:
        return any(a.line_no == line_no and a.kind == kind for a in self.acceptances)

    def recompute_totals(self) -> None:
        self.subtotal = sum((x.net_amount for x in self.lines), ZERO)
        self.tax_total = sum((x.tax_amount for x in self.lines), ZERO)
        self.total = self.subtotal + self.tax_total

    # ── accepting a difference ─────────────────────────────────────────────────────────────
    def accept_variance(self, line_no: int, kind: str, *, email, sub, reason, today, difference: str = "", tolerance: str = "",
                        key: str = "", by_system: bool = False) -> VarianceAcceptance:
        if self.status not in ("recorded",):
            raise RuleBroken(f"the bill is {self.status}: a difference is accepted while it is recorded")
        if not by_system:
            if not (reason or "").strip():
                raise RuleBroken("accepting a difference needs a reason")
            if same_person(identities(email, sub), self.recorder_ids) or not identities(email, sub):
                raise GateUnmet([no("second_person", "you recorded this bill (same email or same sign-in)", "another approver")])
        acc = VarianceAcceptance(self.bill_id, int(line_no), kind, key, difference, tolerance, (email or "").lower(), (sub or "").lower(),
                                 today, (reason or "").strip())
        self.acceptances = [a for a in self.acceptances if (a.line_no, a.kind) != (acc.line_no, acc.kind)] + [acc]
        return acc

    # ── hold, release, cancel ─────────────────────────────────────────────────────────────
    def dispute(self, reason: str, by: str) -> None:
        if self.status not in ("recorded", "approved"):
            raise RuleBroken(f"a {self.status} bill cannot be disputed")
        if not (reason or "").strip():
            raise RuleBroken("disputing a bill needs a reason")
        self.pre_dispute = self.status
        self.status, self.dispute_reason, self.dispute_by = "disputed", reason.strip(), by

    def resolve(self, outcome: str, was_approved: bool) -> None:
        if self.status != "disputed":
            raise RuleBroken("only a disputed bill has a dispute to resolve")
        if not (outcome or "").strip():
            raise RuleBroken("resolving a dispute needs an outcome")
        self.resolve_outcome = outcome.strip()
        self.status = "approved" if was_approved else "recorded"

    def cancel(self, reason: str, allocated: bool) -> None:
        if self.status not in ("recorded", "disputed") or self.bill_no:
            raise RuleBroken(f"a {self.status} bill cannot be cancelled: only one never approved")
        if not (reason or "").strip():
            raise RuleBroken("cancelling a bill needs a reason")
        if allocated:
            raise RuleBroken("a bill with an allocation cannot be cancelled")
        self.status, self.cancel_reason = "cancelled", reason.strip()