dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/ai.py
"""The AI steps beside the process (GL-15): a check of a proposed manual journal, and an explanation of a reconciliation break.
Each only SUGGESTS. Findings are listed with their evidence; the approver accepts or dismisses each (recorded with who and when);
nothing in the journal changes, and approval never waits on a finding. The owner turns a step on (settings ai.<step>.mode =
suggest); off, it refuses. The baseline needs no model (infrastructure/ai_steps.py); the audit trail is gl_ai_finding."""
import json
import uuid
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId
from pydantic import Field
from brick_gl.application.brick import (
GL,
JOURNAL_APPROVE,
JOURNAL_CHECK,
RECONCILE_RUN,
can_see,
company_row,
decide,
real_now,
repository,
sub_of,
today,
who,
)
from brick_gl.application.inputs import Code, CompanyCode, Reason, _In
from brick_gl.application.posting import number_of
from brick_gl.domain.errors import RuleBroken
from brick_gl.infrastructure import ai_steps
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, utc
_find = lambda repo: _Keyed(repo.conn, t.gl_ai_finding, ("finding_id",)) # noqa: E731
def _view(r: dict) -> dict:
return {k: r[k] for k in ("finding_id", "journal_id", "recon_id", "model", "finding", "evidence", "decision", "decided_by", "decided_at")}
def run_check(repo, j: dict, td) -> list:
"""Write the findings on a proposed journal (the check is on): replaced when it runs again, decisions kept per finding."""
lines = repo.journals.lines(j["journal_id"])
history = []
for h in repo.journals.rows(t.gl_journal.c.company_code == j["company_code"], t.gl_journal.c.source_module == "gl",
t.gl_journal.c.event.in_(("manual", "adjustment")), t.gl_journal.c.status.in_(("posted", "reversed"))):
history.append({"journal": h, "lines": repo.journals.lines(h["journal_id"]), "number": number_of(h)})
found = ai_steps.baseline_check(j, lines, history, td)
dg = ai_steps.digest(j["journal_id"], [[ln["account_id"], str(ln["debit"]), str(ln["credit"])] for ln in lines], len(history))
existing = {r["finding"] + "|" + (r["evidence"] or ""): r for r in _find(repo).rows(t.gl_ai_finding.c.journal_id == j["journal_id"])}
out = []
for f in found:
key = f["finding"] + "|" + f["evidence"]
if key in existing:
out.append(existing[key])
continue
row = {"finding_id": uuid.uuid4().hex, "journal_id": j["journal_id"], "recon_id": None, "model": "baseline", "inputs_digest": dg,
"finding": f["finding"], "evidence": f["evidence"], "decision": "open", "decided_by": None, "decided_sub": None, "decided_at": None}
_find(repo).add(row)
out.append(row)
return out
@GL.use_case(kind="command", route="POST /journals/{journal_id}/ai-check-journal", permission=JOURNAL_CHECK, idempotent=True)
def ai_check_journal(caller: dict, journal_id: BigId) -> dict:
"""The system checks a PROPOSED manual journal before approval (suggest mode): a duplicate of an earlier journal, an account pair
never paired before, an amount ten times the usual. Findings are listed with their evidence for the approver; nothing in the
journal changes and approval never waits on one. Refused while the owner has the step off."""
with repository() as repo, decide():
j = repo.journals.get(int(journal_id))
if j is None or not can_see(repo, caller, j["company_code"]):
raise Refused(404, f"no journal {journal_id}")
if j["status"] != "proposed":
raise RuleBroken(f"journal {journal_id} is {j['status']}: only a proposed journal is checked", status=409)
td = today(repo)
if repo.settings.in_force(td, j["company_code"]).mode("ai_check_journal") == "off":
raise RuleBroken("the AI check of manual journals is off (the owner turns it on: ai.ai_check_journal.mode = suggest)", status=409)
return {"journal_id": j["journal_id"], "mode": "suggest", "findings": [_view(r) for r in run_check(repo, j, td)]}
class DecisionModel(_In):
finding_id: Annotated[str, Field(max_length=64)]
decision: Code
@GL.use_case(kind="command", route="POST /journals/{journal_id}/accept-journal-ai-findings", permission=JOURNAL_APPROVE)
def accept_journal_ai_findings(caller: dict, journal_id: BigId, decisions: Annotated[list[DecisionModel], Field(max_length=200)]) -> dict:
"""Accept or dismiss each AI finding of a journal: recorded with who and when. The journal is never changed by a decision, and
approval does not wait for one."""
with repository() as repo, decide():
j = repo.journals.get(int(journal_id))
if j is None or not can_see(repo, caller, j["company_code"]):
raise Refused(404, f"no journal {journal_id}")
for d in decisions:
if d.decision not in ("accepted", "dismissed"):
raise RuleBroken("a decision is accepted or dismissed", status=422)
r = _find(repo).get(d.finding_id)
if r is None or r["journal_id"] != j["journal_id"]:
raise Refused(404, f"no finding {d.finding_id} on journal {journal_id}")
_find(repo).set((d.finding_id,), decision=d.decision, decided_by=who(caller), decided_sub=sub_of(caller), decided_at=utc(real_now()))
return {"journal_id": j["journal_id"], "findings": [_view(r) for r in _find(repo).rows(t.gl_ai_finding.c.journal_id == j["journal_id"])]}
@GL.use_case(kind="command", route="POST /reconciliations/explain-break", permission=RECONCILE_RUN, idempotent=True)
def explain_reconciliation_break(caller: dict, recon_id: Annotated[str, Field(max_length=200)], document: Annotated[str, Field(max_length=200)],
company_code: Optional[CompanyCode] = None) -> dict:
"""An assistant task beside the process: for one break of a stored reconciliation, a likely cause with its evidence (a receipt not
allocated, posted twice, a credit note in the wrong period). Nothing is posted or accepted by it; the bookkeeper decides the fix.
Recorded in gl_ai_finding with the recon_id."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
r = _Keyed(repo.conn, t.gl_reconciliation, ("recon_id",)).get(recon_id)
if r is None or r["company_code"] != company["company_code"]:
raise Refused(404, f"no reconciliation {recon_id}")
if repo.settings.in_force(today(repo), company["company_code"]).mode("explain_reconciliation_break") == "off":
raise RuleBroken("the explanation step is off (ai.explain_reconciliation_break.mode)", status=409)
brk = next((b for b in json.loads(r["breaks_json"] or "[]") if b["document"] == document), None)
if brk is None:
raise Refused(404, f"no break {document} in {recon_id}")
journals = repo.journals.rows(t.gl_journal.c.company_code == company["company_code"], t.gl_journal.c.source_document == document.replace("unattributed:", ""))
exp = ai_steps.baseline_explain(brk, {"journals": journals})
row = {"finding_id": uuid.uuid4().hex, "journal_id": None, "recon_id": recon_id, "model": "baseline",
"inputs_digest": ai_steps.digest(recon_id, document, brk), "finding": exp["cause"], "evidence": exp["evidence"], "decision": "open",
"decided_by": None, "decided_sub": None, "decided_at": None}
_find(repo).add(row)
return {"recon_id": recon_id, "document": document, "cause": exp["cause"], "evidence": exp["evidence"], "finding_id": row["finding_id"],
"note": "a suggestion: nothing was posted or accepted"}
_ = (Reason,)