dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/allocations.py
"""Cost allocation (GL-46, GL-47): cycles of ordered segments that assess and distribute overheads, the statistical key figures
they follow, and the simulate / post / reverse runs.
A cycle is written by a controller and approved by someone else (where allocation_cycle_approval is on): a controller who wrote
a cycle cannot approve it. A run is SIMULATED first (nothing posts), POSTED only while the simulation is current (the senders'
balances unchanged) and no run of the cycle is posted for the period and ledger, and REVERSED with mirror journals, after which
the next run_no may post. A posted run freezes the key figures it used.
"""
import hashlib
import json
from datetime import date
from decimal import Decimal
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from pydantic import Field
from sqlalchemy import func, select
from brick_gl.application import ledgerops
from brick_gl.application.brick import (
ALLOCATION_APPROVE,
ALLOCATION_RUN,
ALLOCATION_SIMULATE,
ALLOCATION_WRITE,
GL,
KEYFIGURE_WRITE,
PERIOD_READ,
claims,
company_row,
decide,
leading_ledger,
real_now,
repository,
same_person,
sub_of,
today,
who,
)
from brick_gl.application.inputs import Code, CompanyCode, KeyFigure, LedgerCode, PeriodId, Reason, _In
from brick_gl.application.posting import finish, post_reversal
from brick_gl.application.scheduled import post_as_gl
from brick_gl.domain import allocation as alloc
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.journal import LineIn
from brick_gl.domain.money import ZERO, money
from brick_gl.domain.readiness import refusal
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, utc
class SegmentModel(_In):
"""One segment of a cycle: who sends, who receives, by what rule."""
seq: Annotated[int, Field(ge=1, le=50)]
kind: Code
rule: Code
sender: dict
receivers: Annotated[list[dict], Field(max_length=500)]
assessment_account: Optional[Code] = None
key_figure: Optional[Code] = None
tracing_account: Optional[Code] = None
CYCLES = _Keyed
_cycle = lambda repo: _Keyed(repo.conn, t.gl_allocation_cycle, ("cycle_id", "version")) # noqa: E731
_run = lambda repo: _Keyed(repo.conn, t.gl_allocation_run, ("run_id",)) # noqa: E731
_figure = lambda repo: _Keyed(repo.conn, t.gl_key_figure_value, ("key_figure", "period_id", "receiver")) # noqa: E731
def _cycle_view(c: dict) -> dict:
return {k: c[k] for k in ("cycle_id", "version", "company_code", "ledger_id", "valid_from_period", "valid_to_period", "status",
"author", "approved_by", "retired_reason")} | {"segments": json.loads(c["segments_json"])}
def _get_cycle(repo, caller, cycle_id: str, version: Optional[int] = None) -> dict:
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_allocation_cycle.select().where(
t.gl_allocation_cycle.c.cycle_id == cycle_id).order_by(t.gl_allocation_cycle.c.version))]
if not rows:
raise Refused(404, f"no allocation cycle {cycle_id}")
row = next((r for r in rows if r["version"] == version), None) if version else rows[-1]
if row is None:
raise Refused(404, f"no version {version} of {cycle_id}")
company_row(repo, caller, row["company_code"])
return row
@GL.use_case(kind="command", route="POST /allocation-cycles", permission=ALLOCATION_WRITE)
def define_allocation_cycle(caller: dict, cycle_id: Code, segments: Annotated[list[SegmentModel], Field(max_length=50)],
valid_from_period: PeriodId, valid_to_period: Optional[PeriodId] = None,
company_code: Optional[CompanyCode] = None, ledger_id: Optional[LedgerCode] = None) -> dict:
"""Define an allocation cycle (or its next version): ordered segments, each an ASSESSMENT (the sender is credited and the
receivers debited on an assessment account) or a DISTRIBUTION (the original accounts move), with senders (an account range,
cost or profit centres), receivers (cost or profit centres) and a rule - fixed_pct (totalling 100), fixed_amount, key_figure
(a statistical key figure such as HEADCOUNT) or tracing_account. The author is recorded (email and sub); the cycle runs only
once another person approves it. A change is a new version."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
segs = [s.model_dump() for s in segments]
alloc.check_cycle(segs)
for s in segs:
for ref in [s.get("assessment_account"), s.get("tracing_account")]:
if ref and repo.accounts.by_code(company["chart_id"], ref) is None:
raise RuleBroken(f"segment {s['seq']}: there is no account {ref} in chart {company['chart_id']}", status=422)
for r in s["receivers"]:
if r.get("cost_centre") and repo.dims.cost_centres.get(r["cost_centre"]) is None:
raise RuleBroken(f"segment {s['seq']}: there is no cost centre {r['cost_centre']}", status=422)
if r.get("profit_centre") and repo.dims.profit_centres.get(r["profit_centre"]) is None:
raise RuleBroken(f"segment {s['seq']}: there is no profit centre {r['profit_centre']}", status=422)
if ledger_id and repo.ledgers.get(ledger_id) is None:
raise Refused(404, f"no ledger {ledger_id}")
last = repo.conn.execute(select(func.max(t.gl_allocation_cycle.c.version)).where(t.gl_allocation_cycle.c.cycle_id == cycle_id)).scalar() or 0
row = {"cycle_id": cycle_id, "version": int(last) + 1, "company_code": code, "ledger_id": ledger_id,
"valid_from_period": valid_from_period, "valid_to_period": valid_to_period, "segments_json": json.dumps(segs),
"status": "draft", "author": who(caller), "author_sub": sub_of(caller), "approved_by": None, "approved_sub": None,
"retired_reason": None}
_cycle(repo).add(row)
return _cycle_view(row)
@GL.use_case(kind="command", route="POST /allocation-cycles/{cycle_id}/approve-allocation-cycle", permission=ALLOCATION_APPROVE)
def approve_allocation_cycle(caller: dict, cycle_id: Code, version: Optional[int] = None) -> dict:
"""Approve a cycle: runnable from its first period. Where allocation_cycle_approval is on, the approver is not the author,
matched on email AND sub (a controller who wrote it cannot approve it). An earlier approved version of the cycle is retired."""
with repository() as repo, decide():
c = _get_cycle(repo, caller, cycle_id, version)
if c["status"] != "draft":
raise RuleBroken(f"{cycle_id} version {c['version']} is {c['status']}, not a draft", status=409)
settings = repo.settings.in_force(today(repo), c["company_code"])
if settings.allocation_cycle_approval and same_person(caller, c["author"], c["author_sub"]):
raise refusal("second_person", "you wrote the cycle, and the owner asks for a second person", "ask another accountant")
for r in [dict(x._mapping) for x in repo.conn.execute(t.gl_allocation_cycle.select().where(
t.gl_allocation_cycle.c.cycle_id == cycle_id, t.gl_allocation_cycle.c.status == "approved"))]:
_cycle(repo).set((cycle_id, r["version"]), status="retired", retired_reason=f"replaced by version {c['version']}")
_cycle(repo).set((cycle_id, c["version"]), status="approved", approved_by=who(caller), approved_sub=sub_of(caller))
return _cycle_view(_get_cycle(repo, caller, cycle_id, c["version"]))
@GL.use_case(kind="command", route="POST /allocation-cycles/{cycle_id}/retire-allocation-cycle", permission=ALLOCATION_APPROVE)
def retire_allocation_cycle(caller: dict, cycle_id: Code, reason: Reason, version: Optional[int] = None) -> dict:
"""Retire a cycle: no new run; posted runs are untouched."""
with repository() as repo, decide():
c = _get_cycle(repo, caller, cycle_id, version)
if c["status"] == "retired":
return _cycle_view(c)
_cycle(repo).set((cycle_id, c["version"]), status="retired", retired_reason=reason)
return _cycle_view(_get_cycle(repo, caller, cycle_id, c["version"]))
@GL.use_case(kind="query", route="GET /allocation-cycles", permission=PERIOD_READ)
def list_allocation_cycles(caller: dict, company_code: Optional[CompanyCode] = None) -> dict:
"""Cycles with their versions and status, and the runs of each (simulated beside posted)."""
with repository() as repo:
company = company_row(repo, caller, company_code)
cycles = [dict(r._mapping) for r in repo.conn.execute(t.gl_allocation_cycle.select().where(
t.gl_allocation_cycle.c.company_code == company["company_code"]))]
runs = [dict(r._mapping) for r in repo.conn.execute(t.gl_allocation_run.select().where(
t.gl_allocation_run.c.company_code == company["company_code"]))]
return {"cycles": [_cycle_view(c) for c in cycles],
"runs": [{k: r[k] for k in ("run_id", "cycle_id", "cycle_version", "period_id", "ledger_id", "run_no", "status")} for r in runs]}
# ── key figures (GL-47) ──────────────────────────────────────────────────────────────────────────────────────────────
@GL.use_case(kind="command", route="POST /key-figures", permission=KEYFIGURE_WRITE, idempotent=True)
def record_key_figure(caller: dict, key_figure: Code, period_id: PeriodId, receiver: Code, value: KeyFigure,
company_code: Optional[CompanyCode] = None) -> dict:
"""Record a statistical key figure (headcount, floor space) for a receiver in a period: NUMERIC(19,4), never negative (422
above it). The receiver must be a valid cost or profit centre. A value a POSTED allocation run used is refused with the run
named - reverse the run first."""
with repository() as repo, decide():
company_row(repo, caller, company_code)
if repo.dims.cost_centres.get(receiver) is None and repo.dims.profit_centres.get(receiver) is None:
raise RuleBroken(f"{receiver} is not a valid cost or profit centre", status=422)
before = _figure(repo).get(key_figure, period_id, receiver)
if before and before["frozen_by_run"]:
raise RuleBroken(f"{key_figure} for {receiver} in {period_id} was used by the posted run {before['frozen_by_run']}: reverse "
"the run first", status=409)
_figure(repo).put({"key_figure": key_figure, "period_id": period_id, "receiver": receiver, "value": value,
"entered_by": who(caller), "entered_at": utc(real_now()), "frozen_by_run": None})
return {"key_figure": key_figure, "period_id": period_id, "receiver": receiver, "value": value}
@GL.use_case(kind="command", route="POST /key-figures/import", permission=KEYFIGURE_WRITE, idempotent=True)
def import_key_figures(caller: dict, csv: Annotated[str, Field(max_length=200000)], company_code: Optional[CompanyCode] = None) -> dict:
"""Record key figures from a CSV with the columns key_figure,period_id,receiver,value (a header row first): all or none."""
rows = []
for i, ln in enumerate(csv.strip().splitlines()):
parts = [p.strip() for p in ln.split(",")]
if i == 0 and parts[:1] == ["key_figure"]:
continue
if len(parts) != 4:
raise Refused(422, f"line {i + 1}: expected key_figure,period_id,receiver,value")
rows.append(parts)
with repository() as repo, decide():
company_row(repo, caller, company_code)
out = []
for kf, pid, rc, val in rows:
try:
v = Decimal(val)
except Exception:
raise Refused(422, f"{val!r} is not a number") from None
if v < 0 or v >= Decimal("1e15") or v.as_tuple().exponent < -4:
raise Refused(422, f"{val} is not a key figure (zero or more, up to 4 places, below 1e15)")
if repo.dims.cost_centres.get(rc) is None and repo.dims.profit_centres.get(rc) is None:
raise Refused(422, f"{rc} is not a valid cost or profit centre")
before = _figure(repo).get(kf, pid, rc)
if before and before["frozen_by_run"]:
raise Refused(409, f"{kf} for {rc} in {pid} was used by the posted run {before['frozen_by_run']}: reverse the run first")
out.append({"key_figure": kf, "period_id": pid, "receiver": rc, "value": v, "entered_by": who(caller),
"entered_at": utc(real_now()), "frozen_by_run": None})
for r in out:
_figure(repo).put(r)
return {"recorded": len(out)}
# ── runs ─────────────────────────────────────────────────────────────────────────────────────────────────────────────
def _sender_rows(repo, company: dict, ledger_id: str, period_id: str, seg: dict) -> list:
"""[(account_id, code, cost_centre, profit_centre, net)] actual postings of the period on the segment's senders."""
snd = seg["sender"]
s = t.gl_split_line.c
extra = []
if snd.get("cost_centres"):
extra.append(s.cost_centre.in_(snd["cost_centres"]))
if snd.get("profit_centres"):
extra.append(s.profit_centre.in_(snd["profit_centres"]))
rows = repo.ledger.sums(company["company_code"], ledger_id, period_id=period_id, group_by=("account_id", "cost_centre", "profit_centre"),
extra=extra)
out = []
for r in rows:
a = repo.accounts.get(r["account_id"])
if a["type"] not in ("expense", "income"):
continue
if not alloc.account_in_range(a["code"], snd.get("account_from"), snd.get("account_to")):
continue
net = r["f_debit"] - r["f_credit"]
if net != 0:
out.append((r["account_id"], a["code"], r["cost_centre"] or "", r["profit_centre"] or "", net))
return sorted(out, key=lambda x: (x[1], x[2], x[3]))
def _compute(repo, company: dict, cycle: dict, ledger_id: str, period_id: str) -> tuple:
"""(segments' movements, sender digest). Segment 2 sees segment 1's result: movements are applied in memory on the actual postings."""
segs = json.loads(cycle["segments_json"])
digest_rows = []
applied: list = [] # (account_id, cost_centre, profit_centre, net) from earlier segments
results = []
chart = company["chart_id"]
def figure_of(kf):
def f(receiver):
row = _figure(repo).get(kf, period_id, receiver)
if row is None:
raise refusal("key_figures_present", f"a receiver has no key figure value for {period_id}: {kf} for {receiver}",
"Allocations: record it (GL-47)")
return row["value"]
return f
for seg in segs:
senders = _sender_rows(repo, company, ledger_id, period_id, seg)
for row in senders:
digest_rows.append([seg["seq"], row[0], row[2], row[3], str(row[4])])
# earlier segments' results change what this segment sees
adj: dict = {}
for acc, cc, pc, net in applied:
adj[(acc, cc, pc)] = adj.get((acc, cc, pc), ZERO) + net
snd = seg["sender"]
seen = {(r[0], r[2], r[3]) for r in senders}
merged = []
for r in senders:
net = r[4] + adj.get((r[0], r[2], r[3]), ZERO)
if net != 0:
merged.append((r[0], r[1], r[2], r[3], net))
for (acc, cc, pc), net in adj.items():
a = repo.accounts.get(acc)
if (acc, cc, pc) in seen or a["type"] not in ("expense", "income"):
continue
if not alloc.account_in_range(a["code"], snd.get("account_from"), snd.get("account_to")):
continue
if snd.get("cost_centres") and cc not in snd["cost_centres"]:
continue
if snd.get("profit_centres") and pc not in snd["profit_centres"]:
continue
if net != 0:
merged.append((acc, a["code"], cc, pc, net))
movements = []
if merged:
total = sum((m[4] for m in merged), ZERO)
def traced(r, seg=seg):
acct = repo.accounts.by_code(chart, seg["tracing_account"])
s_ = t.gl_split_line.c
rows = repo.ledger.sums(company["company_code"], ledger_id, period_id=period_id, account_ids=[acct["account_id"]],
group_by=("account_id",), extra=[s_.cost_centre == r["cost_centre"]] if r.get("cost_centre") else [])
return rows[0]["f_debit"] - rows[0]["f_credit"] if rows else ZERO
ws = alloc.weights(seg, figure_of(seg.get("key_figure")) if seg["rule"] == "key_figure" else None, traced)
rec = seg["receivers"]
if seg["kind"] == "assessment":
acct = repo.accounts.by_code(chart, seg["assessment_account"])["account_id"]
shares = alloc.share(total, ws)
sender_cc = snd["cost_centres"][0] if snd.get("cost_centres") and len(snd["cost_centres"]) == 1 else ""
movements.append({"account_id": acct, "side": "credit", "amount": total, "cost_centre": sender_cc, "profit_centre": "",
"role": "sender"})
for r, sh in zip(rec, shares):
if sh != 0:
movements.append({"account_id": acct, "side": "debit", "amount": sh, "cost_centre": r.get("cost_centre") or "",
"profit_centre": r.get("profit_centre") or "", "role": "receiver"})
else:
for acc, code, cc, pc, net in merged:
shares = alloc.share(net, ws)
movements.append({"account_id": acc, "side": "credit" if net > 0 else "debit", "amount": abs(net), "cost_centre": cc,
"profit_centre": pc, "role": "sender"})
for r, sh in zip(rec, shares):
if sh != 0:
movements.append({"account_id": acc, "side": "debit" if net > 0 else "credit", "amount": abs(sh),
"cost_centre": r.get("cost_centre") or "", "profit_centre": r.get("profit_centre") or "",
"role": "receiver"})
for m in movements:
applied.append((m["account_id"], m["cost_centre"], m["profit_centre"], m["amount"] if m["side"] == "debit" else -m["amount"]))
results.append({"seq": seg["seq"], "kind": seg["kind"], "rule": seg["rule"], "movements": movements})
digest = hashlib.sha256(json.dumps(digest_rows, sort_keys=True).encode()).hexdigest()
return results, digest
def _run_view(r: dict) -> dict:
out = {k: r[k] for k in ("run_id", "cycle_id", "cycle_version", "company_code", "ledger_id", "period_id", "run_no", "status",
"sender_digest", "simulated_by", "posted_by", "reversed_by", "reason")}
out["segments"] = json.loads(r["results_json"] or "[]")
out["journal_ids"] = [x for x in (r["journal_ids"] or "").split(",") if x]
return out
def _runs_of(repo, cycle_id: str, version: int, period_id: str, ledger_id: str) -> list:
return [dict(r._mapping) for r in repo.conn.execute(t.gl_allocation_run.select().where(
t.gl_allocation_run.c.cycle_id == cycle_id, t.gl_allocation_run.c.cycle_version == version,
t.gl_allocation_run.c.period_id == period_id, t.gl_allocation_run.c.ledger_id == ledger_id).order_by(t.gl_allocation_run.c.run_no))]
@GL.use_case(kind="command", route="POST /allocation-cycles/{cycle_id}/simulate-allocation", permission=ALLOCATION_SIMULATE)
def simulate_allocation(caller: dict, cycle_id: Code, period_id: PeriodId, version: Optional[int] = None) -> dict:
"""Simulate an approved cycle for a period: the senders, receivers, rule, key figure and amounts of every segment, with a
digest of the senders' balances - and NOTHING posted. The period must be open and a key figure present for every receiver the
rules need (gate key_figures_present). A later segment sees the earlier one's result."""
with repository() as repo, decide():
c = _get_cycle(repo, caller, cycle_id, version)
company = repo.companies.get(c["company_code"])
if c["status"] != "approved":
raise RuleBroken(f"{cycle_id} version {c['version']} is {c['status']}: only an approved cycle runs", status=409)
if not (c["valid_from_period"] <= period_id and (not c["valid_to_period"] or period_id <= c["valid_to_period"])):
raise RuleBroken(f"{cycle_id} is not valid in {period_id}", status=409)
p = repo.periods.get(c["company_code"], period_id)
if p is None or p["status"] != "open":
raise refusal("period_open", f"the period {period_id} is not open", "simulate in an open period")
ledger_id = c["ledger_id"] or leading_ledger(repo)
results, digest = _compute(repo, company, c, ledger_id, period_id)
runs = _runs_of(repo, cycle_id, c["version"], period_id, ledger_id)
used = [r["run_no"] for r in runs if r["status"] in ("posted", "reversed")]
run_no = max(used, default=0) + 1 # above the high-water mark, never a gap below it
run_id = f"{cycle_id}:{c['version']}:{period_id}:{ledger_id}:{run_no}"
row = {"run_id": run_id, "cycle_id": cycle_id, "cycle_version": c["version"], "company_code": c["company_code"], "ledger_id": ledger_id,
"period_id": period_id, "run_no": run_no, "status": "simulated", "sender_digest": digest,
"results_json": json.dumps(results, default=str), "journal_ids": "", "simulated_by": who(caller), "posted_by": None,
"reversed_by": None, "reason": None}
_run(repo).put(row)
return _run_view(row)
@GL.use_case(kind="command", route="POST /allocation-cycles/{cycle_id}/post-allocation", permission=ALLOCATION_RUN)
def post_allocation(caller: dict, cycle_id: Code, period_id: PeriodId, version: Optional[int] = None) -> dict:
"""Post the simulated run: one journal per segment keyed derived_id('allocation', cycle_id, period_id, ledger_id, run_no,
segment) - an assessment credits the sender and debits the receivers on the assessment account, a distribution moves the
original accounts; the amounts sum to the sender's exactly. Refused while the senders changed since the simulation (gate
simulation_current: simulate again) and while a run of this cycle is posted for the period and ledger (gate
prior_run_reversed). A posted run freezes the key figures it used."""
def unit():
with repository() as repo, decide():
c = _get_cycle(repo, caller, cycle_id, version)
company = repo.companies.get(c["company_code"])
ledger_id = c["ledger_id"] or leading_ledger(repo)
runs = _runs_of(repo, cycle_id, c["version"], period_id, ledger_id)
posted = [r for r in runs if r["status"] == "posted"]
if posted:
raise refusal("prior_run_reversed", f"run {posted[0]['run_id']} of this cycle is posted for {period_id} and {ledger_id}",
"reverse it first")
sims = [r for r in runs if r["status"] == "simulated"]
if not sims:
raise RuleBroken(f"simulate {cycle_id} for {period_id} first", status=409)
run = sims[-1]
# the run number is CLAIMED by a plain INSERT keyed on it (company, cycle/ledger book, period, run_no): two posters
# of one run collide (23505) and the loser retries, finding the run posted
repo.numbers.add({"company_code": c["company_code"], "book": f"allocation-run:{cycle_id}:{c['version']}:{ledger_id}",
"fiscal_year": period_id, "entry_no": run["run_no"], "journal_id": 0, "claimed_at": real_now()})
results, digest = _compute(repo, company, c, ledger_id, period_id)
if digest != run["sender_digest"]:
raise refusal("simulation_current", "postings on the senders since the simulation changed their balance", "simulate again")
p = repo.periods.get(c["company_code"], period_id)
if p is None or p["status"] != "open":
raise refusal("period_open", f"the period {period_id} is not open", "post in an open period")
on = dates.period_bounds(period_id)[1]
at = real_now()
jids, touched_all = [], []
for seg in json.loads(run["results_json"]):
if not seg["movements"]:
continue
lines = [LineIn(account_id=int(m["account_id"]), cost_centre=m["cost_centre"] or None, profit_centre=m["profit_centre"] or None,
memo=f"{cycle_id} segment {seg['seq']}", **{m["side"]: Decimal(str(m["amount"]))}) for m in seg["movements"]]
key = derived_id("allocation", cycle_id, period_id, ledger_id, run["run_no"], seg["seq"])
summary, touched = post_as_gl(
repo, company_code=c["company_code"], key=key,
source_document=f"allocation:{cycle_id}:{period_id}:{ledger_id}:{run['run_no']}:{seg['seq']}", event="allocation", on=on,
currency=company["functional_currency"], lines=lines, ledger_id=c["ledger_id"] or None,
memo=f"{cycle_id} {period_id} segment {seg['seq']}", book="allocation", extras={"allocation_run_id": run["run_id"]},
actor=who(caller), actor_sub=sub_of(caller), at=at, manual=False)
jids.append(str(key))
if touched:
touched_all.append((touched, summary))
for s in json.loads(c["segments_json"]):
if s["rule"] == "key_figure":
for r in s["receivers"]:
_figure(repo).set((s["key_figure"], period_id, r.get("cost_centre") or r.get("profit_centre")), frozen_by_run=run["run_id"])
_run(repo).set((run["run_id"],), status="posted", journal_ids=",".join(jids), posted_by=who(caller))
return _run_view({**run, "status": "posted", "journal_ids": ",".join(jids), "posted_by": who(caller)}), touched_all
out, touched_all = claims(unit)
for touched, summary in touched_all:
finish(touched, summary)
return out
@GL.use_case(kind="command", route="POST /allocation-cycles/{cycle_id}/reverse-allocation", permission=ALLOCATION_RUN)
def reverse_allocation(caller: dict, cycle_id: Code, period_id: PeriodId, reason: Reason, version: Optional[int] = None) -> dict:
"""Reverse the posted run of a cycle for a period with mirror journals of every segment (the period must be open); the key
figures it froze are unfrozen, and the next run (run_no + 1) may then be simulated and posted."""
def unit():
with repository() as repo, decide():
c = _get_cycle(repo, caller, cycle_id, version)
ledger_id = c["ledger_id"] or leading_ledger(repo)
posted = [r for r in _runs_of(repo, cycle_id, c["version"], period_id, ledger_id) if r["status"] == "posted"]
if not posted:
raise RuleBroken(f"{cycle_id} has no posted run for {period_id}", status=409)
run = posted[-1]
p = repo.periods.get(c["company_code"], period_id)
if p is None or p["status"] != "open":
raise refusal("period_open", f"the period {period_id} is closed", "reopen it, or leave the run")
at = real_now()
touched_all = []
for jid in [int(x) for x in run["journal_ids"].split(",") if x]:
j = repo.journals.get(jid)
if j and j["status"] == "posted":
touched, _ = post_reversal(repo, j, j["journal_date"], reason, who(caller), sub_of(caller), at)
if touched:
touched_all.append(touched)
for s in json.loads(c["segments_json"]):
if s["rule"] == "key_figure":
for r in s["receivers"]:
_figure(repo).set((s["key_figure"], period_id, r.get("cost_centre") or r.get("profit_centre")), frozen_by_run=None)
_run(repo).set((run["run_id"],), status="reversed", reversed_by=who(caller), reason=reason)
return _run_view({**run, "status": "reversed", "reversed_by": who(caller), "reason": reason}), touched_all
out, touched_all = claims(unit)
for touched in touched_all:
with repository() as repo:
ledgerops.derive_balances(repo, touched[0], touched[1], touched[2])
return out
_ = (money, ALLOCATION_WRITE, date, ZERO)