Pure-Python helper that resolves a fusion.report's line_specs against account_totals -> ordered list of report row dicts. Supports three spec types: account_type_prefix (sum accounts by type), account_id (single account, drill-downable), and compute='subtotal' (sum last N rows). Comparison-period support: variance_pct computed automatically when comparison_totals are supplied. 5 new tests, 32 total passing. Made-with: Cursor
144 lines
4.9 KiB
Python
144 lines
4.9 KiB
Python
"""Resolve a fusion.report definition into report rows.
|
|
|
|
Pure-Python: takes line_specs (list of dicts), a period, and aggregated
|
|
move-line data (per-account totals) - returns ordered list of report row
|
|
dicts ready for the OWL frontend or PDF rendering.
|
|
|
|
Row shape:
|
|
{
|
|
'id': 'line_<index>',
|
|
'label': str,
|
|
'level': int, # indentation depth
|
|
'is_subtotal': bool,
|
|
'amount': float,
|
|
'amount_comparison': float | None,
|
|
'variance_pct': float | None,
|
|
'account_id': int | None, # for drill-down (None for subtotals)
|
|
'children': list[dict], # populated when expanded
|
|
}"""
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from .totaling import TotalLine
|
|
|
|
|
|
@dataclass
|
|
class ReportRow:
|
|
id: str
|
|
label: str
|
|
level: int = 0
|
|
is_subtotal: bool = False
|
|
amount: float = 0.0
|
|
amount_comparison: float | None = None
|
|
variance_pct: float | None = None
|
|
account_id: int | None = None
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'label': self.label,
|
|
'level': self.level,
|
|
'is_subtotal': self.is_subtotal,
|
|
'amount': self.amount,
|
|
'amount_comparison': self.amount_comparison,
|
|
'variance_pct': self.variance_pct,
|
|
'account_id': self.account_id,
|
|
}
|
|
|
|
|
|
def resolve(
|
|
line_specs: list[dict],
|
|
*,
|
|
account_totals: dict[int, TotalLine],
|
|
accounts_by_id: dict[int, dict],
|
|
comparison_totals: dict[int, TotalLine] | None = None,
|
|
) -> list[dict]:
|
|
"""Resolve line_specs against actual account totals -> list of row dicts.
|
|
|
|
Args:
|
|
line_specs: report definition line specs (from fusion.report.line_specs).
|
|
account_totals: {account_id: TotalLine} for the period.
|
|
accounts_by_id: {account_id: {code, name, account_type, ...}}.
|
|
comparison_totals: optional {account_id: TotalLine} for comparison period.
|
|
|
|
Returns: list of row dicts."""
|
|
rows: list[ReportRow] = []
|
|
|
|
for idx, spec in enumerate(line_specs):
|
|
if spec.get('compute') == 'subtotal':
|
|
n = spec.get('above', 1)
|
|
sign = spec.get('sign', 1)
|
|
recent = [r.amount for r in rows[-n:] if not r.is_subtotal]
|
|
row = ReportRow(
|
|
id=f'line_{idx}',
|
|
label=spec.get('label', 'Subtotal'),
|
|
level=spec.get('level', 0),
|
|
is_subtotal=True,
|
|
amount=sum(recent) * sign,
|
|
)
|
|
if comparison_totals is not None:
|
|
comp_recent = [
|
|
r.amount_comparison
|
|
for r in rows[-n:]
|
|
if not r.is_subtotal and r.amount_comparison is not None
|
|
]
|
|
row.amount_comparison = (
|
|
sum(comp_recent) * sign if comp_recent else None
|
|
)
|
|
rows.append(row)
|
|
|
|
elif spec.get('account_type_prefix'):
|
|
prefix = spec['account_type_prefix']
|
|
sign = spec.get('sign', 1)
|
|
matched_ids = [
|
|
aid for aid, info in accounts_by_id.items()
|
|
if info.get('account_type', '').startswith(prefix)
|
|
]
|
|
amount = sum(
|
|
account_totals.get(aid, TotalLine()).balance * sign
|
|
for aid in matched_ids
|
|
)
|
|
row = ReportRow(
|
|
id=f'line_{idx}',
|
|
label=spec.get('label', prefix),
|
|
level=spec.get('level', 0),
|
|
amount=amount,
|
|
)
|
|
if comparison_totals is not None:
|
|
comp_amount = sum(
|
|
comparison_totals.get(aid, TotalLine()).balance * sign
|
|
for aid in matched_ids
|
|
)
|
|
row.amount_comparison = comp_amount
|
|
if comp_amount != 0:
|
|
row.variance_pct = (
|
|
(amount - comp_amount) / abs(comp_amount)
|
|
) * 100
|
|
rows.append(row)
|
|
|
|
elif spec.get('account_id'):
|
|
aid = spec['account_id']
|
|
sign = spec.get('sign', 1)
|
|
tot = account_totals.get(aid, TotalLine())
|
|
label = spec.get('label') or accounts_by_id.get(aid, {}).get(
|
|
'name', f'Account {aid}'
|
|
)
|
|
row = ReportRow(
|
|
id=f'line_{idx}',
|
|
label=label,
|
|
level=spec.get('level', 0),
|
|
amount=tot.balance * sign,
|
|
account_id=aid,
|
|
)
|
|
if comparison_totals is not None:
|
|
comp = comparison_totals.get(aid, TotalLine())
|
|
row.amount_comparison = comp.balance * sign
|
|
if row.amount_comparison and row.amount_comparison != 0:
|
|
row.variance_pct = (
|
|
(row.amount - row.amount_comparison)
|
|
/ abs(row.amount_comparison)
|
|
) * 100
|
|
rows.append(row)
|
|
|
|
return [r.to_dict() for r in rows]
|