dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

src/brick_fixed_assets/domain/journals.py

"""The journals an asset event posts to gl, per area — built from the figures, balanced before they leave.

gl's POST /journal-entry/journals takes lines as `{account_id | purpose, debit | credit, asset_id, cost_centre,
movement_type, memo}`. Accounts come from the class's account determination for the area; asset clearing is a PURPOSE
(purchasing's account, which this brick never knows the id of). Amounts are exact strings.
"""
from decimal import Decimal
from typing import Optional

from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import ZERO

MAX_LINES = 5000


def line(side: str, amount: Decimal, *, account_id: Optional[int] = None, purpose: Optional[str] = None,
         asset_id: Optional[str] = None, cost_centre: Optional[str] = None, movement: Optional[str] = None,
         memo: str = "") -> dict:
    out: dict = {"account_id": str(account_id)} if account_id is not None else {"purpose": purpose}
    out[side] = str(amount)
    if asset_id:
        out["asset_id"] = asset_id
    if cost_centre:
        out["cost_centre"] = cost_centre
    if movement:
        out["movement_type"] = movement
    if memo:
        out["memo"] = memo
    return out


def put(lines: list, amount: Decimal, debit: dict, credit: dict, **common) -> None:
    """Add a debit and a credit of `amount`; a negative amount is the opposite pair; zero adds nothing."""
    if amount == ZERO:
        return
    if amount < 0:
        debit, credit, amount = credit, debit, -amount
    lines.append(line("debit", amount, **debit, **common))
    lines.append(line("credit", amount, **credit, **common))


def check_balanced(lines: list) -> None:
    d = sum((Decimal(x["debit"]) for x in lines if "debit" in x), ZERO)
    c = sum((Decimal(x["credit"]) for x in lines if "credit" in x), ZERO)
    if d != c:
        raise RuleBroken(f"the journal does not balance: debits {d}, credits {c}")


def negate(lines: list) -> list:
    """A reversal: every line the other way round."""
    out = []
    for x in lines:
        y = dict(x)
        if "debit" in y:
            y["credit"] = y.pop("debit")
        else:
            y["debit"] = y.pop("credit")
        out.append(y)
    return out


def merge(lines: list) -> list:
    """Collapse lines with the same account, side, asset, cost centre and movement (a depreciation run has thousands)."""
    totals: dict = {}
    order: list = []
    for x in lines:
        side = "debit" if "debit" in x else "credit"
        key = (x.get("account_id"), x.get("purpose"), side, x.get("asset_id"), x.get("cost_centre"), x.get("movement_type"))
        if key not in totals:
            totals[key] = [dict(x), Decimal(x[side])]
            order.append(key)
        else:
            totals[key][1] += Decimal(x[side])
    out = []
    for k in order:
        base, amt = totals[k]
        side = k[2]
        base[side] = str(amt)
        out.append(base)
    return out


def split_by_class(lines_by_class: dict) -> list:
    """[(suffix, lines)]: one journal, or one per asset class when it would pass gl's 5,000-line limit."""
    every = [x for ls in lines_by_class.values() for x in ls]
    if len(every) <= MAX_LINES:
        return [("", every)]
    return [(f":{c}", ls) for c, ls in sorted(lines_by_class.items())]