dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/intercompany.py
"""Intercompany: one entry that charges a cost between two of our companies and posts in both with the due-to and due-from lines
made for it (GL-51), and the matching of each pair of companies' balances before the close (GL-52).
A cross-company document is ONE proposal (one ic_document_id) that becomes one journal per company, keyed
derived_id(ic_document_id, company_code), each balanced in its own functional currency. It is approved as a whole - by a
second person, applying to the whole document - and posts every company's journal in one transaction, or none.
"""
import json
import uuid
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from brick_gl.application import ledgerops
from brick_gl.application.brick import (
GL,
GROUP_EXTRACT,
JOURNAL_WRITE,
PERIOD_READ,
RECONCILE_RUN,
RepoLookups,
can_see,
company_row,
decide,
leading_ledger,
real_now,
repository,
same_person,
sub_of,
today,
who,
)
from brick_gl.application.inputs import Code, CompanyCode, Currency, IsoDate, Lines, Memo, Reason
from brick_gl.application.posting import _line_rows, approval_needed, fx_for, to_lines
from brick_gl.domain import dates
from brick_gl.domain import journal as jr
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.journal import Draft, LineIn
from brick_gl.domain.money import ZERO
from brick_gl.domain.readiness import refusal
from brick_gl.infrastructure import seams
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import utc
def settle_pairs(nets: dict) -> list:
"""[(debtor company, creditor company, amount)]: a company whose lines net to a debit (it took the cost) owes the company whose
lines net to a credit (it paid). Greedy, largest first, so every company's offset is exactly its imbalance."""
debit = sorted(((c, n) for c, n in nets.items() if n > 0), key=lambda x: (-x[1], x[0]))
credit = sorted(((c, -n) for c, n in nets.items() if n < 0), key=lambda x: (-x[1], x[0]))
out, i, j = [], 0, 0
debit, credit = [list(x) for x in debit], [list(x) for x in credit]
while i < len(debit) and j < len(credit):
m = min(debit[i][1], credit[j][1])
if m > 0:
out.append((debit[i][0], credit[j][0], m))
debit[i][1] -= m
credit[j][1] -= m
if debit[i][1] == 0:
i += 1
if credit[j][1] == 0:
j += 1
return out
@GL.use_case(kind="command", route="POST /journals/propose-cross-company", permission=JOURNAL_WRITE)
def propose_cross_company(caller: dict, journal_date: IsoDate, lines: Lines, currency: Currency, memo: Optional[Memo] = None) -> dict:
"""Propose ONE entry that charges a cost between companies (UK01 pays a licence that DE01 uses): each line names its
company_code; the due-to and due-from lines (ic_receivable / ic_payable per partner company) are computed for you and shown
before approval, with partner_company on every intercompany line, each company's journal balanced in its own functional
currency (converted at its rate on the date). Needs ic_receivable and ic_payable mapped for every pair (gate
ic_accounts_mapped) and every company's period open (gate ic_periods_open). The second approver applies to the whole
document; approve_manual posts every company's journal in one transaction."""
on = dates.parse(journal_date)
actor, actor_sub = who(caller), sub_of(caller)
def unit():
with repository() as repo, decide():
td = today(repo)
if on > td:
raise RuleBroken(f"a manual journal is not dated after today ({td}): {on} is in the future", status=422)
default = repo.companies.default()
groups: dict = {}
for li in to_lines(lines):
code = li.company_code or (default["company_code"] if default else "")
groups.setdefault(code, []).append(li)
if len(groups) < 2:
raise RuleBroken("a cross-company journal has lines in two or more companies: use propose_manual for one", status=422)
companies = {}
for code in sorted(groups):
c = repo.companies.get(code)
if c is None or not can_see(repo, caller, code):
raise Refused(404, f"no company {code}")
companies[code] = c
nets = {code: sum((li.debit - li.credit for li in ls), ZERO) for code, ls in groups.items()}
if sum(nets.values(), ZERO) != 0:
raise refusal("balanced", f"debits and credits differ by {abs(sum(nets.values(), ZERO))} {currency} across the companies",
"the lines of all companies together must balance", status=422)
doc_id = derived_id("ic", uuid.uuid4().hex)
for debtor, creditor, m in settle_pairs(nets):
groups[debtor].append(LineIn(purpose="ic_payable", qualifier=creditor, partner_company=creditor, credit=m,
ic_reference=str(doc_id), memo=f"due to {creditor}"))
groups[creditor].append(LineIn(purpose="ic_receivable", qualifier=debtor, partner_company=debtor, debit=m,
ic_reference=str(doc_id), memo=f"due from {debtor}"))
built_all, shown = {}, []
for code, ls in sorted(groups.items()):
c = companies[code]
try:
p = ledgerops.period_for(repo, code, on)
except RuleBroken as e:
raise refusal("ic_periods_open", f"the period is closed in {code}: {e}", "date it in a period open in both",
status=409) from None
ctx = RepoLookups(repo, c)
try:
built = jr.build(Draft(code, on, currency, ls, None, None, memo or "", "gl", "", "intercompany"), ctx, manual=True,
fx=fx_for(repo, c, currency, None, on))
except RuleBroken as e:
if e.gate == "purposes_mapped" and ("ic_" in str(e)):
raise refusal("ic_accounts_mapped", f"{e}", "Account map: map ic_receivable and ic_payable per partner company",
status=422) from None
raise
built_all[code] = (c, p, built)
shown.append({"company_code": code, "functional_currency": c["functional_currency"], "fx_rate": built.fx_rate,
"functional_total": built.f_total,
"lines": [{"line_no": ln.line_no, "account_id": ln.account_id, "purpose": ln.purpose,
"debit": ln.debit, "credit": ln.credit, "functional_debit": ln.f_debit,
"functional_credit": ln.f_credit, "partner_company": ln.partner_company} for ln in built.lines]})
settings = repo.settings.in_force(td)
top = max(b.f_total for _, _, b in built_all.values())
at = real_now()
journals = {}
for code, (c, p, built) in built_all.items():
jid = derived_id(doc_id, code)
journals[code] = jid
repo.journals.add({**ledgerops.HEADER_DEFAULTS, "journal_id": jid, "company_code": code, "journal_date": on,
"period_id": p["period_id"], "fiscal_year": p["fiscal_year"], "book": "misc", "source_module": "gl",
"source_document": f"ic:{doc_id}", "event": "intercompany", "currency": currency,
"memo": memo or "", "ic_document_id": doc_id, "body_digest": "", "proposed_by": actor,
"proposed_sub": actor_sub, "status": "proposed", "total": built.total,
"functional_total": built.f_total, "fx_rate": built.fx_rate})
repo.journals.add_lines(_line_rows(jid, built))
repo.journals.audit(jid, "proposed", actor, actor_sub, at)
_ic = repo.conn.execute
_ic(t.gl_ic_document.insert().values(
ic_document_id=doc_id, companies=",".join(sorted(journals)), journal_ids=",".join(str(v) for k, v in sorted(journals.items())),
proposed_by=actor, proposed_sub=actor_sub, status="proposed", header_json=json.dumps({"date": str(on), "currency": currency}),
lines_json="[]", created_at=utc(at), updated_at=utc(at)))
return {"ic_document_id": doc_id, "status": "proposed", "needs_approval": True, "journals": shown,
"journal_ids": {k: str(v) for k, v in journals.items()}, "top_functional_total": top,
"approval_required_by_settings": approval_needed(settings, top)}
from brick_gl.application.brick import claims
return claims(unit)
# ── GL-52: matching a pair of companies ──────────────────────────────────────────────────────────────────────────────
def _pair_balances(repo, company_code: str, partner: str, as_at: date, ledger_id: str) -> dict:
"""{currency: transaction-currency net (debit +)} of the company's intercompany ('ic') accounts with the partner."""
s = t.gl_split_line.c
rows = repo.ledger.sums(company_code, ledger_id, upto=as_at, group_by=("currency",), extra=[
s.partner_company == partner, s.account_id.in_(
t.gl_account.select().with_only_columns(t.gl_account.c.account_id).where(t.gl_account.c.subtype == "ic"))])
return {r["currency"]: r["debit"] - r["credit"] for r in rows}
@GL.use_case(kind="command", route="POST /intercompany/match", permission=RECONCILE_RUN, idempotent=True)
def match_intercompany(caller: dict, company_code: CompanyCode, partner_company: CompanyCode, as_at: IsoDate) -> dict:
"""Match a company's intercompany balances and charges against its partner's as at a date: items are matched by
ic_document_id (always equal for what gl itself generated), then by ic_reference (an ar invoice of this company against the
partner's purchasing bill that carries its number: a difference is a break with its fix), then the remaining balance per
pair and transaction currency. Breaks at or below ic_tolerance are within tolerance. The result is keyed
'<company>:<partner>:<as_at>' and read by consolidation; running it again keeps what was accepted."""
day = dates.parse(as_at)
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
partner = company_row(repo, caller, partner_company)
code, pcode = company["company_code"], partner["company_code"]
settings = repo.settings.in_force(today(repo), code)
tol = Decimal(str(settings.ic_tolerance))
ledger = leading_ledger(repo)
matched, breaks = [], []
ours, theirs = _pair_balances(repo, code, pcode, day, ledger), _pair_balances(repo, pcode, code, day, ledger)
# 1. gl's own documents: both sides posted by one approval, always equal
docs = {j["ic_document_id"] for j in repo.journals.rows(t.gl_journal.c.company_code == code, t.gl_journal.c.ic_document_id.isnot(None),
t.gl_journal.c.status.in_(("posted", "reversed")),
t.gl_journal.c.journal_date <= day)}
for d in sorted(docs):
mine = [j for j in repo.journals.by_ic_document(d) if j["company_code"] == code]
matched.append({"key": f"ic:{d}", "amount": mine[0]["total"], "currency": mine[0]["currency"]})
# 2. modules' documents, by ic_reference
a_ar = seams.neighbours.intercompany_open("ar", code, pcode, day, who(caller))
a_pur = seams.neighbours.intercompany_open("purchasing", pcode, code, day, who(caller))
if a_ar.ok and a_pur.ok:
sell = {d.get("number") or d.get("invoice_id") or d.get("credit_id"): d for d in a_ar.data.get("documents", [])}
buy = {d.get("ic_reference"): d for d in a_pur.data.get("documents", [])}
for ref in sorted(set(sell) | set(buy), key=str):
mine, other = sell.get(ref), buy.get(ref)
if mine and other:
diff = Decimal(str(mine["total"])) - Decimal(str(other["total"]))
if diff == 0:
matched.append({"key": ref, "amount": Decimal(str(mine["total"])), "currency": mine["currency"]})
else:
breaks.append({"key": ref, "ours": Decimal(str(mine["total"])), "theirs": Decimal(str(other["total"])),
"currency": mine["currency"], "difference": diff, "within_tolerance": abs(diff) <= tol,
"accepted": False, "reason": "", "fix": "post the difference or correct the bill"})
else:
side = mine or other
amt = Decimal(str(side["total"]))
breaks.append({"key": str(ref), "ours": Decimal(str(mine["total"])) if mine else ZERO,
"theirs": Decimal(str(other["total"])) if other else ZERO, "currency": side["currency"],
"difference": amt if mine else -amt, "within_tolerance": amt <= tol, "accepted": False, "reason": "",
"fix": "post the missing side" if mine else "the partner's bill has no invoice here: correct its ic_reference"})
# 3. what is left: the balance per pair and transaction currency
for cur in sorted(set(ours) | set(theirs)):
a, b = ours.get(cur, ZERO), -theirs.get(cur, ZERO)
if a == b:
if a != 0:
matched.append({"key": f"balance:{cur}", "amount": a, "currency": cur})
continue
diff = a - b
breaks.append({"key": f"balance:{cur}", "ours": a, "theirs": b, "currency": cur, "difference": diff,
"within_tolerance": abs(diff) <= tol, "accepted": False, "reason": "",
"fix": "post the difference to the right company, or correct the entry"})
mid = f"{code}:{pcode}:{day}"
old = repo.conn.execute(t.gl_ic_match.select().where(t.gl_ic_match.c.match_id == mid)).first()
if old is not None:
kept = {b["key"]: b for b in json.loads(dict(old._mapping)["breaks_json"] or "[]") if b.get("accepted")}
for b in breaks:
k = kept.get(b["key"])
if k and Decimal(str(k["difference"])) == b["difference"]:
b["accepted"], b["reason"] = True, k.get("reason", "")
at = real_now()
row = {"match_id": mid, "company_code": code, "partner_company": pcode, "as_at": day,
"matched_json": json.dumps(matched, default=str), "breaks_json": json.dumps(breaks, default=str),
"within_tolerance": all(b["within_tolerance"] or b["accepted"] for b in breaks), "accepted_by": None,
"accepted_sub": None, "accepted_reason": None, "run_at": utc(at)}
if old is not None:
repo.conn.execute(t.gl_ic_match.update().where(t.gl_ic_match.c.match_id == mid).values(**row, updated_at=utc(at)))
else:
repo.conn.execute(t.gl_ic_match.insert().values(**row, created_at=utc(at), updated_at=utc(at)))
repo.audit_master("ic_match_run", mid, None, {"as_at": str(day)}, who(caller), sub_of(caller), at)
return {"match_id": mid, "as_at": day, "matched": matched, "breaks": breaks, "run_at": utc(at), "run_by": who(caller)}
def _match(repo, mid: str) -> Optional[dict]:
r = repo.conn.execute(t.gl_ic_match.select().where(t.gl_ic_match.c.match_id == mid)).first()
return dict(r._mapping) if r is not None else None
def _match_view(m: dict) -> dict:
return {"match_id": m["match_id"], "as_at": m["as_at"], "matched": json.loads(m["matched_json"] or "[]"),
"breaks": json.loads(m["breaks_json"] or "[]"), "run_at": m["run_at"]}
@GL.use_case(kind="command", route="POST /intercompany/accept-difference", permission=RECONCILE_RUN)
def accept_ic_difference(caller: dict, company_code: CompanyCode, partner_company: CompanyCode, as_at: IsoDate, key: Code,
reason: Reason) -> dict:
"""Accept an intercompany difference that is above tolerance, with a reason, recorded beside the break (consolidation
reads it). The accepter is not who ran the match (gate second_person): ask another accountant."""
day = dates.parse(as_at)
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
mid = f"{company['company_code']}:{partner_company}:{day}"
m = _match(repo, mid)
if m is None:
raise Refused(404, f"no match was run for {mid} (fix: run match_intercompany)")
breaks = json.loads(m["breaks_json"] or "[]")
target = next((b for b in breaks if b["key"] == key), None)
if target is None:
raise Refused(404, f"no break {key} in {mid}")
if target["within_tolerance"]:
raise RuleBroken(f"{key} is within tolerance: nothing to accept", status=409)
# the person who ran the match is recorded in the change log (ic_match_run); the run's author cannot accept
if _ran_by(repo, mid, caller):
raise refusal("second_person", "you ran this match, and the difference is accepted by a second person", "ask another accountant")
target["accepted"], target["reason"] = True, reason
target["accepted_by"] = who(caller)
done = all(b["within_tolerance"] or b["accepted"] for b in breaks)
repo.conn.execute(t.gl_ic_match.update().where(t.gl_ic_match.c.match_id == mid).values(
breaks_json=json.dumps(breaks, default=str), within_tolerance=done, accepted_by=who(caller), accepted_sub=sub_of(caller),
accepted_reason=reason, updated_at=utc(real_now())))
return _match_view(_match(repo, mid))
def _ran_by(repo, mid: str, caller: dict) -> bool:
from brick_gl.infrastructure.repositories import _Keyed
rows = _Keyed(repo.conn, t.gl_master_audit, ("audit_id",)).rows(
t.gl_master_audit.c.entity == "ic_match_run", t.gl_master_audit.c.natural_key == mid)
return any(same_person(caller, r["by_email"], r["by_sub"]) for r in rows)
@GL.use_case(kind="query", route="GET /intercompany/reconciliation", permission=GROUP_EXTRACT)
def get_ic_reconciliation(caller: dict, company_code: CompanyCode, partner_company: CompanyCode, as_at: IsoDate) -> dict:
"""Consolidation reads the result of match_intercompany for a pair as at a date: {match_id, as_at, matched, breaks (each with
ours, theirs, difference, within_tolerance, accepted, reason, fix), run_at}. 404 when no match was run (fix: run
match_intercompany) or the company is outside the client's scope."""
with repository() as repo:
company = company_row(repo, caller, company_code)
mid = f"{company['company_code']}:{partner_company}:{dates.parse(as_at)}"
m = _match(repo, mid)
if m is None:
raise Refused(404, f"[match] no intercompany match was run for {mid} (fix: run match_intercompany)")
return _match_view(m)
__all__ = ["PERIOD_READ"]