admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(fusion_accounting_ai): route reporting tools through ReportsAdapter

Browse Source 
Task 13 Step 9 of phase-0 plan.

All Enterprise account.report entry points now go through ReportsAdapter:

  - get_profit_loss → ReportsAdapter.run_report(account_reports.profit_and_loss)
  - get_balance_sheet → ReportsAdapter.run_report(account_reports.balance_sheet)
  - get_trial_balance → ReportsAdapter.run_report(...) with Community fallback
    to the existing trial_balance() account.move.line aggregation
  - get_cash_flow → ReportsAdapter.run_report(account_reports.cash_flow_statement)
  - compare_periods → two run_report() calls
  - export_report → ReportsAdapter.export_report() (PDF/XLSX via Enterprise)

ReportsAdapter extended with:

  - run_report(ref_id, date_from, date_to, limit) — generic Enterprise
    account.report wrapper. Enterprise mode returns {report_name, lines};
    Community mode returns a graceful error dict pointing users at the
    raw trial_balance() aggregation tool.
  - export_report(ref_id, fmt, date_from, date_to) — Enterprise-only PDF/XLSX
    export; Community mode returns an error dict.

Pure-Community tools in reporting.py (get_invoicing_summary, get_billing_summary,
get_collections_summary) unchanged — they aggregate account.move /
account.payment directly which is tri-mode safe.

3 new data-adapter tests added for run_report happy/error paths and
export_report shape. Total: 12 tests, all passing on westin-v19.

Made-with: Cursor
This commit is contained in:
gsinghpal
2026-04-18 23:33:54 -04:00
parent 2ead351c30
commit e983a370aa
3 changed files with 222 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
fusion_accounting_ai/services/data_adapters/reports.py
View File

