dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/application/services.py
"""What the use cases share: who sees which invoice, the views, the issue gates, posting to the ledger, the export file.
Nothing here is a route or a tool. Every function takes the open repository, so a use case stays one unit of work.
"""
import copy
import csv
import dataclasses
import io
import json
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from erp_platform import auth
from erp_platform.bricks import Refused
from brick_ar.domain import customer as customer_mod
from brick_ar.domain import deferral, einvoice, numbering
from brick_ar.domain.credit_note import CreditNote
from brick_ar.domain.dates import due_date
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.invoice import Invoice, Line
from brick_ar.domain.money import ZERO, line_net, money
from brick_ar.domain.readiness import Gate, met, unmet
from brick_ar.domain.values import InvoiceState, Settings
from brick_ar.infrastructure import seams
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
READ, WRITE, ISSUE, SEND, COLLECT, VOID = ("ar:invoice:read", "ar:invoice:write", "ar:invoice:issue", "ar:invoice:send",
"ar:invoice:collect", "ar:invoice:void")
CREDIT_WRITE, CREDIT_APPROVE = "ar:credit_note:write", "ar:credit_note:approve"
WRITEOFF_WRITE, WRITEOFF_APPROVE = "ar:writeoff:write", "ar:writeoff:approve"
SCHEDULE_READ, EXPORT, STATUS = "ar:schedule:read", "ar:export_batch:run", "ar:status:read"
CUSTOMER_WRITE, SETTINGS_WRITE, CONTROLS = "ar:customer:write", "ar:settings:write", "ar:settings:controls"
MIGRATION_READ, MIGRATION_SIGN = "ar:migration:read", "ar:migration:sign"
RAISE, CHECK, POST = "ar:invoice:raise", "ar:invoice:check", "ar:invoice:post" # machine
PLAN, STOP, FINISH = "ar:schedule:plan", "ar:schedule:stop", "ar:schedule:finish" # machine
CLAIM, LOAD, RECONCILE = "ar:number_series:claim", "ar:migration:load", "ar:migration:reconcile" # machine
SETTLE_DISCOUNT, SETTLE_REQUEST, HAND_OVER = "ar:invoice:settle_discount", "ar:payment_request:settle", "ar:intercompany:hand_over" # machine
RELEASE, EINVOICE = "ar:revenue:release", "ar:einvoice:deliver" # machine
SCHEDULE_WRITE, PROVISION_WRITE, PROVISION_APPROVE = "ar:schedule:write", "ar:provision:write", "ar:provision:approve"
MACHINE = (RAISE, CHECK, POST, PLAN, STOP, FINISH, CLAIM, LOAD, RECONCILE, SETTLE_DISCOUNT, SETTLE_REQUEST, HAND_OVER, RELEASE, EINVOICE)
# Who sees a DRAFT or a cancelled draft: the people and machines that author or approve invoices. The collector (and any
# other reader) sees what has been issued and what became of it.
_AUTHORS = (WRITE, ISSUE, RAISE, CHECK, VOID, CREDIT_APPROVE, WRITEOFF_APPROVE, CONTROLS, SETTINGS_WRITE)
def who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def sub_of(caller: dict) -> str:
return (caller.get("sub") or "").strip().lower()
def ids(caller: dict) -> tuple:
"""Every identity the caller has (email and sub, lower-case): the same person under either is the same person."""
return tuple(dict.fromkeys(x.strip().lower() for x in (caller.get("email"), caller.get("sub")) if x and x.strip()))
def has(caller: dict, permission: str) -> bool:
return auth.has_permission(caller, permission)
def sees_drafts(caller: dict) -> bool:
return any(has(caller, p) for p in _AUTHORS)
def visible_invoice(repo, caller: dict, invoice_id: str) -> tuple:
"""(Invoice, version), or 404: a draft the caller may not see does not exist for them (a 403 would confirm it)."""
found = repo.invoices.get(invoice_id)
if found is None or (found[0].lifecycle in ("draft", "cancelled") and not sees_drafts(caller)):
raise Refused(404, f"no invoice {invoice_id}")
return found
def visible_credit(repo, caller: dict, credit_id: str) -> tuple:
found = repo.credits.get(credit_id)
if found is None or (found[0].status in ("draft", "cancelled") and not sees_drafts(caller)):
raise Refused(404, f"no credit note {credit_id}")
return found
# ── views ──────────────────────────────────────────────────────────────────────────────────────
def _plain(obj) -> dict:
return {f.name: getattr(obj, f.name) for f in dataclasses.fields(obj) if f.name != "events"}
def settlement_view(s: Optional[seams.Settlement], reason: str = "") -> dict:
if s is None:
return {"known": False, "reason": reason or "payments has no answer for this invoice"}
return {"known": True, "outstanding": s.outstanding, "allocated": s.allocated, "state": s.state, "age_days": s.age_days}
def invoice_view(inv: Invoice, version: int, *, settlement: Optional[dict] = None) -> dict:
out = {f.name: getattr(inv, f.name) for f in dataclasses.fields(inv)
if f.name not in ("events", "lines", "tax", "sends", "findings")}
out["lines"] = [_plain(ln) for ln in inv.lines]
out["tax"] = [_plain(t) for t in inv.tax]
out["sends"] = [_plain(s) for s in inv.sends]
out["findings"] = [_plain(f) for f in inv.findings]
out["version"] = version
out["settlement_discount_amount"] = inv.settlement_discount_amount
out["tax_pending"] = bool(inv.lines) and not inv.tax and inv.is_draft
mail = [s for s in inv.sends if s.message_id]
if mail: # AR-44: the access point's word, and a rejection with its reason
seen = json.loads(mail[-1].statuses_json or "[]")
out["einvoice"] = {"message_id": mail[-1].message_id, "statuses": seen,
"rejection": (next((x["reason"] for x in reversed(seen) if x["status"] == "RE"), None)
if _einv_current(seen) == "RE" else None)}
if settlement is not None:
out["settlement"] = settlement
return out
def invoice_row(h: dict, s: Optional[seams.Settlement], reason: str) -> dict:
row = {k: h[k] for k in ("invoice_id", "invoice_no", "bp_id", "lifecycle", "kind", "source_kind", "currency", "issue_date",
"due_date", "subtotal", "tax_total", "total", "export_batch_id")}
if h["lifecycle"] == "issued":
row["outstanding"] = s.outstanding if s else None
row["settlement_state"] = s.state if s else None
if s is None:
row["settlement_unknown"] = reason or "payments has no answer for this invoice"
return row
def credit_view(cn: CreditNote, version: int) -> dict:
out = {f.name: getattr(cn, f.name) for f in dataclasses.fields(cn) if f.name not in ("events", "lines", "tax")}
out["lines"] = [_plain(ln) for ln in cn.lines]
out["tax"] = [_plain(t) for t in cn.tax]
out["version"] = version
return out
# ── lines ──────────────────────────────────────────────────────────────────────────────────────
def build_lines(inputs: list, *, start_no: int = 1) -> list:
"""Line objects from inputs. A line named after a product takes its description, unit_code and tax category from the product
(copied at line time, never asked again); what the caller sent wins. Net amounts are computed, never typed."""
out, products = [], {}
for i, li in enumerate(inputs, start=start_no):
d = li if isinstance(li, dict) else li.model_dump(exclude_none=True)
pid = d.get("product_id") or ""
if pid and pid not in products:
products[pid] = seams.get_product(pid) or {}
p = products.get(pid, {})
if (d.get("so_line_no") is None) != (not d.get("sales_order_id")):
raise RuleBroken(f"line {i}: an order line is named by sales_order_id AND so_line_no, together")
usage = bool(d.get("usage"))
if d.get("quantity") is None and not usage:
raise RuleBroken(f"line {i} needs a quantity; leave it out only on a usage line, which the biller fills in each period")
if d.get("unit_price") is None:
raise RuleBroken(f"line {i} needs a unit price")
desc = d.get("description") or p.get("name") or ""
if not desc:
raise RuleBroken(f"line {i} needs a description (or a product that has one)")
pu = d.get("pricing_unit_code") or ""
if pu and d.get("quantity_priced") is None and not usage:
raise Refused(422, f"line {i}: quantity_priced is required with pricing_unit_code (the quantity in {pu})", gate="quantity_priced",
fix="send the line's quantity in the pricing unit")
given = d.get("net_amount")
out.append(Line(
line_no=i, description=desc, quantity=None if usage else d["quantity"], unit_price=d["unit_price"],
discount_pct=d.get("discount_pct", Decimal(0)), unit_code=d.get("unit_code") or p.get("sales_unit_code") or p.get("base_unit_code") or "",
product_id=pid, tax_category=d.get("tax_category") or p.get("tax_category") or "", tax_code=d.get("tax_code") or "",
sales_order_id=d.get("sales_order_id") or "", so_line_no=d.get("so_line_no"),
service_from=_day(d.get("service_from")), service_to=_day(d.get("service_to")), item_id=d.get("item_id") or "", cost_centre=d.get("cost_centre") or "", usage=usage,
pricing_quantity=d.get("pricing_quantity"), pricing_unit_code=pu, quantity_priced=d.get("quantity_priced"),
net_amount=given if (given is not None and not usage) else ZERO, net_given=given is not None and not usage))
return out
def _day(v) -> Optional[date]:
return None if not v else (v if isinstance(v, date) else date.fromisoformat(v))
def check_order_lines(lines: list) -> None:
"""Where sales-order is installed, a line that names an order line names one that exists (and takes its unit)."""
cache: dict = {}
for ln in lines:
if not ln.sales_order_id:
continue
if ln.sales_order_id not in cache:
cache[ln.sales_order_id] = seams.order_lines(ln.sales_order_id)
known = cache[ln.sales_order_id]
if known is None:
continue
row = known.get(ln.so_line_no)
if row is None:
raise RuleBroken(f"line {ln.line_no}: order {ln.sales_order_id} has no line {ln.so_line_no}")
ln.unit_code = ln.unit_code or row.get("unit_code", "")
ln.product_id = ln.product_id or row.get("product_id", "")
# ── tax and the issue gates ────────────────────────────────────────────────────────────────────
def draft_tax(inv: Invoice, today: date) -> None:
"""Compute the draft's tax at its tax point (today when the source has none yet). Tax unavailable leaves it pending: the
draft is a draft, and the readiness answer says why it cannot issue."""
if not inv.lines:
return
outcome = seams.tax_compute(inv.lines, bp_id=inv.bp_id, tax_point=inv.tax_point_date or today, currency=inv.currency,
document_ref=inv.invoice_id, fx_rate=inv.fx_rate)
if outcome.quote is not None:
inv.apply_tax(outcome.quote)
def profile_for(repo, bp_id: int, on: date) -> Optional[customer_mod.Profile]:
return customer_mod.in_force(repo.customers.rows(bp_id), on)
def address_profile(repo, inv: Invoice, payer_profile, on: date):
"""v6: the invoice is addressed from the bill-to partner's billing profile (default: the payer's own); bp_id stays the payer,
whose terms, default cost centre and receivable are used."""
if inv.bill_to_bp_id and inv.bill_to_bp_id != inv.bp_id:
return profile_for(repo, inv.bill_to_bp_id, on)
return payer_profile
def fx_gate(inv: Invoice, functional: str) -> Gate:
"""A FOREIGN invoice - in a currency other than the functional currency the books are kept in (the owner's dated control
functional_currency, default GBP) - carries the rate to its functional total."""
home = (functional or "GBP").upper()
if inv.currency != home and not inv.fx_rate:
return unmet("fx_rate", f"the invoice is in {inv.currency} and the books are kept in {home}: its functional-currency "
"total needs an exchange rate", "enter the exchange rate on the draft (fx_rate, up to 4 decimals)")
return met("fx_rate")
def supplier(settings) -> dict:
"""Our name and address as the invoice shows them (ar's settings supplier_name / supplier_address)."""
return {"name": settings.supplier_name or "", "address": settings.supplier_address or ""}
def issue_gates(repo, inv: Invoice, today: date) -> tuple:
"""Every gate of `issue` as {gate, met, reason, fix}, with the tax quote and profile the issue would use."""
gates = list(inv.local_gates())
profile = profile_for(repo, inv.bp_id, today)
address = address_profile(repo, inv, profile, today)
gates.append(customer_mod.profile_gate(address, today))
settings = repo.settings.in_force(today)
gates.append(fx_gate(inv, settings.functional_currency))
gates.append(met("number_series") if "{seq" in settings.invoice_number_format else
unmet("number_series", "no series for this document type and year", "Settings: number format"))
if repo.invoices.migration_in_progress(inv.bp_id):
gates.append(unmet("migration_signed", "this customer's Odoo history is loaded but its reconciliation is not signed",
"Migration: the owner signs the reconciliation"))
tax_point = inv.tax_point_date or today
quote, tax_gates = None, []
if inv.lines and all(g.met for g in inv.local_gates()):
outcome = seams.tax_compute(inv.lines, bp_id=inv.bp_id, tax_point=tax_point, currency=inv.currency,
document_ref=inv.invoice_id, fx_rate=inv.fx_rate)
quote = outcome.quote
if quote is None:
tax_gates.append(unmet("tax_content", outcome.reason, "install or restore the tax module, or fix the line tax names"))
else:
for raw in seams.tax_check_contents(document_ref=inv.invoice_id, bill_to_json=address.bill_to_json() if address else "",
issue_date=today, supplier=supplier(settings)):
tax_gates.append(Gate(raw["gate"], bool(raw["met"]), raw.get("reason", ""), raw.get("fix", "")))
if all(g.met for g in tax_gates):
tax_gates = [met("tax_content")]
else:
tax_gates.append(unmet("tax_content", "the tax cannot be checked until the invoice has lines (and usage filled in)",
"add the lines, then ask again"))
gates += tax_gates
tp = seams.tax_point_status(tax_point)
closed = seams.period_open(tax_point) is False if seams.gl_installed() else False
if tp is not None and not tp.known: # tax is installed and did not answer: nobody can vouch for the period
gates.append(unmet("tax_point_open", tp.reason, tp.fix))
elif closed or (tp is not None and (tp.filed or not tp.gate_met)):
gates.append(unmet("tax_point_open", (tp.reason if tp is not None and tp.reason else
"the tax point falls in a filed VAT return or a closed ledger period"),
(tp.fix if tp is not None and tp.fix else "issue with a tax point of today, or ask the accountant to reopen")))
else:
gates.append(met("tax_point_open"))
gates += deposit_gates(repo, inv)
gates.append(cost_centre_gate(inv, profile, settings))
if profile is not None and profile.einvoice_channel == "peppol": # validated BEFORE a number is claimed (AR-43)
gates.append(einvoice_gate(inv, profile, settings, quote, today))
return gates, quote, profile
def effective_cost_centre(ln: Line, profile) -> str:
return ln.cost_centre or (profile.default_cost_centre if profile is not None else "")
def cost_centre_gate(inv: Invoice, profile, settings) -> Gate:
"""AR-47: where the ledger requires a cost centre on revenue, every revenue line has one (its own, else the customer's default)."""
if not settings.cost_centre_required:
return met("cost_centre")
blank = [str(ln.line_no) for ln in inv.lines if not ln.deducts_invoice_id and not effective_cost_centre(ln, profile)]
return met("cost_centre") if not blank else unmet(
"cost_centre", f"the ledger requires a cost centre on revenue and line {', '.join(blank)} has none", "name a cost centre on the line, or set the customer's default cost centre")
def supplier_party(settings) -> dict:
return {"name": settings.supplier_name or "", "address": settings.supplier_address or "", "country": settings.supplier_country or "",
"endpoint": settings.einvoice_sender_id or ""}
def einvoice_model(doc, *, kind: str, number: str, issue_date, due_date, settings, bill_to: dict, participant: str) -> dict:
return einvoice.build(doc, kind=kind, number=number, issue_date=issue_date, due_date=due_date, supplier=supplier_party(settings),
bill_to=bill_to, buyer_endpoint=participant)
def einvoice_gate(inv: Invoice, profile, settings, quote, today: date) -> Gate:
"""The draft as it WOULD be rendered must pass EN 16931 / Peppol BIS 3 before a number is claimed. The number is the one rule not
yet checkable (the series claims it next), so a provisional one stands in."""
probe = copy.deepcopy(inv)
try:
if quote is not None:
probe.apply_tax(quote)
except RuleBroken as exc:
return unmet("einvoice_valid", str(exc), "recompute the tax")
m = einvoice_model(probe, kind="invoice", number="(next number)", issue_date=today, due_date=today, settings=settings,
bill_to=json.loads(profile.bill_to_json()), participant=profile.peppol_participant_id)
found = einvoice.validate(m)
return met("einvoice_valid") if not found else Gate(**found[0].as_gate())
def _einv_current(seen: list) -> Optional[str]:
rank = {"delivered": 0, "AB": 1, "IP": 2, "UQ": 3, "CA": 4, "RE": 5, "AP": 5, "PD": 6}
return max((x["status"] for x in seen), key=lambda st: rank.get(st, -1), default=None)
def deducted_net(inv: Invoice) -> dict:
"""The net a document deducts from each deposit: {deposit invoice id: positive net}."""
out: dict = {}
for ln in inv.lines:
if ln.deducts_invoice_id:
out[ln.deducts_invoice_id] = out.get(ln.deducts_invoice_id, ZERO) - ln.net_amount
return out
def deposit_remaining(repo, deposit: Invoice) -> Decimal:
"""What remains of a deposit's net: its net less the deductions on ISSUED finals (the claims) and less credits against it.
Re-derived on every ask, never stored."""
return deposit.subtotal - repo.deposits.claimed(deposit.invoice_id) - repo.deposits.credited_net(deposit.invoice_id)
def creditable_left(repo, inv: Invoice, credited: Decimal) -> Decimal:
"""What a credit may still take from an invoice: total less credits; for a deposit also within its remaining (gross-scaled)."""
left = inv.total - credited
if inv.kind == "deposit" and inv.subtotal:
cap = (max(deposit_remaining(repo, inv), ZERO) * inv.total / inv.subtotal).quantize(Decimal("0.01"))
left = min(left, cap)
return max(left, ZERO)
def deposit_gates(repo, inv: Invoice) -> list:
"""v4 gate deposit_remaining: each deduction within what remains of its deposit."""
out = []
for dep_id, net in sorted(deducted_net(inv).items()):
found = repo.invoices.get(dep_id)
if found is None:
out.append(unmet("deposit_remaining", f"deposit invoice {dep_id} does not exist", "deduct an issued deposit"))
continue
left = deposit_remaining(repo, found[0])
out.append(met("deposit_remaining") if net <= left else
unmet("deposit_remaining", f"a deduction takes {net} but only {left} remains of deposit {found[0].invoice_no}",
"deduct at most what remains"))
return out
def terms_due(repo, terms_id: str, issue: date) -> date:
t = repo.terms.get(terms_id)
if t is None:
raise RuleBroken(f"the customer's payment terms {terms_id} do not exist: set them in Customer billing")
return due_date(issue, t["days"], t["basis"])
# ── void ───────────────────────────────────────────────────────────────────────────────────────
def void_gates(repo, inv: Invoice, today: date) -> list:
gates = []
credits = repo.credits.approved_for(inv.invoice_id)
writeoffs = [w for w in repo.writeoffs.for_invoice(inv.invoice_id) if w["status"] == "approved"]
allocated, why = bool(credits or writeoffs), ""
if seams.PAYMENTS.installed:
sr = seams.settlement([inv.invoice_id])
s = sr.by_invoice.get(inv.invoice_id)
if s is None:
why = sr.reason or "payments has no answer for this invoice"
elif s.allocated > 0:
allocated = True
gates.append(met("no_allocations") if not allocated and not why else
unmet("no_allocations", why or "a receipt, credit or write-off is allocated to it", "credit it instead (AR-04)"))
gates.append(met("no_released_revenue") if not (repo.releases.any_for(inv.invoice_id) or repo.releases.totals(inv.invoice_id)) else
unmet("no_released_revenue", "part of its deferred revenue is already released to revenue", "credit it instead (AR-04)"))
standing = repo.deposits.standing(inv.invoice_id) if inv.kind == "deposit" else []
gates.append(met("no_deposit_deducted") if not standing else
unmet("no_deposit_deducted", f"issued final {standing[0][0]} deducts it ({standing[0][1]})",
"void the final first, or credit the remainder"))
drafts = repo.credits.drafts_for(inv.invoice_id)
gates.append(met("no_draft_credit") if not drafts else
unmet("no_draft_credit", f"a draft credit note names it ({drafts[0]}): approving that after the void would credit a "
"void invoice", "approve or cancel the draft credit note first"))
policy = repo.settings.in_force(today).void_policy
reason = ""
if policy == "never":
reason = "the owner's void_policy does not allow voiding an issued invoice"
elif inv.sends:
reason = "the owner's void_policy does not allow voiding an invoice that was sent"
else:
tp = seams.tax_point_status(inv.tax_point_date or inv.issue_date or today)
if tp is not None and not tp.known:
reason = "tax did not say whether the VAT return is filed, and nobody can vouch for the period"
elif tp is not None and tp.filed:
reason = "the owner's void_policy does not allow voiding an invoice whose VAT return is filed"
gates.append(met("void_policy") if not reason else unmet("void_policy", reason, "credit it instead (AR-04)"))
return gates
# ── the ledger ─────────────────────────────────────────────────────────────────────────────────
def ledger_mode(repo, on: date) -> str:
"""post | export | off, in force on a day. Unset (the reference) = post when the ledger is installed, else export monthly."""
mode = repo.settings.in_force(on).ledger_mode
return mode or ("post" if seams.gl_installed() else "export")
def _account(repo, purpose: str, tax_code: str, on: date, needs: list) -> Optional[int]:
a = repo.accounts.account(purpose, tax_code, on)
if a is None and tax_code:
a = repo.accounts.account(purpose, "", on)
if a is None:
needs.append(f"{purpose}{' ' + tax_code if tax_code else ''}")
return a
def defers(repo, inv: Invoice) -> bool:
"""Whether this invoice's service-period lines were posted to deferred revenue: the owner's switch in force on its issue date."""
return bool(inv.issue_date) and bool(repo.settings.in_force(inv.issue_date).defer_service_revenue)
def invoice_journal(repo, inv: Invoice, event: str, on: date):
"""(journal body, []) or (None, the account-map rows that are missing). issue: receivable / revenue / VAT by code, and
a deposit already invoiced is cleared against the deposit account; a service period that runs past the month goes to deferred revenue
(by purpose: the ledger maps it, AR-46); revenue lines carry their cost centre and nothing else is split here (the ledger splits the
receivable and the VAT by them, AR-47). void: the same, reversed, dated `on`."""
if inv.trading_partner:
return ic_journal(inv, event, on), []
needs: list = []
rec = _account(repo, "receivable", "", on, needs)
deducted = sum((-ln.net_amount for ln in inv.lines if ln.deducts_invoice_id), ZERO)
rev = _account(repo, "deposit" if inv.kind == "deposit" else "revenue", "", on, needs)
dep = _account(repo, "deposit", "", on, needs) if deducted else None
tax_lines = [(r.tax_code, _account(repo, "tax", r.tax_code, on, needs), r.tax_amount) for r in inv.tax if r.tax_amount]
if needs:
return None, needs
rows = [(rec, inv.total, True)]
if deducted:
rows.append((dep, deducted, True))
groups: dict = {}
for ln in inv.lines:
if ln.deducts_invoice_id:
continue
deferred = deferral.is_deferred(ln, inv.issue_date, defers(repo, inv)) if inv.kind in ("regular", "final") else False
key = (deferred, ln.cost_centre or "")
groups[key] = groups.get(key, ZERO) + ln.net_amount
for (deferred, cc), amount in sorted(groups.items()):
extra = {**({"cost_centre": cc} if cc else {}), **({"purpose": "deferred_revenue"} if deferred else {})}
rows.append((None if deferred else rev, amount, False, extra))
rows += [(a, amt, False) for _c, a, amt in tax_lines]
return _journal(inv.invoice_id, event, on, inv.currency, f"{inv.invoice_no} {event}", rows, 1 if event == "issue" else -1), []
def ic_journal(inv: Invoice, event: str, on: date) -> dict:
"""An intercompany invoice posts by PURPOSE (gl maps the accounts): ic_receivable against ic_revenue and the tax, each line naming the
partner company, so the ledger's intercompany reconciliation (GL-52) matches it to the partner's payable. A void swaps the sides."""
partner = inv.trading_partner
rows = [{"purpose": "ic_receivable", "debit": inv.total, "bp_id": int(inv.bp_id), "partner_company": partner},
{"purpose": "ic_revenue", "credit": inv.subtotal, "partner_company": partner}]
rows += [{"purpose": "tax", "tax_code": r.tax_code, "credit": r.tax_amount, "partner_company": partner} for r in inv.tax if r.tax_amount]
flip = {"debit": "credit", "credit": "debit"}
lines = []
for r in rows:
side = next(k for k in ("debit", "credit") if k in r)
amount = str(abs(money(r[side])))
lines.append({**{k: v for k, v in r.items() if k != side}, (side if event == "issue" else flip[side]): amount, "memo": f"{inv.invoice_no} {event}"})
body = {"journal_id": str(seams.journal_id(inv.invoice_id, event)), "source_module": "ar", "source_document": inv.invoice_id, "event": event,
"journal_date": on.isoformat(), "currency": inv.currency, "memo": f"{inv.invoice_no} {event}", "lines": lines}
if inv.company_code:
body["company_code"] = inv.company_code
return body
def credit_journal(repo, cn: CreditNote, on: date):
"""A credit note reverses revenue; what the invoice still held in deferred revenue is taken back first (AR-46), and a line's cost
centre travels with its revenue (AR-47)."""
needs: list = []
rec = _account(repo, "receivable", "", on, needs)
rev = _account(repo, "revenue", "", on, needs)
tax_lines = [(_account(repo, "tax", r.tax_code, on, needs), r.tax_amount) for r in cn.tax if r.tax_amount]
if needs:
return None, needs
taken = sum((r["amount"] for r in repo.releases.of_credit(cn.credit_id)), ZERO)
rows = [(rec, cn.total, False)]
if taken:
rows.append((None, taken, True, {"purpose": "deferred_revenue"}))
groups: dict = {}
for ln in cn.lines:
groups[ln.cost_centre or ""] = groups.get(ln.cost_centre or "", ZERO) + ln.net_amount
if taken and groups: # the deferred part is not revenue: take it off the first centres
for cc in sorted(groups):
cut = min(groups[cc], taken)
groups[cc] -= cut
taken -= cut
for cc, amount in sorted(groups.items()):
if amount:
rows.append((rev, amount, True, {"cost_centre": cc} if cc else {}))
rows += [(a, amt, True) for a, amt in tax_lines]
return _journal(cn.credit_id, "approve", on, cn.currency, f"{cn.credit_no} credit note", rows, 1), []
def writeoff_journal(repo, invoice_id: str, writeoff_id: str, amount: Decimal, currency: str, on: date):
needs: list = []
bad = _account(repo, "bad_debt", "", on, needs)
rec = _account(repo, "receivable", "", on, needs)
if needs:
return None, needs
return _journal(writeoff_id, "approve", on, currency, f"write-off of {invoice_id}", [(bad, amount, True), (rec, amount, False)], 1), []
def _journal(document_id: str, event: str, on: date, currency: str, memo: str, rows: list, sign: int) -> dict:
"""rows are (account_id, amount, debit?) or (account_id, amount, debit?, extras) - extras are line keys such as purpose or
cost_centre, the account then left out; sign -1 swaps the sides (a reversal). Zero rows are left out."""
lines = []
for row in rows:
account, amount, debit = row[:3]
extras = row[3] if len(row) > 3 else {}
if not amount:
continue
side = debit if sign == 1 else not debit
line = {"debit" if side else "credit": str(abs(money(amount))), "memo": memo, **extras}
if account is not None:
line = {"account_id": str(account), **line}
lines.append(line)
return {"journal_id": str(seams.journal_id(document_id, event)), "source_module": "ar", "source_document": document_id,
"event": event, "journal_date": on.isoformat(), "currency": currency, "memo": memo, "lines": lines}
# ── export (the monthly file for the accountant) ───────────────────────────────────────────────
CSV_HEAD = ["kind", "number", "date", "bp_id", "currency", "subtotal", "tax_total", "total", "status", "document_id"]
def export_csv(invoices: list, credits: list, with_head: bool = True) -> str:
buf = io.StringIO()
w = csv.writer(buf, lineterminator="\n")
if with_head:
w.writerow(CSV_HEAD)
for h in invoices:
w.writerow(["invoice", h["invoice_no"], h["issue_date"], h["bp_id"], h["currency"], money(h["subtotal"]), money(h["tax_total"]),
money(h["total"]), h["lifecycle"], h["invoice_id"]])
for h in credits:
w.writerow(["credit_note", h["credit_no"], h["credit_date"], h["bp_id"], h["currency"], money(h["subtotal"]),
money(h["tax_total"]), money(h["total"]), h["status"], h["credit_id"]])
return buf.getvalue()
# ── numbering ──────────────────────────────────────────────────────────────────────────────────
def claim_number(repo, settings: Settings, doc_kind: str, document_id: str, on: date, at: datetime, company: str = "") -> tuple:
"""(number, series id) for the next invoice, credit note or payment request: counted in the series the format and the year name -
and, with several companies, in the ISSUING company's own series (the claim key carries the company, the number printed does not)."""
fmt = settings.invoice_number_format
if doc_kind == "credit":
fmt = numbering.credit_format(fmt)
elif doc_kind == "request":
fmt = numbering.request_format(fmt)
scope = f"{company}|" if company else ""
try:
numbering.check_pair(settings.invoice_number_format, settings.number_restarts_yearly)
except ValueError as exc:
raise RuleBroken(str(exc)) from None
sid = numbering.series_id(fmt, on.year, settings.number_restarts_yearly)
used = [numbering.match(fmt, n[len(scope):], settings.number_restarts_yearly)
for n in repo.numbers.numbers_like(scope + numbering.prefix(fmt, on.year))]
floor = max([seq for got in used if got and got[0] == sid for seq in [got[1]]], default=0)
seq = repo.numbers.claim(scope + sid, document_id, at, free=lambda n: not repo.numbers.taken(scope + numbering.render(fmt, on.year, n)),
floor=floor)
number = numbering.render(fmt, on.year, seq)
repo.numbers.register(scope + number, document_id, at) # plain INSERT keyed on the number: the second document fails 23505
return number, scope + sid
def issued_state(inv: Invoice) -> bool:
return inv.lifecycle == InvoiceState.ISSUED.value