"""Reports data adapter. Routes report-data lookups across: - FUSION: fusion.account.report (added by fusion_accounting_reports, Phase 2) - ENTERPRISE: account.report from account_reports - 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) def trial_balance_via_fusion(self, date_to=None, company_ids=None): # Phase 2 will implement; for now defer to community. return self.trial_balance_via_community(date_to=date_to, company_ids=company_ids) def trial_balance_via_enterprise(self, date_to=None, company_ids=None): # Enterprise account_reports has rich filters; for AI-tool consumption, # the community shape suffices and avoids brittle coupling to Odoo's # report-line internals. return self.trial_balance_via_community(date_to=date_to, company_ids=company_ids) def trial_balance_via_community(self, date_to=None, company_ids=None): domain = [('parent_state', '=', 'posted')] if date_to: domain.append(('date', '<=', date_to)) if company_ids: domain.append(('company_id', 'in', list(company_ids))) Line = self.env['account.move.line'].sudo() groups = Line._read_group( domain=domain, groupby=['account_id'], aggregates=['debit:sum', 'credit:sum'], ) return [ { 'account_id': account.id, 'account_code': account.code, 'account_name': account.name, 'debit': debit_sum, 'credit': credit_sum, 'balance': debit_sum - credit_sum, } 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)