dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/application/use_cases_v4.py
"""v4 slice A: payment requests (AR-35, AR-36), the early-payment discount (AR-37, AR-38), statutory interest (AR-39) and the
intercompany invoice (AR-40, AR-41). Same brick, same conventions as use_cases.py: one declaration per route AND tool, the
permission on the declaration, the rules in domain/, the neighbours behind infrastructure/seams.py.
Every call reads `uc.repository` at call time, so a composition or a test that swaps the repository swaps it here too."""
import dataclasses
import uuid
from datetime import date, timedelta
from decimal import Decimal
from typing import Literal, Optional
from erp_platform import work
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, Name
from brick_ar.application import services as sv
from brick_ar.application import use_cases as uc
from brick_ar.application.inputs import Code, Currency, Money, Pct, Reason
from brick_ar.application.services import who
from brick_ar.domain import customer as customer_mod
from brick_ar.domain import finance
from brick_ar.domain.credit_note import CreditNote
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.invoice import Invoice, Line
from brick_ar.domain.money import ZERO, money
from brick_ar.domain.payment_request import PaymentRequest
from brick_ar.domain.readiness import GatesUnmet, met, unmet
from brick_ar.infrastructure import seams
AR = uc.AR
SETTLE_REQUEST, SETTLE_DISCOUNT, HAND_OVER = "ar:payment_request:settle", "ar:invoice:settle_discount", "ar:intercompany:hand_over"
def _refuse(gate: str, reason: str, fix: str = "") -> Refused:
"""A 422 that names its gate, the way an inbound seam answers: {gate, reason, fix}."""
return Refused(422, f"[{gate}] {reason}" + (f" (fix: {fix})" if fix else ""))
# ═══ payment requests (AR-35) ═══════════════════════════════════════════════════════════════════
def _request_view(repo, pr: PaymentRequest, version: int) -> dict:
out = {f.name: getattr(pr, f.name) for f in dataclasses.fields(pr) if f.name != "events"}
received = repo.requests.received(pr.request_id)
out.update(title=pr.title, received=received, remaining=pr.remaining(received), version=version,
owed=ZERO, posted_to_ledger=False, in_vat_return=False) # a request is shown as requested, never as owed
return out
def _visible_request(repo, request_id: str) -> tuple:
found = repo.requests.get(request_id)
if found is None:
raise Refused(404, f"no payment request {request_id}")
return found
@AR.use_case(kind="command", route="POST /payment-requests", permission=sv.WRITE, idempotent=True)
def raise_payment_request(caller: dict, bp_id: BigId, currency: Currency, net_amount: Money,
kind: Literal["down_payment", "pro_forma"] = "down_payment", sales_order_id: Optional[Name] = None,
request_no: Optional[int] = None, tax_category: Optional[Code] = None) -> dict:
"""Raise a payment request: for an order's deposit (sales-order calls it with sales_order_id and request_no, keyed
"dpr:<sales_order_id>:<request_no>", so asking twice answers the same request), or a pro forma for an order or a quote by hand
(a uuid key). Only where deposit_mode is request (a down payment) - a pro forma is always possible. The tax shown on it is
tax's, for display only. It is NOT a VAT invoice: it posts nothing and is in no VAT return."""
with uc.repository() as repo, uc._decide():
rid = f"dpr:{sales_order_id}:{request_no}" if sales_order_id and request_no is not None else f"dpr:{uuid.uuid4().hex}"
found = repo.requests.get(rid)
if found is not None:
return _request_view(repo, *found)
today = uc._today(repo)
s = repo.settings.in_force(today)
if kind == "down_payment" and s.deposit_mode != "request":
raise RuleBroken("this business invoices a deposit as a VAT invoice (deposit_mode invoice): asking for a deposit is "
"raise_draft kind deposit. Raise a pro forma, or ask the owner to switch deposit_mode to request")
uc._check_customer(repo, caller, bp_id)
shown = Line(line_no=1, description="Payment request", quantity=Decimal(1), unit_price=net_amount,
tax_category=tax_category or "standard")
shown.price()
out = seams.tax_compute([shown], bp_id=bp_id, tax_point=today, currency=currency, document_ref=rid)
vat = sum((r.tax_amount for r in out.quote.summary), ZERO) if out.quote is not None else ZERO
pr = PaymentRequest.raise_new(rid, by=who(caller), bp_id=bp_id, currency=currency, net_amount=net_amount, tax_amount=vat,
kind=kind, sales_order_id=sales_order_id or "", tax_category=tax_category or "standard",
company_code=s.default_company_code or "")
view = _request_view(repo, pr, repo.requests.save(pr, 0))
view["tax_pending"] = out.quote is None
return view
@AR.use_case(kind="command", route="POST /payment-requests/{request_id}/issue", permission=sv.ISSUE, idempotent=True)
def issue_payment_request(caller: dict, request_id: Name) -> dict:
"""Issue a drafted request: a billing profile in force today, a number from the DPR- series, titled "not a VAT invoice".
Nothing is posted and nothing enters a VAT return."""
def unit():
with uc.repository() as repo, uc._decide():
pr, current = _visible_request(repo, request_id)
if pr.lifecycle != "draft":
return _request_view(repo, pr, current)
today = uc._today(repo)
profile = sv.profile_for(repo, pr.bp_id, today)
gate = customer_mod.profile_gate(profile, today)
if not gate.met:
raise GatesUnmet("issue the payment request", [gate])
number, _sid = sv.claim_number(repo, repo.settings.in_force(today), "request", pr.request_id, today, work.now(),
company=pr.company_code)
pr.issue(number=number, today=today, due=sv.terms_due(repo, profile.terms_id, today))
return _request_view(repo, pr, repo.requests.save(pr, current))
return uc._claiming(unit)
@AR.use_case(kind="command", route="POST /payment-requests/{request_id}/cancel", permission=sv.WRITE, idempotent=True)
def cancel_payment_request(caller: dict, request_id: Name, reason: Reason) -> dict:
"""Cancel a draft or issued request while nothing has been received against it."""
with uc.repository() as repo, uc._decide():
pr, current = _visible_request(repo, request_id)
pr.cancel(reason, repo.requests.received(request_id))
return _request_view(repo, pr, repo.requests.save(pr, current) if pr.lifecycle == "cancelled" and current else current)
@AR.use_case(kind="query", route="GET /payment-requests/{request_id}", permission=sv.READ)
def get_payment_request(caller: dict, request_id: Name) -> dict:
"""One payment request: its title, amounts, what has been received and what remains. Never owed, never posted."""
with uc.repository() as repo:
return _request_view(repo, *_visible_request(repo, request_id))
@AR.use_case(kind="query", route="GET /payment-requests", permission=sv.READ)
def list_payment_requests(caller: dict, bp_id: Optional[BigId] = None) -> dict:
"""The payment requests, optionally for one customer."""
with uc.repository() as repo:
return {"requests": [_request_view(repo, *repo.requests.get(i)) for i in repo.requests.page(bp_id=bp_id)]}
# ═══ AR-36: payments settles a request ══════════════════════════════════════════════════════════
@AR.use_case(kind="command", route="POST /payment-requests/{request_id}/settle", permission=SETTLE_REQUEST, idempotent=True)
def settle_payment_request(caller: dict, request_id: Name, payment_id: Name, received_on: IsoDate, amount: Money,
currency: Currency) -> dict:
"""payments names a receipt that paid (part of) a request. A deposit invoice keyed "dep:<request_id>:<n>" is raised for the
amount RECEIVED - net and VAT split at the request's rates, tax point the received date (a document-dated fact), issue date
today - issued and numbered from the INV series at once, frozen by tax and posted in post mode. payments then allocates the
receipt to that invoice (this brick never writes a pay_ table). The same payment settles once (keyed on payment_id); a part
payment leaves the request open for the rest; more than what remains is refused (gate within_request)."""
def unit():
with uc.repository() as repo, uc._decide():
pr, current = _visible_request(repo, request_id)
prior = repo.requests.settlement_of(payment_id)
if prior is not None:
if prior["request_id"] != request_id:
raise Refused(409, f"payment {payment_id} already settled request {prior['request_id']}")
return _settled_view(repo, pr, prior["deposit_invoice_id"], prior["amount"]), None
if pr.lifecycle != "issued":
raise _refuse("within_request", f"request {pr.request_no or pr.request_id} is {pr.lifecycle}: only an issued request "
"is settled", "issue the request first")
if currency.upper() != pr.currency:
raise _refuse("within_request", f"the request is in {pr.currency}, the receipt in {currency}", "")
received = repo.requests.received(request_id)
left = pr.remaining(received)
if amount <= 0 or amount > left:
raise _refuse("within_request", f"{amount} is more than the {left} that remains of request {pr.request_no}",
"settle at most what remains")
today = uc._today(repo)
on = uc._date(received_on, "received_on")
if on > today:
raise _refuse("within_request", f"received_on {on} is in the future", "")
n = len(repo.requests.settlements(request_id)) + 1
dep_id = f"dep:{request_id}:{n}"
previous = sum((v[0].subtotal for v in repo.invoices.get_many(pr.deposit_ids).values()), ZERO)
net = pr.net_amount - previous if amount == left else money(amount * pr.net_amount / pr.total)
line = Line(line_no=1, description=f"Deposit against {pr.request_no}", quantity=Decimal(1), unit_price=net,
tax_category=pr.tax_category)
inv = Invoice.raise_draft(dep_id, by=who(caller), bp_id=pr.bp_id, currency=pr.currency, source_kind="payment_request",
source_id=dep_id, lines=[line], kind="deposit", tax_point_date=on, company_code=pr.company_code)
sv.draft_tax(inv, today)
version = repo.invoices.save(inv, 0)
gates, quote, profile = sv.issue_gates(repo, inv, today)
if not all(g.met for g in gates) or quote is None:
raise GatesUnmet("issue the deposit invoice", gates)
inv, version = uc._issue_now(repo, caller, inv, version, today, quote, profile)
repo.requests.record_settlement(payment_id, request_id, n, on, amount, dep_id, work.now()) # plain INSERT, never swallowed
pr.note_settled(dep_id, received + amount)
repo.requests.save(pr, current)
return _settled_view(repo, pr, dep_id, amount, received + amount), dep_id
view, dep_id = uc._claiming(unit)
if dep_id is not None: # after the commit: tax freezes, the ledger posts
with uc.repository() as repo:
frozen = uc._freeze_tax(uc._full_view(*repo.invoices.get(dep_id), repo))
view["tax_frozen"] = frozen.get("tax_frozen")
try:
with uc.repository() as repo:
view["posted"] = bool(uc._post_document(repo, dep_id, who(caller))[0].get("posted"))
except (RuleBroken, Refused):
view["posted"] = False
return view
def _settled_view(repo, pr: PaymentRequest, dep_id: str, amount, received=None) -> dict:
received = repo.requests.received(pr.request_id) if received is None else received
inv = repo.invoices.get(dep_id)[0]
return {"request_id": pr.request_id, "deposit_invoice_id": dep_id, "invoice_no": inv.invoice_no, "amount": Decimal(str(amount)),
"remaining": pr.remaining(received), "status": "settled" if pr.remaining(received) <= 0 else "issued"}
# ═══ AR-38: the early-payment discount taken ════════════════════════════════════════════════════
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/settlement-discount", permission=SETTLE_DISCOUNT, idempotent=True)
def take_settlement_discount(caller: dict, invoice_id: Name, payment_id: Name, received_on: IsoDate, discount_amount: Money) -> dict:
"""payments reports a receipt within the discount window that is short by the discount. A credit note kind settlement_discount
is raised for it - the lines pinned to the invoice's frozen codes pro rata (tax TAX-47), approved by the rule (the terms
set the amount, approved_by "terms"), numbered CN-, dated today, posted - and allocated through payments. Refused when
received_on is after the window (gate within_discount_window) or the amount is not the terms' percentage of the gross
outstanding (gate discount_matches_terms). The same payment answers the same credit note. Under settlement_discount_vat
invoice_wording no credit note is raised: payments records the discount as an allocation and tax reads it."""
def unit():
with uc.repository() as repo, uc._decide():
prior = repo.discounts.get(payment_id)
if prior is not None:
if prior["invoice_id"] != invoice_id:
raise Refused(409, f"payment {payment_id} already took a discount on invoice {prior['invoice_id']}")
return _discount_view(repo.credits.get(prior["credit_id"])[0]), None
inv = sv.visible_invoice(repo, caller, invoice_id)[0]
today = uc._today(repo)
if inv.lifecycle != "issued":
raise _refuse("within_discount_window", f"invoice {invoice_id} is {inv.lifecycle}", "")
if not inv.settlement_discount_pct:
raise _refuse("within_discount_window", f"invoice {inv.invoice_no} carries no early-payment discount", "")
if repo.settings.in_force(today).settlement_discount_vat == "invoice_wording":
raise RuleBroken("settlement_discount_vat is invoice_wording: no credit note is raised; payments records the discount "
"as an allocation (source_kind settlement_discount) and tax reads it for the return")
on = uc._date(received_on, "received_on")
if not finance.window_open(on, inv.settlement_discount_until):
raise _refuse("within_discount_window", f"received {on}, after {inv.settlement_discount_until} when the discount ended",
"the full amount is due")
outstanding, why = uc._outstanding_of(invoice_id)
if outstanding is None:
raise Refused(502, f"payments could not say what is outstanding on {inv.invoice_no}: {why}")
expected = money(outstanding * Decimal(inv.settlement_discount_pct) / 100)
if discount_amount != expected:
raise _refuse("discount_matches_terms", f"{discount_amount} is not {inv.settlement_discount_pct.normalize():f}% of the "
f"{outstanding} outstanding ({expected})", "send the terms' percentage, exactly")
cn = _discount_credit(inv, payment_id, discount_amount)
tax_point = inv.tax_point_date or inv.issue_date
cn.apply_tax(_quote(cn, inv, tax_point))
if cn.total != discount_amount: # a penny of rounding: move it onto the last line, once
cn.lines[-1].unit_price += discount_amount - cn.total
cn.lines[-1].price()
cn.apply_tax(_quote(cn, inv, tax_point))
if cn.total != discount_amount:
raise _refuse("discount_matches_terms", f"the discount cannot be split exactly at the invoice's rates ({cn.total})", "")
s = repo.settings.in_force(today)
gates = cn.gates(remaining_creditable=inv.total - repo.credits.credited_total(invoice_id), second_approver=False,
approver="terms", invoice_lifecycle=inv.lifecycle)
number, sid = sv.claim_number(repo, s, "credit", cn.credit_id, today, work.now(), company=inv.company_code)
repo.credits.claim_slot(invoice_id, cn.credit_id, work.now())
cn.approve(today=today, number=number, series_id=sid, by="terms", at=work.now(), quote=None, gates=gates)
repo.credits.save(cn, 0)
repo.discounts.record(payment_id, invoice_id, cn.credit_id, work.now()) # plain INSERT keyed on the receipt
return _discount_view(cn), (cn.credit_id, cn.total)
body, made = uc._claiming(unit)
if made is not None: # after the commit: allocate through payments, post
credit_id, total = made
a = seams.allocate("credit_note", credit_id, invoice_id, total, work.now().date(), who(caller))
body["allocated"] = a.ok
if not a.ok:
body["allocation_pending"] = a.reason
try:
with uc.repository() as repo:
uc._post_document(repo, credit_id, who(caller))
except (RuleBroken, Refused):
pass
return body
def _quote(cn: CreditNote, inv: Invoice, tax_point: date):
out = seams.tax_compute(cn.lines, bp_id=cn.bp_id, tax_point=tax_point, currency=cn.currency, document_ref=cn.credit_id)
if out.quote is None:
raise Refused(502, f"the discount's tax cannot be computed: {out.reason}")
return out.quote
def _discount_credit(inv: Invoice, payment_id: str, discount: Decimal) -> CreditNote:
shares = finance.discount_shares([(r.tax_code, r.net_amount, r.tax_amount) for r in inv.tax], discount)
lines = []
for i, (code, net, _share) in enumerate(shares, start=1):
ln = Line(line_no=i, description=f"Early-payment discount on {inv.invoice_no}", quantity=Decimal(1), unit_price=net,
tax_code=code, pinned_from=inv.invoice_id)
ln.price()
lines.append(ln)
cn = CreditNote.draft(f"sd:{payment_id}", by="terms", bp_id=inv.bp_id, currency=inv.currency, lines=lines, invoice_id=inv.invoice_id,
reason=f"Early-payment discount {inv.settlement_discount_pct.normalize():f}% on {inv.invoice_no}")
cn.source_kind, cn.payment_id, cn.company_code = "settlement_discount", payment_id, inv.company_code
return cn
def _discount_view(cn: CreditNote) -> dict:
return {"credit_id": cn.credit_id, "credit_no": cn.credit_no, "net": cn.subtotal, "tax": cn.tax_total, "total": cn.total,
"approved_by": cn.approved_by}
# ═══ AR-39: statutory interest ══════════════════════════════════════════════════════════════════
def _interest_case(repo, inv: Invoice, today: date) -> dict:
gates = []
mode = repo.settings.in_force(today).late_interest_mode
gates.append(met("interest_mode_on") if mode != "off" else unmet("interest_mode_on", "late_interest_mode is off", "Settings (the owner's control)"))
treatment = seams.party_treatment(inv.bp_id)
business = bool(treatment) and "consumer" not in treatment
gates.append(met("business_customer") if business else unmet(
"business_customer", "statutory interest applies only between businesses" + ("" if treatment else " and tax holds no treatment for this customer"),
"none: a consumer is never charged"))
case = {"invoice_id": inv.invoice_id, "gates": [g.as_dict() for g in gates], "suggested": False}
if inv.lifecycle != "issued" or inv.kind == "interest" or not inv.due_date:
gates.append(unmet("late_invoice", f"invoice {inv.invoice_no or inv.invoice_id} is not an issued invoice with a due date", ""))
first = inv.due_date + timedelta(days=1) if inv.due_date else today
row = repo.rates.in_force(finance.reference_date(first))
gates.append(met("reference_rate") if row else unmet("reference_rate", f"no reference rate is in force on {finance.reference_date(first)}",
"add the Bank of England base rate for that half-year (POST /interest-rates)"))
allocs = seams.late_allocations([inv.invoice_id])
by, why = uc._settled([inv.invoice_id])
s = by.get(inv.invoice_id)
if allocs is None or s is None:
gates.append(unmet("payments_answer", "payments did not say what was paid and when" + (f" ({why})" if why else ""), "try again"))
case["gates"] = [g.as_dict() for g in gates]
if not all(g.met for g in gates):
return case
pieces = [(Decimal(str(a["amount"])), date.fromisoformat(a["received_on"])) for a in allocs
if a.get("source_kind") == "receipt" and not a.get("reverses_seq") and Decimal(str(a["amount"])) > 0]
if s.outstanding and s.outstanding > 0:
pieces.append((Decimal(str(s.outstanding)), today))
rate = finance.statutory_rate(row["reference_rate_pct"], row["margin_pct"])
raw = sum((amt * rate / 100 * max((d - inv.due_date).days, 0) / 365 for amt, d in pieces), ZERO)
days = max((max(d for _a, d in pieces) - inv.due_date).days, 0) if pieces else 0
interest = raw.quantize(Decimal("0.01"))
if interest <= 0:
case["gates"].append(unmet("late_invoice", "the invoice was not paid late", "").as_dict())
return case
case.update(suggested=True, interest=interest, compensation=finance.compensation(inv.total), days=days, rate_pct=rate,
reference_rate_pct=row["reference_rate_pct"], principal=inv.total, due_date=inv.due_date)
return case
@AR.use_case(kind="query", route="GET /invoices/{invoice_id}/interest-suggestion", permission=sv.READ)
def interest_suggestion(caller: dict, invoice_id: Name) -> dict:
"""What statutory interest and fixed compensation a late-paid (or still overdue) business invoice is owed: simple interest on
the money, daily, from the day after the due date to each receipt's received_on (still-open amounts to today), at the
reference rate of the due date's half-year plus 8%, rounded once; plus the fixed sum by size of debt. Nothing is suggested
for a consumer or with late_interest_mode off; the gates say why."""
with uc.repository() as repo:
inv = sv.visible_invoice(repo, caller, invoice_id)[0]
return _interest_case(repo, inv, uc._today(repo))
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/interest-draft", permission=sv.WRITE, idempotent=True)
def raise_interest_draft(caller: dict, invoice_id: Name) -> dict:
"""The biller accepts the suggested interest note: a DRAFT invoice kind interest (id "int:<invoice_id>", so once per invoice),
two lines coded OS (outside the scope of VAT) - the interest, naming the invoice, the days and the rate, and the fixed
compensation - reviewed and issued like any other."""
with uc.repository() as repo, uc._decide():
inv = sv.visible_invoice(repo, caller, invoice_id)[0]
draft_id = f"int:{invoice_id}"
if repo.invoices.get(draft_id) is not None:
return uc._full_view(*repo.invoices.get(draft_id), repo)
case = _interest_case(repo, inv, uc._today(repo))
if not case["suggested"]:
raise GatesUnmet("raise the interest note", [unmet(g["gate"], g["reason"], g["fix"]) for g in case["gates"] if not g["met"]])
label = f"invoice {inv.invoice_no}, {case['days']} days at {case['rate_pct'].normalize():f}%"
lines = [Line(line_no=1, description=f"Statutory interest on {label}", quantity=Decimal(1), unit_price=case["interest"],
unit_code="C62", tax_category="outside_scope", tax_code="OS", interest_on_invoice_id=invoice_id,
interest_days=case["days"], interest_rate_pct=case["rate_pct"]),
Line(line_no=2, description=f"Fixed compensation for the late payment of {inv.invoice_no}", quantity=Decimal(1),
unit_price=case["compensation"], unit_code="C62", tax_category="outside_scope", tax_code="OS")]
return uc._create_draft(repo, caller, invoice_id=draft_id, bp_id=inv.bp_id, currency=inv.currency, source_kind="manual",
source_id=draft_id, lines=lines, kind="interest", auto_issue=False, company_code=inv.company_code)
@AR.use_case(kind="command", route="POST /interest-rates", permission=sv.SETTINGS_WRITE)
def set_interest_rate(caller: dict, valid_from: IsoDate, reference_rate_pct: Pct, source: Optional[Name] = None) -> dict:
"""A reference rate from a date (30 June or 31 December; the Bank of England base rate then). Statutory margin is 8%. Rows are
facts about the past, so a date earlier than today is fine; a row is never edited."""
with uc.repository() as repo, uc._decide():
vf = uc._date(valid_from, "valid_from")
if (vf.month, vf.day) not in ((6, 30), (12, 31)):
raise RuleBroken("a reference rate starts on 30 June or 31 December")
try:
repo.rates.add(vf, reference_rate_pct, Decimal(8), source or "Bank of England base rate", who(caller))
except Exception as exc: # noqa: BLE001
if "uplicate" in str(exc) or "23505" in str(exc) or "UNIQUE" in str(exc):
raise Refused(409, f"there is already a reference rate from {vf}: rows are never edited") from None
raise
return {"valid_from": vf, "reference_rate_pct": reference_rate_pct, "margin_pct": Decimal(8)}
@AR.use_case(kind="query", route="GET /interest-rates", permission=sv.READ)
def list_interest_rates(caller: dict) -> dict:
"""The statutory reference rates, oldest first."""
with uc.repository() as repo:
return {"rates": repo.rates.all()}
# ═══ AR-40 / AR-41: the intercompany invoice ════════════════════════════════════════════════════
@AR.use_case(kind="query", route="GET /intercompany/open", permission=sv.READ)
def intercompany_open(caller: dict, company_code: Name, trading_partner: Name, as_at: Optional[IsoDate] = None) -> dict:
"""The receivable side of the ledger's intercompany reconciliation (GL-52): the issued invoices from company_code to
trading_partner with what is still outstanding, read from payments' settlement."""
with uc.repository() as repo:
day = uc._date(as_at, "as_at") or uc._today(repo)
ic_ids = repo.invoices.intercompany_ids(company_code, trading_partner)
loaded = repo.invoices.get_many(ic_ids)
invs = [loaded[i][0] for i in ic_ids]
invs = [i for i in invs if i.issue_date and i.issue_date <= day]
by, why = uc._settled([i.invoice_id for i in invs]) if invs else ({}, "")
docs = [{"invoice_id": i.invoice_id, "number": i.invoice_no, "currency": i.currency, "total": i.total,
"outstanding": by[i.invoice_id].outstanding if i.invoice_id in by else None} for i in invs]
credits = [repo.credits.get(c)[0] for i in invs for c in repo.credits.approved_for(i.invoice_id)]
credits = [c for c in credits if c.credit_date and c.credit_date <= day]
left = seams.credit_unallocated([(c.credit_id, c.bp_id) for c in credits])
docs += [{"credit_id": c.credit_id, "number": c.credit_no, "currency": c.currency, "total": -c.total, # a credit reduces what is owed
"outstanding": (-left[c.credit_id]) if c.credit_id in left else None} for c in credits]
return {"company_code": company_code, "trading_partner": trading_partner, "as_at": day, "documents": docs,
**({"settlement_unknown": why} if why else {})}
def _bill_body(inv: Invoice) -> dict:
return {"company_code": inv.trading_partner, "bill_id": f"ic:{inv.invoice_id}", "supplier_company_code": inv.company_code,
"supplier_invoice_no": inv.invoice_no, "issue_date": inv.issue_date.isoformat(),
"tax_point_date": (inv.tax_point_date or inv.issue_date).isoformat(), "due_date": inv.due_date.isoformat(), "currency": inv.currency,
"lines": [{"line_no": ln.line_no, "description": ln.description, "quantity": str(ln.quantity), "unit_code": ln.unit_code or "C62",
"unit_price": str(ln.unit_price), "net_amount": str(ln.net_amount), "tax_code": ln.tax_code,
"tax_amount": str(ln.tax_amount)} for ln in inv.lines],
"totals": {"net": str(inv.subtotal), "vat": str(inv.tax_total), "gross": str(inv.total)}}
@AR.use_case(kind="command", route="POST /invoices/{invoice_id}/hand-over", permission=HAND_OVER, idempotent=True)
def hand_over_intercompany(caller: dict, invoice_id: Name) -> dict:
"""Hand an issued intercompany invoice to the partner company's purchasing as a bill keyed "ic:<invoice_id>" (the same lines,
amounts and tax). Once: purchasing answers the same bill for the same key, and this brick records the hand-over. Where the
partner company is elsewhere, nothing is handed over and the invoice is sent like any other."""
with uc.repository() as repo, uc._decide():
inv = sv.visible_invoice(repo, caller, invoice_id)[0]
if inv.lifecycle != "issued" or not inv.trading_partner:
raise RuleBroken(f"invoice {invoice_id} is not an issued intercompany invoice")
done = repo.handovers.get(invoice_id)
if done is not None:
return {"invoice_id": invoice_id, "handed": True, "bill_id": done["bill_id"], "already": True}
known = seams.company_known(inv.trading_partner)
if known is False or (known is None and not seams.PURCHASING.installed):
return {"invoice_id": invoice_id, "handed": False,
"reason": f"company {inv.trading_partner} is not in this installation: send the invoice like any other"}
res = seams.hand_bill(_bill_body(inv), who(caller))
if not res.ok:
return {"invoice_id": invoice_id, "handed": False, "reason": res.reason}
repo.handovers.record(invoice_id, f"ic:{invoice_id}", inv.trading_partner, work.now()) # plain INSERT keyed on the invoice
return {"invoice_id": invoice_id, "handed": True, "bill_id": f"ic:{invoice_id}"}
def _credit_bill_body(cn: CreditNote, inv: Invoice) -> dict:
"""A credit note against an intercompany invoice goes over the same route as the invoice, as a bill keyed "ic:<credit_id>" with the
amounts negative (quantity too), so the partner's payable falls by it."""
return {"company_code": inv.trading_partner, "bill_id": f"ic:{cn.credit_id}", "supplier_company_code": inv.company_code,
"supplier_invoice_no": cn.credit_no, "issue_date": cn.credit_date.isoformat(), "tax_point_date": cn.credit_date.isoformat(),
"due_date": (inv.due_date or cn.credit_date).isoformat(), "currency": cn.currency,
"lines": [{"line_no": ln.line_no, "description": ln.description, "quantity": str(-(ln.quantity or Decimal(1))),
"unit_code": ln.unit_code or "C62", "unit_price": str(ln.unit_price), "net_amount": str(-ln.net_amount),
"tax_code": ln.tax_code, "tax_amount": str(-ln.tax_amount)} for ln in cn.lines],
"totals": {"net": str(-cn.subtotal), "vat": str(-cn.tax_total), "gross": str(-cn.total)}}
@AR.use_case(kind="command", route="POST /credit-notes/{credit_id}/hand-over", permission=HAND_OVER, idempotent=True)
def hand_over_credit_note(caller: dict, credit_id: Name) -> dict:
"""AR-41: an approved credit note against an intercompany invoice is handed to the partner company's purchasing the same way as the
invoice (once, keyed "ic:<credit_id>")."""
with uc.repository() as repo, uc._decide():
cn = sv.visible_credit(repo, caller, credit_id)[0]
inv = repo.invoices.get(cn.invoice_id)[0] if cn.invoice_id and repo.invoices.get(cn.invoice_id) else None
if cn.status != "approved" or inv is None or not inv.trading_partner:
raise RuleBroken(f"credit note {credit_id} is not an approved credit against an intercompany invoice")
done = repo.handovers.get(credit_id)
if done is not None:
return {"credit_id": credit_id, "handed": True, "bill_id": done["bill_id"], "already": True}
known = seams.company_known(inv.trading_partner)
if known is False or (known is None and not seams.PURCHASING.installed):
return {"credit_id": credit_id, "handed": False, "reason": f"company {inv.trading_partner} is not in this installation"}
res = seams.hand_bill(_credit_bill_body(cn, inv), who(caller))
if not res.ok:
return {"credit_id": credit_id, "handed": False, "reason": res.reason}
repo.handovers.record(credit_id, f"ic:{credit_id}", inv.trading_partner, work.now())
return {"credit_id": credit_id, "handed": True, "bill_id": f"ic:{credit_id}"}
def hand_over_after_issue(body: dict, caller: Optional[dict] = None) -> dict:
"""Once an intercompany invoice has issued and committed, it is handed to the partner company (best effort: a failure leaves the
invoice issued, says so, and a repeat of the hand-over route finishes it)."""
system = {"sub": "ar", "email": "ar@system", "permissions": [HAND_OVER, sv.READ]}
try:
got = hand_over_intercompany(system, body["invoice_id"])
except (RuleBroken, Refused) as exc:
return {**body, "handed_to_partner": False, "hand_over_reason": str(exc)}
return {**body, "handed_to_partner": got["handed"], **({} if got["handed"] else {"hand_over_reason": got.get("reason", "")})}