dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/postings.py
"""Ask the ledger to post the value of each stock movement; never write a gl_ table.
One journal per (document, event), with journal_id derived from them, so posting again changes the ledger once. A refusal (a
closed period, a missing account) leaves the posting PENDING with the ledger's reason: the stock move stands, and the Valuation
screen lists it to fix. In export mode (no ledger) nothing is posted and the month's movements are a file for the accountant.
The call is made AFTER the stock commits, in its own short transaction, so a ledger that is down never blocks a shipment.
"""
import json
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from erp_platform import seam as platform_seam
from brick_inventory.application import unit as unit_mod
from brick_inventory.application.unit import Unit
from brick_inventory.domain import valuation as val
from brick_inventory.domain.values import ZERO
from brick_inventory.infrastructure import seams
DEBIT, CREDIT = "debit", "credit"
def journal_id(document_id: str, event: str) -> int:
"""INV-24 / gl GL-30: derived_id(source_document, event) - the ledger recomputes it and refuses any other id."""
return platform_seam.derived_id(document_id, event)
def line(purpose: str, side: str, amount: Decimal, *, valuation_class: str = "*", cost_centre: str = "", memo: str = "") -> dict:
return {"purpose": purpose, "side": side, "amount": f"{abs(Decimal(amount)):.2f}", "valuation_class": valuation_class or "*",
"cost_centre": cost_centre, "memo": memo}
def entry(purpose: str, amount: Decimal, *, valuation_class: str = "*", cost_centre: str = "", memo: str = "",
natural: str = DEBIT) -> Optional[dict]:
"""A line on its natural side for a positive amount, on the other for a negative one; nothing for zero."""
if amount == 0:
return None
other = CREDIT if natural == DEBIT else DEBIT
return line(purpose, natural if amount > 0 else other, amount, valuation_class=valuation_class, cost_centre=cost_centre, memo=memo)
def stage(unit: Unit, document_id: str, event: str, company_code: str, on: date, memo: str, lines: list) -> bool:
"""Plan one journal for (document, event). False when there is nothing to post (all lines are zero)."""
lines = [l for l in lines if l]
if not lines:
return False
debit = sum((Decimal(l["amount"]) for l in lines if l["side"] == DEBIT), ZERO)
credit = sum((Decimal(l["amount"]) for l in lines if l["side"] == CREDIT), ZERO)
if debit != credit:
raise AssertionError(f"the journal for {document_id} {event} does not balance ({debit} against {credit})")
mode = unit.settings.ledger_mode(unit.today, seams.ledger_installed())
if mode == "off":
return False
payload = {"journal_id": str(journal_id(document_id, event)), "source_module": "inventory", "source_document": document_id,
"event": event, "company_code": company_code, "journal_date": on.isoformat(), "memo": memo, "lines": lines}
existing = unit.repo.get("inv_posting", document_id=document_id, event=event)
if existing is not None and existing["status"] in ("posted", "exported"):
return True # a repeat: the ledger has it already
status = "exported" if mode == "export" else "pending"
unit.put("inv_posting", {"document_id": document_id, "event": event, "company_code": company_code,
"journal_id": journal_id(document_id, event), "status": status,
"reason": "ledger mode export: the month's movements are a file" if status == "exported" else "",
"attempts": (existing or {}).get("attempts") or 0, "payload": json.dumps(payload)},
("document_id", "event"), stored=existing is not None)
if status == "pending":
unit.after_commit.append(lambda: send(document_id, event))
return True
def _resolve(repo, payload: dict, on: date) -> dict:
"""Account ids from the account map where it has them; the ledger maps the rest by purpose."""
rows = repo.find("inv_account_map", company_code=payload["company_code"])
out = []
for l in payload["lines"]:
body = {k: v for k, v in l.items() if k not in ("side", "amount", "purpose", "valuation_class", "cost_centre", "memo") and v}
best = None
for r in rows:
if r["purpose"] == l["purpose"] and r["valuation_class"] in (l["valuation_class"], "*") and r["valid_from"] <= on:
if best is None or (r["valuation_class"] != "*", r["valid_from"]) > (best["valuation_class"] != "*", best["valid_from"]):
best = r
if best is not None and best["account_id"] is not None:
body["account_id"] = str(best["account_id"]) # INV-24: account_id OR purpose, exactly one (gl refuses both)
else:
body["purpose"] = l["purpose"]
if l["valuation_class"] != "*":
body["valuation_class"] = l["valuation_class"]
body[l["side"]] = l["amount"]
if l["cost_centre"]:
body["cost_centre"] = l["cost_centre"]
if l["memo"]:
body["memo"] = l["memo"]
out.append(body)
return {**{k: v for k, v in payload.items() if k != "lines"}, "lines": out}
def send(document_id: str, event: str) -> dict:
"""Post one pending journal to the ledger and record its answer (in its own transaction, after the stock committed)."""
with unit_mod.repository() as repo:
row = repo.get("inv_posting", document_id=document_id, event=event)
if row is None or row["status"] in ("posted", "exported"):
return row or {}
payload = json.loads(row["payload"])
on = date.fromisoformat(payload["journal_date"])
attempts = (row["attempts"] or 0) + 1
update = {"document_id": document_id, "event": event, "attempts": attempts}
if not seams.ledger_installed():
update.update(status="pending", reason="the ledger (gl) is not installed: install it, or set ledger_mode to export")
else:
try:
body = _resolve(repo, payload, on)
body.setdefault("currency", seams.company_currency(body["company_code"])) # INV-24: currency (the company's)
answer = seams.post_journal(body)
update.update(status="posted", reason="", posted_at=datetime.now(timezone.utc))
update["journal_id"] = int(answer.get("journal_id") or row["journal_id"])
except seams.Missing as m:
text = m.text
fix = m.reason.get("fix") if isinstance(m.reason, dict) else ""
update.update(status="pending", reason=(f"{text} ({fix})" if fix else text)[:300])
repo.upsert("inv_posting", [update], ("document_id", "event"))
return {**row, **update}
def retry_pending() -> dict:
"""The composition's clock worker (or an external scheduler) retries what the ledger refused."""
with unit_mod.repository() as repo:
pending = [(r["document_id"], r["event"]) for r in repo.find("inv_posting", status="pending")]
done = [send(d, e) for d, e in pending]
return {"tried": len(done), "posted": sum(1 for r in done if r.get("status") == "posted"),
"still_pending": sum(1 for r in done if r.get("status") != "posted")}
def revaluation(unit: Unit, company: str, product_id: str, since: date, step: val.Step, prev: val.Position) -> None:
"""A standard price (or a switch to standard) revalued the stock on hand: stock_valuation against revaluation, posted once."""
stage(unit, f"reval:{company}:{product_id}:{since.isoformat()}", "revaluation", company, since,
f"revaluation of {product_id} to standard {step.after.average}",
[entry("stock_valuation", step.value_change), entry("revaluation", step.value_change, natural=CREDIT)])
def receipt_lines(value: Decimal, purchase_value: Optional[Decimal], price_difference: Decimal, vclass: str, memo: str = "") -> list:
"""stock_valuation against stock_received_not_billed; under standard cost the price difference is its own line, so the
receipt's value + the difference equals what purchasing netted."""
out = [entry("stock_valuation", value, valuation_class=vclass, memo=memo)]
if price_difference:
out.append(entry("price_difference", price_difference, memo=memo))
paid = purchase_value if purchase_value is not None else value
out.append(entry("stock_received_not_billed", paid, natural=CREDIT, memo=memo))
return out
def movement_lines(value: Decimal, expense_purpose: str, vclass: str, *, cost_centre: str = "", memo: str = "") -> list:
"""Stock leaving for an expense (cost of goods sold, loss, consumption, scrap): the expense against stock_valuation. `value`
is the stock's change (negative on the way out): a gain is the same pair the other way round."""
return [entry(expense_purpose, -value, cost_centre=cost_centre, memo=memo),
entry("stock_valuation", value, valuation_class=vclass, memo=memo)]