e983a370aa
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
293 lines
10 KiB
Python
293 lines
10 KiB
Python
import logging
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Enterprise account.report wrappers — all routed through ReportsAdapter.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_profit_loss(env, 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):
|
|
"""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):
|
|
"""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):
|
|
"""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')
|
|
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}
|
|
|
|
|
|
def answer_financial_question(env, params):
|
|
question = params.get('question', '')
|
|
sql_query = params.get('sql_query')
|
|
if sql_query:
|
|
return {'error': 'Direct SQL not permitted. Use report tools instead.'}
|
|
return {'status': 'info', 'message': f'Use specific report tools to answer: {question}'}
|
|
|
|
|
|
def export_report(env, params):
|
|
"""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'),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
import calendar
|
|
|
|
year = int(params.get('year', date.today().year))
|
|
partner_name = params.get('partner_name')
|
|
date_from = params.get('date_from')
|
|
date_to = params.get('date_to')
|
|
|
|
domain = [
|
|
('move_type', '=', 'out_invoice'),
|
|
('state', '=', 'posted'),
|
|
('company_id', '=', env.company.id),
|
|
]
|
|
|
|
if partner_name:
|
|
partner = env['res.partner'].search([('name', 'ilike', partner_name)], limit=1)
|
|
if partner:
|
|
domain.append(('partner_id', '=', partner.id))
|
|
else:
|
|
return {'error': f'Partner not found: {partner_name}'}
|
|
|
|
if date_from and date_to:
|
|
domain += [('date', '>=', date_from), ('date', '<=', date_to)]
|
|
invoices = env['account.move'].search(domain, order='date desc')
|
|
total = sum(inv.amount_total for inv in invoices)
|
|
return {
|
|
'period': f'{date_from} to {date_to}',
|
|
'count': len(invoices),
|
|
'total': total,
|
|
'invoices': [{
|
|
'id': inv.id, 'name': inv.name, 'partner': inv.partner_id.name,
|
|
'date': str(inv.date), 'amount': inv.amount_total,
|
|
'payment_state': inv.payment_state,
|
|
} for inv in invoices[:30]],
|
|
}
|
|
|
|
months = []
|
|
grand_total = 0
|
|
for month in range(1, 13):
|
|
m_start = f'{year}-{month:02d}-01'
|
|
last_day = calendar.monthrange(year, month)[1]
|
|
m_end = f'{year}-{month:02d}-{last_day}'
|
|
m_domain = domain + [('date', '>=', m_start), ('date', '<=', m_end)]
|
|
invoices = env['account.move'].search(m_domain)
|
|
total = sum(inv.amount_total for inv in invoices)
|
|
grand_total += total
|
|
months.append({
|
|
'month': f'{year}-{month:02d}',
|
|
'month_name': calendar.month_name[month],
|
|
'count': len(invoices),
|
|
'total': round(total, 2),
|
|
})
|
|
|
|
return {
|
|
'year': year,
|
|
'grand_total': round(grand_total, 2),
|
|
'months': months,
|
|
'partner': partner_name or 'All',
|
|
}
|
|
|
|
|
|
def get_billing_summary(env, params):
|
|
"""Get billing (vendor bills) summary — total billed by month or date range."""
|
|
from datetime import date
|
|
import calendar
|
|
|
|
year = int(params.get('year', date.today().year))
|
|
partner_name = params.get('partner_name')
|
|
date_from = params.get('date_from')
|
|
date_to = params.get('date_to')
|
|
|
|
domain = [
|
|
('move_type', '=', 'in_invoice'),
|
|
('state', '=', 'posted'),
|
|
('company_id', '=', env.company.id),
|
|
]
|
|
|
|
if partner_name:
|
|
partner = env['res.partner'].search([('name', 'ilike', partner_name)], limit=1)
|
|
if partner:
|
|
domain.append(('partner_id', '=', partner.id))
|
|
else:
|
|
return {'error': f'Partner not found: {partner_name}'}
|
|
|
|
if date_from and date_to:
|
|
domain += [('date', '>=', date_from), ('date', '<=', date_to)]
|
|
bills = env['account.move'].search(domain, order='date desc')
|
|
total = sum(b.amount_total for b in bills)
|
|
return {
|
|
'period': f'{date_from} to {date_to}',
|
|
'count': len(bills),
|
|
'total': total,
|
|
'bills': [{
|
|
'id': b.id, 'name': b.name, 'partner': b.partner_id.name,
|
|
'date': str(b.date), 'amount': b.amount_total,
|
|
'payment_state': b.payment_state,
|
|
} for b in bills[:30]],
|
|
}
|
|
|
|
months = []
|
|
grand_total = 0
|
|
for month in range(1, 13):
|
|
m_start = f'{year}-{month:02d}-01'
|
|
last_day = calendar.monthrange(year, month)[1]
|
|
m_end = f'{year}-{month:02d}-{last_day}'
|
|
m_domain = domain + [('date', '>=', m_start), ('date', '<=', m_end)]
|
|
bills = env['account.move'].search(m_domain)
|
|
total = sum(b.amount_total for b in bills)
|
|
grand_total += total
|
|
months.append({
|
|
'month': f'{year}-{month:02d}',
|
|
'month_name': calendar.month_name[month],
|
|
'count': len(bills),
|
|
'total': round(total, 2),
|
|
})
|
|
|
|
return {
|
|
'year': year,
|
|
'grand_total': round(grand_total, 2),
|
|
'months': months,
|
|
'partner': partner_name or 'All',
|
|
}
|
|
|
|
|
|
def get_collections_summary(env, params):
|
|
"""Get payment collections summary — how much was collected (received) in a period."""
|
|
date_from = params.get('date_from')
|
|
date_to = params.get('date_to')
|
|
if not date_from or not date_to:
|
|
from datetime import date
|
|
today = date.today()
|
|
date_from = date_from or f'{today.year}-{today.month:02d}-01'
|
|
date_to = date_to or str(today)
|
|
|
|
payments = env['account.payment'].search([
|
|
('payment_type', '=', 'inbound'),
|
|
('state', '=', 'posted'),
|
|
('date', '>=', date_from),
|
|
('date', '<=', date_to),
|
|
('company_id', '=', env.company.id),
|
|
], order='date desc')
|
|
|
|
total = sum(p.amount for p in payments)
|
|
by_partner = {}
|
|
for p in payments:
|
|
pname = p.partner_id.name if p.partner_id else 'Unknown'
|
|
by_partner.setdefault(pname, {'count': 0, 'total': 0})
|
|
by_partner[pname]['count'] += 1
|
|
by_partner[pname]['total'] += p.amount
|
|
|
|
top_partners = sorted(by_partner.items(), key=lambda x: -x[1]['total'])[:15]
|
|
|
|
return {
|
|
'period': f'{date_from} to {date_to}',
|
|
'total_collected': round(total, 2),
|
|
'payment_count': len(payments),
|
|
'by_partner': [{'partner': k, 'count': v['count'], 'total': round(v['total'], 2)} for k, v in top_partners],
|
|
}
|
|
|
|
|
|
TOOLS = {
|
|
'get_profit_loss': get_profit_loss,
|
|
'get_balance_sheet': get_balance_sheet,
|
|
'get_trial_balance': get_trial_balance,
|
|
'get_cash_flow': get_cash_flow,
|
|
'compare_periods': compare_periods,
|
|
'answer_financial_question': answer_financial_question,
|
|
'export_report': export_report,
|
|
'get_invoicing_summary': get_invoicing_summary,
|
|
'get_billing_summary': get_billing_summary,
|
|
'get_collections_summary': get_collections_summary,
|
|
}
|