@@ -6,14 +6,22 @@ Routes report-data lookups across:
- COMMUNITY: raw aggregations on account.move.line
"""
import base64
import logging
from .base import DataAdapter
from ._registry import register_adapter
_logger = logging.getLogger(__name__)
class ReportsAdapter(DataAdapter):
FUSION_MODEL = 'fusion.account.report'
ENTERPRISE_MODULE = 'account_reports'
# ------------------------------------------------------------------
# trial_balance (Community-computable from account.move.line)
# ------------------------------------------------------------------
def trial_balance(self, date_to=None, company_ids=None):
return self._dispatch('trial_balance', date_to=date_to, company_ids=company_ids)
@@ -52,5 +60,111 @@ class ReportsAdapter(DataAdapter):
for account, debit_sum, credit_sum in groups
]
# ------------------------------------------------------------------
# run_report — generic Enterprise account.report wrapper
#
# Returns either {'report_name', 'lines'} or {'error': ...}.
# Used by profit_loss / balance_sheet / cash_flow / trial_balance_lines
# tool wrappers that want Enterprise's hierarchical report shape when
# available.
# ------------------------------------------------------------------
def run_report(self, ref_id, date_from=None, date_to=None, limit=100):
return self._dispatch(
'run_report',
ref_id=ref_id, date_from=date_from, date_to=date_to, limit=limit,
)
def run_report_via_fusion(self, ref_id, date_from=None, date_to=None, limit=100):
# Phase 2: fusion.account.report will implement equivalent rendering.
return self.run_report_via_community(
ref_id=ref_id, date_from=date_from, date_to=date_to, limit=limit,
)
def run_report_via_enterprise(self, ref_id, date_from=None, date_to=None, limit=100):
try:
report = self.env.ref(ref_id, raise_if_not_found=False)
except Exception:
report = None
if not report:
return {'error': f'Report {ref_id} not found'}
date_opts = {}
if date_from:
date_opts['date_from'] = date_from
if date_to:
date_opts['date_to'] = date_to
options = report.get_options({'date': date_opts} if date_opts else {})
lines = report._get_lines(options)
return {
'report_name': report.name,
'lines': [{
'name': line.get('name', ''),
'level': line.get('level', 0),
'columns': [c.get('no_format', c.get('name', '')) for c in line.get('columns', [])],
} for line in lines[:limit]],
}
def run_report_via_community(self, ref_id, date_from=None, date_to=None, limit=100):
return {
'error': (
f'Report {ref_id!r} is only available when account_reports (Enterprise) '
'or a fusion reports module is installed. For pure Community installs, '
'use the raw trial_balance() adapter method or the tools that aggregate '
'account.move.line directly.'
),
}
# ------------------------------------------------------------------
# export_report — Enterprise-only PDF/XLSX export
# ------------------------------------------------------------------
def export_report(self, ref_id, fmt='pdf', date_from=None, date_to=None):
return self._dispatch(
'export_report',
ref_id=ref_id, fmt=fmt, date_from=date_from, date_to=date_to,
)
def export_report_via_fusion(self, ref_id, fmt='pdf', date_from=None, date_to=None):
return self.export_report_via_community(
ref_id=ref_id, fmt=fmt, date_from=date_from, date_to=date_to,
)
def export_report_via_enterprise(self, ref_id, fmt='pdf', date_from=None, date_to=None):
try:
report = self.env.ref(ref_id, raise_if_not_found=False)
except Exception:
report = None
if not report:
return {'error': f'Report {ref_id} not found'}
date_opts = {}
if date_from:
date_opts['date_from'] = date_from
if date_to:
date_opts['date_to'] = date_to
options = report.get_options({'date': date_opts} if date_opts else {})
try:
if fmt == 'xlsx':
result = report.dispatch_report_action(options, 'export_to_xlsx')
else:
result = report.dispatch_report_action(options, 'export_to_pdf')
if isinstance(result, dict) and result.get('file_content'):
return {
'file_name': result.get('file_name', f'report.{fmt}'),
'file_type': result.get('file_type', fmt),
'file_content_b64': base64.b64encode(result['file_content']).decode(),
}
return {
'status': 'generated',
'message': f'Report exported as {fmt}. Use the Odoo UI to download.',
}
except Exception as e:
return {'error': f'Export failed: {str(e)}'}
def export_report_via_community(self, ref_id, fmt='pdf', date_from=None, date_to=None):
return {
'error': (
f'Exporting report {ref_id!r} is only available with Enterprise '
'account_reports installed.'
),
}
register_adapter('reports', ReportsAdapter)

159
fusion_accounting_ai/services/tools/reporting.py
View File

@@ -1,67 +1,91 @@
import logging
import base64
_logger = logging.getLogger(__name__)
def _get_report(env, ref_id):
try:
return env.ref(ref_id)
except Exception:
return None
def _run_report(env, report_ref, params):
report = _get_report(env, report_ref)
if not report:
return {'error': f'Report {report_ref} not found'}
date_opts = {}
if params.get('date_from'):
date_opts['date_from'] = params['date_from']
if params.get('date_to'):
date_opts['date_to'] = params['date_to']
options = report.get_options({'date': date_opts} if date_opts else {})
lines = report._get_lines(options)
return {
'report_name': report.name,
'lines': [{
'name': l.get('name', ''),
'level': l.get('level', 0),
'columns': [c.get('no_format', c.get('name', '')) for c in l.get('columns', [])],
} for l in lines[:100]],
}
# ---------------------------------------------------------------------------
# Enterprise account.report wrappers — all routed through ReportsAdapter.
# ---------------------------------------------------------------------------
def get_profit_loss(env, params):
return _run_report(env, 'account_reports.profit_and_loss', params)
"""Route through ReportsAdapter for tri-mode consistency."""
from ..data_adapters import get_adapter
adapter = get_adapter(env, 'reports')
return adapter.run_report(
ref_id='account_reports.profit_and_loss',
date_from=params.get('date_from'),
date_to=params.get('date_to'),
)
def get_balance_sheet(env, params):
return _run_report(env, 'account_reports.balance_sheet', params)
"""Route through ReportsAdapter for tri-mode consistency."""
from ..data_adapters import get_adapter
adapter = get_adapter(env, 'reports')
return adapter.run_report(
ref_id='account_reports.balance_sheet',
date_from=params.get('date_from'),
date_to=params.get('date_to'),
)
def get_trial_balance(env, params):
return _run_report(env, 'account_reports.trial_balance_report', params)
"""Route through ReportsAdapter for tri-mode consistency.
In Enterprise mode returns the hierarchical report lines. In Community
mode falls back to the adapter's trial_balance() aggregation so the tool
continues to return useful data with a compatible shape.
"""
from ..data_adapters import get_adapter
adapter = get_adapter(env, 'reports')
result = adapter.run_report(
ref_id='account_reports.trial_balance_report',
date_from=params.get('date_from'),
date_to=params.get('date_to'),
)
if isinstance(result, dict) and result.get('error'):
rows = adapter.trial_balance(
date_to=params.get('date_to'),
company_ids=[env.company.id],
)
return {
'report_name': 'Trial Balance (Community aggregation)',
'lines': [{
'name': f"{r['account_code']} {r['account_name']}",
'level': 2,
'columns': [r['debit'], r['credit'], r['balance']],
} for r in rows],
}
return result
def get_cash_flow(env, params):
return _run_report(env, 'account_reports.cash_flow_statement', params)
"""Route through ReportsAdapter for tri-mode consistency."""
from ..data_adapters import get_adapter
adapter = get_adapter(env, 'reports')
return adapter.run_report(
ref_id='account_reports.cash_flow_statement',
date_from=params.get('date_from'),
date_to=params.get('date_to'),
)
def compare_periods(env, params):
"""Run the same report over two periods and return both results. Routes
both runs through ReportsAdapter."""
from ..data_adapters import get_adapter
adapter = get_adapter(env, 'reports')
report_ref = params.get('report_ref', 'account_reports.profit_and_loss')
report = _get_report(env, report_ref)
if not report:
return {'error': f'Report {report_ref} not found'}
period1 = _run_report(env, report_ref, {
'date_from': params.get('period1_from'),
'date_to': params.get('period1_to'),
})
period2 = _run_report(env, report_ref, {
'date_from': params.get('period2_from'),
'date_to': params.get('period2_to'),
})
period1 = adapter.run_report(
ref_id=report_ref,
date_from=params.get('period1_from'),
date_to=params.get('period1_to'),
)
period2 = adapter.run_report(
ref_id=report_ref,
date_from=params.get('period2_from'),
date_to=params.get('period2_to'),
)
return {'period_1': period1, 'period_2': period2}
@@ -74,42 +98,27 @@ def answer_financial_question(env, params):
def export_report(env, params):
report_ref = params.get('report_ref', 'account_reports.profit_and_loss')
fmt = params.get('format', 'pdf')
report = _get_report(env, report_ref)
if not report:
return {'error': f'Report {report_ref} not found'}
date_opts = {}
if params.get('date_from'):
date_opts['date_from'] = params['date_from']
if params.get('date_to'):
date_opts['date_to'] = params['date_to']
options = report.get_options({'date': date_opts} if date_opts else {})
"""Route through ReportsAdapter for tri-mode consistency."""
from ..data_adapters import get_adapter
adapter = get_adapter(env, 'reports')
return adapter.export_report(
ref_id=params.get('report_ref', 'account_reports.profit_and_loss'),
fmt=params.get('format', 'pdf'),
date_from=params.get('date_from'),
date_to=params.get('date_to'),
)
try:
if fmt == 'xlsx':
result = report.dispatch_report_action(options, 'export_to_xlsx')
else:
result = report.dispatch_report_action(options, 'export_to_pdf')
if isinstance(result, dict) and result.get('file_content'):
return {
'file_name': result.get('file_name', f'report.{fmt}'),
'file_type': result.get('file_type', fmt),
'file_content_b64': base64.b64encode(result['file_content']).decode(),
}
return {
'status': 'generated',
'message': f'Report exported as {fmt}. Use the Odoo UI to download.',
}
except Exception as e:
return {'error': f'Export failed: {str(e)}'}
# ---------------------------------------------------------------------------
# Pure-Community tools — search account.move / account.payment directly.
# These are tri-mode safe (the data lives in the same tables regardless of
# install profile) so they don't need adapter routing.
# ---------------------------------------------------------------------------
def get_invoicing_summary(env, params):
"""Get invoicing summary — total invoiced by month, by partner, or for a date range.
Supports: monthly breakdown for a year, current month totals, or filtered by partner."""
from datetime import date, timedelta
from datetime import date
import calendar
year = int(params.get('year', date.today().year))
@@ -145,7 +154,6 @@ def get_invoicing_summary(env, params):
} for inv in invoices[:30]],
}
# Monthly breakdown for the year
months = []
grand_total = 0
for month in range(1, 13):
@@ -209,7 +217,6 @@ def get_billing_summary(env, params):
} for b in bills[:30]],
}
# Monthly breakdown
months = []
grand_total = 0
for month in range(1, 13):

28
fusion_accounting_ai/tests/test_data_adapters.py
View File

@@ -66,15 +66,37 @@ class TestReportsAdapter(TransactionCase):
def test_trial_balance_returns_rows_in_pure_community(self):
adapter = get_adapter(self.env, 'reports')
# Compute an empty-filter trial balance for the current company. Should
# return a list (possibly empty in a fresh test DB) without errors.
result = adapter.trial_balance()
self.assertIsInstance(result, list)
# Each row should have account_id and balance keys
for row in result:
self.assertIn('account_id', row)
self.assertIn('balance', row)
def test_run_report_returns_lines_or_error_dict(self):
"""run_report() must always return either an Enterprise-shaped
{'report_name', 'lines'} dict or an {'error': ...} dict — never raise."""
adapter = get_adapter(self.env, 'reports')
result = adapter.run_report(ref_id='account_reports.profit_and_loss')
self.assertIsInstance(result, dict)
# Either a report_name+lines response or an error — both valid
self.assertTrue(
('lines' in result and 'report_name' in result) or 'error' in result,
f"Unexpected result shape: {result!r}",
)
def test_run_report_with_unknown_ref_returns_error(self):
adapter = get_adapter(self.env, 'reports')
result = adapter.run_report(ref_id='nonexistent.report.xml_id')
self.assertIsInstance(result, dict)
self.assertIn('error', result)
def test_export_report_returns_dict(self):
adapter = get_adapter(self.env, 'reports')
result = adapter.export_report(
ref_id='account_reports.profit_and_loss', fmt='pdf',
)
self.assertIsInstance(result, dict)
@tagged('post_install', '-at_install')
class TestFollowupAdapter(TransactionCase):