dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/audit.py
"""Audit: prove nothing in the ledger was changed after it was posted, and take the file the tax authority asks for (GL-59).
Every posted journal carries chain_seq, prev_hash and row_hash PER COMPANY (sha-256 over the previous link and the canonical
journal and lines); verification answers met, or the first broken link with its journal. The FEC (France, article A47 A-1 LPF,
18 columns, one row per line, entry numbers without gaps, the opening a-nouveaux lines from the carry-forward, alternative
account numbers where set, named <SIREN>FEC<yyyymmdd>.txt) and a SAF-T file (OECD SAF-T Financial 2.0 layout) are produced from
the books and stored as objects under raw/audit/<company>/<fy>/ with their sha-256 and who produced them. Re-exporting an
unchanged year gives the same sha-256. Master data and settings changes are in the change log with before and after.
"""
import hashlib
import json
import xml.etree.ElementTree as ET
from datetime import date
from decimal import Decimal
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from pydantic import Field
from sqlalchemy import select
from brick_gl.application import carry
from brick_gl.application.brick import (
AUDIT_EXPORT,
AUDIT_VERIFY,
GL,
JOURNAL_READ,
company_row,
decide,
leading_ledger,
real_now,
repository,
sub_of,
who,
)
from brick_gl.application.inputs import Code, CompanyCode, FiscalYearName
from brick_gl.domain import dates, hashchain
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.money import ZERO
from brick_gl.infrastructure import objects
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, utc
@GL.use_case(kind="query", route="GET /audit/verify-chain", permission=AUDIT_VERIFY)
def verify_audit_chain(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""Verify the hash chain of a company's posted journals: each link's row_hash is the sha-256 of the previous link and its
canonical journal and lines, so changing any stored column of a posted journal or its lines breaks verification at that journal.
Answers {met: true, journals} or {met: false, first_broken: {journal_id, chain_seq, why}}."""
with repository() as repo:
company = company_row(repo, caller, company_code)
chain = [dict(r._mapping) for r in repo.conn.execute(t.gl_journal.select().where(
t.gl_journal.c.company_code == company["company_code"], t.gl_journal.c.chain_seq.isnot(None)).order_by(t.gl_journal.c.chain_seq))]
broken = hashchain.verify(chain, lambda jid: repo.journals.lines(jid), lambda jid: repo.journals.splits(jid))
out = {"company_code": company["company_code"], "journals": len(chain), "met": broken is None}
if broken:
out["first_broken"] = broken
return out
def _fec_amount(v: Decimal) -> str:
return f"{v:.2f}".replace(".", ",")
def _day(d) -> str:
return d.strftime("%Y%m%d")
FEC_COLUMNS = ["JournalCode", "JournalLib", "EcritureNum", "EcritureDate", "CompteNum", "CompteLib", "CompAuxNum", "CompAuxLib", "PieceRef", "PieceDate",
"EcritureLib", "Debit", "Credit", "EcritureLet", "DateLet", "ValidDate", "Montantdevise", "Idevise"]
def _alt(repo, code: str, account: dict, on: date) -> tuple:
a = repo.alt_accounts.of(code, account["account_id"], on)
return (a["alt_code"], a["alt_name"] or account["name"]) if a else (account["code"], account["name"])
def build_fec(repo, company: dict, fy: str, ledger_id: str) -> tuple:
"""(rows, text): the FEC of a financial year in the leading ledger; the a-nouveaux lines first, then every journal line."""
code = company["company_code"]
year = dates.fy_number(fy)
start, end = carry.year_bounds(company, year)
rows, accounts = [], {}
def acct(aid):
if aid not in accounts:
accounts[aid] = repo.accounts.get(aid)
return accounts[aid]
opening = carry.opening_for_report(repo, company, ledger_id, year)
merged: dict = {}
for (aid, pc, seg, partner), (f, cur) in opening.items():
merged[aid] = merged.get(aid, ZERO) + f
for n, (aid, f) in enumerate(sorted(merged.items(), key=lambda kv: acct(kv[0])["code"]), 1):
if f == 0:
continue
num, name = _alt(repo, code, acct(aid), start)
rows.append(["AN", "A nouveaux", str(n), _day(start), num, name, "", "", "AN", _day(start), "A nouveau", _fec_amount(f if f > 0 else ZERO),
_fec_amount(-f if f < 0 else ZERO), "", "", _day(start), "", ""])
j, ln = t.gl_journal.c, t.gl_journal_line.c
q = (select(t.gl_journal, ln.line_no, ln.account_id, ln.debit, ln.credit, ln.functional_debit, ln.functional_credit, ln.bp_id,
ln.line_memo, ln.currency.label("line_currency"))
.select_from(t.gl_journal.join(t.gl_journal_line, ln.journal_id == j.journal_id))
.where(j.company_code == code, j.status.in_(("posted", "reversed")), j.journal_date >= start, j.journal_date <= end,
(j.ledger_id.is_(None)) | (j.ledger_id == ledger_id))
.order_by(j.book, j.entry_no, ln.line_no))
for r in repo.conn.execute(q):
m = dict(r._mapping)
a = acct(m["account_id"])
num, name = _alt(repo, code, a, m["journal_date"])
foreign = m["currency"] != company["functional_currency"]
fd, fc = m["functional_debit"] or ZERO, m["functional_credit"] or ZERO
rows.append([m["book"].upper(), m["book"].capitalize(), str(m["entry_no"]), _day(m["journal_date"]), num, name,
str(m["bp_id"]) if m["bp_id"] is not None and a["subtype"] in ("receivable", "payable") else "", "",
m["source_document"] or "", _day(m["journal_date"]), (m["line_memo"] or m["memo"] or "").replace("\t", " ").replace("\n", " "),
_fec_amount(fd), _fec_amount(fc), "", "", _day(m["posted_at"].date() if m["posted_at"] else m["journal_date"]),
(_fec_amount((m["debit"] or ZERO) - (m["credit"] or ZERO)) if foreign else ""), m["currency"] if foreign else ""])
text = "\t".join(FEC_COLUMNS) + "\n" + "".join("\t".join(r) + "\n" for r in rows)
return rows, text
def build_saft(repo, company: dict, fy: str, ledger_id: str) -> tuple:
code = company["company_code"]
year = dates.fy_number(fy)
start, end = carry.year_bounds(company, year)
root = ET.Element("AuditFile", {"xmlns": "urn:OECD:StandardAuditFile-Tax:2.00"})
h = ET.SubElement(root, "Header")
for k, v in (("AuditFileVersion", "2.00"), ("AuditFileCountry", company["country"] or ""), ("AuditFileDateCreated", _iso(end)),
("SoftwareCompanyName", "DODIL"), ("SoftwareID", "erp-brick-gl"), ("SoftwareVersion", "0.1.0")):
ET.SubElement(h, k).text = v
c = ET.SubElement(h, "Company")
ET.SubElement(c, "RegistrationNumber").text = code
ET.SubElement(c, "Name").text = company["name"] or code
sel = ET.SubElement(h, "SelectionCriteria")
ET.SubElement(sel, "PeriodStart").text = _iso(start)
ET.SubElement(sel, "PeriodEnd").text = _iso(end)
ET.SubElement(h, "DefaultCurrencyCode").text = company["functional_currency"]
ET.SubElement(h, "TaxAccountingBasis").text = "A"
mf = ET.SubElement(root, "MasterFiles")
gla = ET.SubElement(mf, "GeneralLedgerAccounts")
opening = carry.opening_for_report(repo, company, ledger_id, year)
op: dict = {}
for (aid, *_), (f, _cur) in opening.items():
op[aid] = op.get(aid, ZERO) + f
sums = {r["account_id"]: r for r in repo.ledger.sums(code, ledger_id, since=start, upto=end, group_by=("account_id",))}
for a in repo.accounts.of_chart(company["chart_id"]):
if a["account_id"] not in op and a["account_id"] not in sums:
continue
o = op.get(a["account_id"], ZERO)
s = sums.get(a["account_id"])
closing = o + ((s["f_debit"] - s["f_credit"]) if s else ZERO)
num, name = _alt(repo, code, a, end)
acc = ET.SubElement(gla, "Account")
ET.SubElement(acc, "AccountID").text = num
ET.SubElement(acc, "AccountDescription").text = name
ET.SubElement(acc, "AccountType").text = a["type"].capitalize()
ET.SubElement(acc, "OpeningDebitBalance").text = f"{o if o > 0 else ZERO:.2f}"
ET.SubElement(acc, "OpeningCreditBalance").text = f"{-o if o < 0 else ZERO:.2f}"
ET.SubElement(acc, "ClosingDebitBalance").text = f"{closing if closing > 0 else ZERO:.2f}"
ET.SubElement(acc, "ClosingCreditBalance").text = f"{-closing if closing < 0 else ZERO:.2f}"
gle = ET.SubElement(root, "GeneralLedgerEntries")
j = t.gl_journal.c
journals = [dict(r._mapping) for r in repo.conn.execute(t.gl_journal.select().where(
j.company_code == code, j.status.in_(("posted", "reversed")), j.journal_date >= start, j.journal_date <= end,
(j.ledger_id.is_(None)) | (j.ledger_id == ledger_id)).order_by(j.book, j.entry_no))]
td = tc = ZERO
for jr in journals:
for ln in repo.journals.lines(jr["journal_id"]):
td += ln["functional_debit"] or ZERO
tc += ln["functional_credit"] or ZERO
ET.SubElement(gle, "NumberOfEntries").text = str(len(journals))
ET.SubElement(gle, "TotalDebit").text = f"{td:.2f}"
ET.SubElement(gle, "TotalCredit").text = f"{tc:.2f}"
by_book: dict = {}
for jr in journals:
by_book.setdefault(jr["book"], []).append(jr)
for book, js in sorted(by_book.items()):
jn = ET.SubElement(gle, "Journal")
ET.SubElement(jn, "JournalID").text = book.upper()
ET.SubElement(jn, "Description").text = book.capitalize()
ET.SubElement(jn, "Type").text = book.upper()
for jr in js:
tr = ET.SubElement(jn, "Transaction")
ET.SubElement(tr, "TransactionID").text = f"{book.upper()}/{jr['fiscal_year']}/{jr['entry_no']}"
ET.SubElement(tr, "Period").text = jr["period_id"][5:]
ET.SubElement(tr, "TransactionDate").text = _iso(jr["journal_date"])
ET.SubElement(tr, "SourceID").text = jr["source_module"] or ""
ET.SubElement(tr, "Description").text = jr["memo"] or jr["source_document"] or ""
ET.SubElement(tr, "GLPostingDate").text = _iso(jr["posted_at"].date() if jr["posted_at"] else jr["journal_date"])
for ln in repo.journals.lines(jr["journal_id"]):
a = repo.accounts.get(ln["account_id"])
num, _name = _alt(repo, code, a, jr["journal_date"])
line = ET.SubElement(tr, "Line")
ET.SubElement(line, "RecordID").text = str(ln["line_no"])
ET.SubElement(line, "AccountID").text = num
ET.SubElement(line, "Description").text = ln["line_memo"] or ""
if (ln["functional_debit"] or ZERO) > 0:
ET.SubElement(line, "DebitAmount").text = f"{ln['functional_debit']:.2f}"
else:
ET.SubElement(line, "CreditAmount").text = f"{ln['functional_credit'] or ZERO:.2f}"
ET.indent(root)
return len(journals), ('<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(root, encoding="unicode") + "\n")
def _iso(d) -> str:
return d.isoformat()
def _export(repo, caller, company: dict, fy: str, kind: str, filename: str, body: str, rows: int, content_type: str) -> dict:
data = body.encode("utf-8")
sha = hashlib.sha256(data).hexdigest()
key = f"raw/audit/{company['company_code']}/{fy}/{sha[:12]}/{filename}"
objects.ports.store(repo).put(key, data, content_type)
at = real_now()
eid = f"{kind}:{company['company_code']}:{fy}:{at.isoformat()}"
row = {"export_id": eid, "kind": kind, "company_code": company["company_code"], "fiscal_year": fy, "object_key": key, "sha256": sha,
"row_count": rows, "produced_by": who(caller), "produced_sub": sub_of(caller), "produced_at": utc(at)}
_Keyed(repo.conn, t.gl_audit_export, ("export_id",)).add(row)
return {"export_id": eid, "kind": kind, "filename": filename, "object_key": key, "sha256": sha, "rows": rows, "produced_by": who(caller)}
@GL.use_case(kind="command", route="POST /audit/export-fec", permission=AUDIT_EXPORT, idempotent=True)
def export_fec(caller: dict, fiscal_year: FiscalYearName, siren: Annotated[str, Field(max_length=20)],
company_code: Optional[CompanyCode] = None) -> dict:
"""Produce the Fichier des Ecritures Comptables of a financial year (France, article A47 A-1 of the LPF): the 18 columns
(JournalCode ... Idevise), one row per line, entry numbers without gaps per journal code, the opening (a-nouveaux) lines from the
carry-forward, alternative account numbers where set, named <SIREN>FEC<yyyymmdd>.txt. Stored as an object under
raw/audit/<company>/<fy>/ with its sha-256 and who produced it; exporting an unchanged year again gives the same sha-256."""
with repository() as repo, decide():
if not (siren.isdigit() and len(siren) == 9):
raise RuleBroken("the SIREN is the company's 9 digits (it names the file)", status=422)
company = company_row(repo, caller, company_code)
y = repo.years.get(company["company_code"], fiscal_year)
if y is None:
raise Refused(404, f"no financial year {fiscal_year} of {company['company_code']}")
rows, text = build_fec(repo, company, fiscal_year, leading_ledger(repo))
name = f"{siren}FEC{y['ends_on'].strftime('%Y%m%d')}.txt"
out = _export(repo, caller, company, fiscal_year, "fec", name, text, len(rows), "text/tab-separated-values")
out["total_debit"] = sum((Decimal(r[11].replace(",", ".")) for r in rows), ZERO)
out["total_credit"] = sum((Decimal(r[12].replace(",", ".")) for r in rows), ZERO)
return out
@GL.use_case(kind="command", route="POST /audit/export-saft", permission=AUDIT_EXPORT, idempotent=True)
def export_saft(caller: dict, fiscal_year: FiscalYearName, company_code: Optional[CompanyCode] = None) -> dict:
"""Produce a SAF-T Financial file (OECD SAF-T 2.0 layout: Header, MasterFiles / GeneralLedgerAccounts, GeneralLedgerEntries with
journals, transactions and lines) of a financial year, stored as an object under raw/audit/<company>/<fy>/ with its sha-256
and who produced it."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
y = repo.years.get(company["company_code"], fiscal_year)
if y is None:
raise Refused(404, f"no financial year {fiscal_year} of {company['company_code']}")
n, xml = build_saft(repo, company, fiscal_year, leading_ledger(repo))
return _export(repo, caller, company, fiscal_year, "saft", f"SAFT_{company['company_code']}_{fiscal_year}.xml", xml, n, "application/xml")
@GL.use_case(kind="query", route="GET /audit/exports", permission=AUDIT_VERIFY)
def list_audit_exports(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""The statutory files produced, with their object key, sha-256 and who produced each."""
with repository() as repo:
company = company_row(repo, caller, company_code)
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_audit_export.select().where(
t.gl_audit_export.c.company_code == company["company_code"]).order_by(t.gl_audit_export.c.produced_at))]
return {"exports": [{k: r[k] for k in ("export_id", "kind", "fiscal_year", "object_key", "sha256", "row_count", "produced_by", "produced_at")} for r in rows]}
@GL.use_case(kind="query", route="GET /audit/exports/{export_id}/content", permission=AUDIT_EXPORT)
def get_audit_export(caller: dict, export_id: Annotated[str, Field(max_length=200)]) -> dict:
"""The content of a produced file (as text), read back from its object."""
with repository() as repo:
r = _Keyed(repo.conn, t.gl_audit_export, ("export_id",)).get(export_id)
if r is None:
raise Refused(404, f"no export {export_id}")
company_row(repo, caller, r["company_code"])
body = objects.ports.store(repo).get(r["object_key"])
if body is None:
raise Refused(404, f"the object {r['object_key']} is not in the store")
return {"export_id": export_id, "object_key": r["object_key"], "sha256": r["sha256"], "content": body.decode("utf-8")}
@GL.use_case(kind="query", route="GET /audit/change-log", permission=JOURNAL_READ)
def list_change_log(caller: dict, entity: Optional[Code] = None, natural_key: Optional[Annotated[str, Field(max_length=200)]] = None,
limit: Annotated[int, Field(ge=1, le=500)] = 100) -> dict:
"""Every change to master data and settings with before and after (GoBD; append only): who, when, what it was and what it became."""
with repository() as repo:
conds = []
if entity:
conds.append(t.gl_master_audit.c.entity == entity)
if natural_key:
conds.append(t.gl_master_audit.c.natural_key == natural_key)
rows = repo.master_audit.rows(*conds, order=[t.gl_master_audit.c.event_at], limit=limit)
return {"changes": [{"entity": r["entity"], "natural_key": r["natural_key"], "before": json.loads(r["before_json"]) if r["before_json"] else None,
"after": json.loads(r["after_json"]) if r["after_json"] else None, "by": r["by_email"], "at": r["event_at"]} for r in rows]}
_ = (RuleBroken,)