dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

src/brick_purchasing/application/intercompany.py

"""Intercompany bills (PUR-64): ar issues an invoice from one of our companies to another; the receiving company's purchasing records it as a bill from
the partner company's business partner, keyed ic:<invoice_id> (a repeat answers the same bill), with ic_reference on every line and ar's frozen tax mapped
to the purchase side (the amounts are ar's, never recomputed here). It is approved like any bill and posts with partner_company and ic_reference; gl reads
what is open between the pair through GET /intercompany/open."""
from typing import Optional

from erp_platform.bricks import Refused
from erp_platform.types import IsoDate, text

from brick_purchasing.application import base, bills
from brick_purchasing.application.base import BILL_READ, BILL_WRITE, PURCHASING, _decide, unit
from brick_purchasing.application.inputs import BillLineIn, Code, Currency, Money, OptCode, Price, Qty, Ref, _In
from brick_purchasing.infrastructure.seams import Missing, ports

repository = lambda: base.repository()          # noqa: E731


class IcLineIn(_In):
    line_no: int
    description: Optional[text(300)] = None
    quantity: Qty
    unit_code: Code
    unit_price: Price
    net_amount: Money
    tax_code: Optional[OptCode] = None
    tax_amount: Money


@PURCHASING.use_case(kind="command", route="POST /bills/intercompany", permission=BILL_WRITE, idempotent=True)
def record_intercompany_bill(caller: dict, company_code: text(4, min_length=1), bill_id: Ref, supplier_company_code: text(4, min_length=1), supplier_invoice_no: Ref,
                             issue_date: IsoDate, tax_point_date: IsoDate, due_date: IsoDate, currency: Currency, lines: list[IcLineIn]) -> dict:
    """Record the bill another of our companies sent: a bill of company_code from supplier_company_code's business partner, status recorded, keyed
    bill_id (ic:<invoice_id>); a second call with the same bill_id returns the same bill. ar's tax amounts are kept as given. A company that is not
    a different company of these books is refused 409 {gate: company_known}."""
    if not bill_id.startswith("ic:"):
        raise Refused(422, "bill_id is ic:<invoice_id>")
    if not lines:
        raise Refused(422, "an intercompany bill has at least one line")

    def run():
        with repository() as repo, _decide():
            have = repo.bills.get(bill_id)
            if have:
                return {"bill_id": bill_id, "status": have[0].status, "replay": True}
            if company_code == supplier_company_code:
                raise Refused(409, "company_known: the bill's company and the sending company are the same; fix: a recharge is between two companies")
            try:
                bp = ports.bp.promote(caller, bp_key=f"company:{supplier_company_code}", legal_name=supplier_company_code)["bp_id"]
            except Missing as m:
                raise Refused(409, f"company_known: {supplier_company_code} is not a business partner of these books ({m.reason}); fix: flag it as an intercompany partner") from None
            idate = base._day(issue_date, "issue_date")
            rows = [BillLineIn(quantity=x.quantity, unit_code=x.unit_code, unit_price=x.unit_price, net_amount=x.net_amount, description=x.description,
                               tax_code=x.tax_code) for x in lines]
            built = bills.build_bill_lines(repo, caller, None, rows, bill_id, bp, base._currency(currency))
            for ln, x in zip(built, lines):
                ln.tax_amount, ln.vat_reclaimable, ln.tax_code = x.tax_amount, x.tax_amount, x.tax_code or ""
            b, _v, _c = bills.record_core(repo, caller, bp_id=bp, supplier_invoice_no=supplier_invoice_no, invoice_date=idate, currency=base._currency(currency), lines=built,
                                          due_date=base._day(due_date, "due_date"), company_code=company_code, source_kind="intercompany", source_id=bill_id,
                                          partner_company=supplier_company_code, ic_reference=bill_id.split(":", 1)[1], tax_lines_prebuilt=True)
            b.bill_id = bill_id
            for ln in b.lines:
                ln.bill_id = bill_id
            repo.bills.save(b, 0)
            return {"bill_id": bill_id, "status": b.status, "replay": False}
    return unit(None, run)


@PURCHASING.use_case(kind="query", route="GET /intercompany/open", permission=BILL_READ, tool=False)
def intercompany_open(caller: dict, company_code: text(4, min_length=1), trading_partner: text(4, min_length=1), as_at: IsoDate) -> dict:
    """The intercompany bills of a company from a trading partner still open on a date, each with its ic_reference and outstanding: gl matches the
    pair on ic_reference first."""
    with repository() as repo:
        d = base._day(as_at, "as_at")
        items = {i["bill_id"]: i for i in bills.open_items(repo, d, None)}
        docs = []
        for h in repo.bills.repo.all_heads():
            if h.source_kind == "intercompany" and h.company_code == company_code and h.partner_company == trading_partner and h.bill_id in items:
                i = items[h.bill_id]
                docs.append({"bill_id": h.bill_id, "ic_reference": h.ic_reference, "currency": h.currency, "total": h.total, "outstanding": i["outstanding"]})
        return {"company_code": company_code, "trading_partner": trading_partner, "as_at": d, "documents": docs}