286 lines
9.7 KiB
Python
286 lines
9.7 KiB
Python
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]],
|
|
}
|
|
|
|
|
|
def get_profit_loss(env, params):
|
|
return _run_report(env, 'account_reports.profit_and_loss', params)
|
|
|
|
|
|
def get_balance_sheet(env, params):
|
|
return _run_report(env, 'account_reports.balance_sheet', params)
|
|
|
|
|
|
def get_trial_balance(env, params):
|
|
return _run_report(env, 'account_reports.trial_balance_report', params)
|
|
|
|
|
|
def get_cash_flow(env, params):
|
|
return _run_report(env, 'account_reports.cash_flow_statement', params)
|
|
|
|
|
|
def compare_periods(env, params):
|
|
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'),
|
|
})
|
|
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):
|
|
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 {})
|
|
|
|
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 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
|
|
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]],
|
|
}
|
|
|
|
# Monthly breakdown for the year
|
|
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]],
|
|
}
|
|
|
|
# Monthly breakdown
|
|
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,
|
|
}
|