feat(fusion_accounting_reports): line_resolver service for report row computation
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
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
{
|
||||
'name': 'Fusion Accounting Reports',
|
||||
'version': '19.0.1.0.1',
|
||||
'version': '19.0.1.0.2',
|
||||
'category': 'Accounting/Accounting',
|
||||
'summary': 'AI-augmented financial reports (P&L, balance sheet, trial balance, GL).',
|
||||
'description': """
|
||||
|
||||
@@ -2,3 +2,4 @@ from . import date_periods
|
||||
from . import account_hierarchy
|
||||
from . import totaling
|
||||
from . import currency_conversion
|
||||
from . import line_resolver
|
||||
|
||||
143
fusion_accounting_reports/services/line_resolver.py
Normal file
143
fusion_accounting_reports/services/line_resolver.py
Normal file
@@ -0,0 +1,143 @@
|
||||
"""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]
|
||||
@@ -1,3 +1,4 @@
|
||||
from . import test_services_unit
|
||||
from . import test_currency_conversion
|
||||
from . import test_fusion_report
|
||||
from . import test_line_resolver
|
||||
|
||||
96
fusion_accounting_reports/tests/test_line_resolver.py
Normal file
96
fusion_accounting_reports/tests/test_line_resolver.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""Tests for line_resolver."""
|
||||
|
||||
from odoo.tests.common import TransactionCase, tagged
|
||||
from odoo.addons.fusion_accounting_reports.services.line_resolver import resolve
|
||||
from odoo.addons.fusion_accounting_reports.services.totaling import TotalLine
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install')
|
||||
class TestLineResolver(TransactionCase):
|
||||
|
||||
def test_resolve_account_type_prefix(self):
|
||||
line_specs = [
|
||||
{'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
|
||||
]
|
||||
accounts_by_id = {
|
||||
1: {'code': '4000', 'name': 'Sales', 'account_type': 'income_other'},
|
||||
2: {'code': '4100', 'name': 'Service Revenue', 'account_type': 'income_service'},
|
||||
3: {'code': '5000', 'name': 'COGS', 'account_type': 'expense_direct_cost'},
|
||||
}
|
||||
account_totals = {
|
||||
1: TotalLine(balance=10000),
|
||||
2: TotalLine(balance=5000),
|
||||
3: TotalLine(balance=4000),
|
||||
}
|
||||
rows = resolve(
|
||||
line_specs,
|
||||
account_totals=account_totals,
|
||||
accounts_by_id=accounts_by_id,
|
||||
)
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0]['label'], 'Revenue')
|
||||
self.assertEqual(rows[0]['amount'], 15000)
|
||||
|
||||
def test_resolve_subtotal(self):
|
||||
line_specs = [
|
||||
{'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
|
||||
{'label': 'COGS', 'account_type_prefix': 'expense_', 'sign': -1},
|
||||
{'label': 'Gross Profit', 'compute': 'subtotal', 'above': 2},
|
||||
]
|
||||
accounts_by_id = {
|
||||
1: {'code': '4000', 'name': 'Sales', 'account_type': 'income_other'},
|
||||
2: {'code': '5000', 'name': 'COGS', 'account_type': 'expense_direct'},
|
||||
}
|
||||
account_totals = {
|
||||
1: TotalLine(balance=10000),
|
||||
2: TotalLine(balance=4000),
|
||||
}
|
||||
rows = resolve(
|
||||
line_specs,
|
||||
account_totals=account_totals,
|
||||
accounts_by_id=accounts_by_id,
|
||||
)
|
||||
self.assertEqual(len(rows), 3)
|
||||
self.assertEqual(rows[0]['amount'], 10000)
|
||||
self.assertEqual(rows[1]['amount'], -4000)
|
||||
self.assertEqual(rows[2]['amount'], 6000)
|
||||
self.assertTrue(rows[2]['is_subtotal'])
|
||||
|
||||
def test_resolve_with_comparison(self):
|
||||
line_specs = [
|
||||
{'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
|
||||
]
|
||||
accounts_by_id = {
|
||||
1: {'code': '4000', 'name': 'Sales', 'account_type': 'income_other'},
|
||||
}
|
||||
account_totals = {1: TotalLine(balance=12000)}
|
||||
comparison_totals = {1: TotalLine(balance=10000)}
|
||||
rows = resolve(
|
||||
line_specs,
|
||||
account_totals=account_totals,
|
||||
accounts_by_id=accounts_by_id,
|
||||
comparison_totals=comparison_totals,
|
||||
)
|
||||
self.assertEqual(rows[0]['amount'], 12000)
|
||||
self.assertEqual(rows[0]['amount_comparison'], 10000)
|
||||
self.assertAlmostEqual(rows[0]['variance_pct'], 20.0)
|
||||
|
||||
def test_resolve_empty_specs(self):
|
||||
rows = resolve([], account_totals={}, accounts_by_id={})
|
||||
self.assertEqual(rows, [])
|
||||
|
||||
def test_resolve_account_id_drill_down(self):
|
||||
line_specs = [
|
||||
{'label': 'Cash', 'account_id': 99, 'sign': 1},
|
||||
]
|
||||
accounts_by_id = {
|
||||
99: {'code': '1100', 'name': 'Cash', 'account_type': 'asset_cash'},
|
||||
}
|
||||
account_totals = {99: TotalLine(balance=5000)}
|
||||
rows = resolve(
|
||||
line_specs,
|
||||
account_totals=account_totals,
|
||||
accounts_by_id=accounts_by_id,
|
||||
)
|
||||
self.assertEqual(rows[0]['account_id'], 99)
|
||||
self.assertEqual(rows[0]['amount'], 5000)
|
||||
Reference in New Issue
Block a user