# Fusion Accounting - Journal Report Handler # Full journal audit with tax summaries, PDF/XLSX export, bank journal support import io import datetime from PIL import ImageFont from markupsafe import Markup from collections import defaultdict from odoo import models, _ from odoo.tools import SQL from odoo.tools.misc import file_path try: from odoo.tools.misc import xlsxwriter except ImportError: import xlsxwriter XLSX_GRAY_200 = '#EEEEEE' XLSX_BORDER_COLOR = '#B4B4B4' XLSX_FONT_SIZE_DEFAULT = 8 XLSX_FONT_SIZE_HEADING = 11 class FusionJournalReportHandler(models.AbstractModel): """Custom handler for the Journal Audit report. Produces detailed per-journal line listings, tax summaries (per-journal and global), and supports PDF and XLSX export.""" _name = "account.journal.report.handler" _inherit = "account.report.custom.handler" _description = "Journal Report Custom Handler" # ================================================================ # OPTIONS # ================================================================ def _custom_options_initializer(self, report, options, previous_options): options['ignore_totals_below_sections'] = True options['show_payment_lines'] = previous_options.get('show_payment_lines', True) def _get_custom_display_config(self): return { 'css_custom_class': 'journal_report', 'pdf_css_custom_class': 'journal_report_pdf', 'components': { 'AccountReportLine': 'fusion_accounting.JournalReportLine', }, 'templates': { 'AccountReportFilters': 'fusion_accounting.JournalReportFilters', 'AccountReportLineName': 'fusion_accounting.JournalReportLineName', }, } # ================================================================ # CUSTOM ENGINE # ================================================================ def _report_custom_engine_journal_report( self, expressions, options, date_scope, current_groupby, next_groupby, offset=0, limit=None, warnings=None, ): def _assemble_result(groupby_key, row): if groupby_key == 'account_id': code = row['account_code'][0] elif groupby_key == 'journal_id': code = row['journal_code'][0] else: code = None return row['grouping_key'], { 'code': code, 'credit': row['credit'], 'debit': row['debit'], 'balance': row['balance'] if groupby_key == 'account_id' else None, } report = self.env['account.report'].browse(options['report_id']) report._check_groupby_fields( (next_groupby.split(',') if next_groupby else []) + ([current_groupby] if current_groupby else []), ) if not current_groupby: return {'code': None, 'debit': None, 'credit': None, 'balance': None} qry = report._get_report_query(options, 'strict_range') acct_alias = qry.join( lhs_alias='account_move_line', lhs_column='account_id', rhs_table='account_account', rhs_column='id', link='account_id', ) acct_code = self.env['account.account']._field_to_sql(acct_alias, 'code', qry) gb_col = SQL.identifier('account_move_line', current_groupby) sel_gb = SQL('%s AS grouping_key', gb_col) stmt = SQL( """ SELECT %(sel_gb)s, ARRAY_AGG(DISTINCT %(acct_code)s) AS account_code, ARRAY_AGG(DISTINCT j.code) AS journal_code, SUM("account_move_line".debit) AS debit, SUM("account_move_line".credit) AS credit, SUM("account_move_line".balance) AS balance FROM %(tbl)s JOIN account_move am ON am.id = account_move_line.move_id JOIN account_journal j ON j.id = am.journal_id JOIN res_company cp ON cp.id = am.company_id WHERE %(pmt_filter)s AND %(where)s GROUP BY %(gb_col)s ORDER BY %(gb_col)s """, sel_gb=sel_gb, acct_code=acct_code, tbl=qry.from_clause, where=qry.where_clause, pmt_filter=self._get_payment_lines_filter_case_statement(options), gb_col=gb_col, ) self.env.cr.execute(stmt) return [_assemble_result(current_groupby, r) for r in self.env.cr.dictfetchall()] # ================================================================ # LINE POST-PROCESSING # ================================================================ def _custom_line_postprocessor(self, report, options, lines): """Inject tax summary sub-tables after journal account sections and append a global tax summary when applicable.""" enriched = [] for idx, ln in enumerate(lines): enriched.append(ln) line_model, res_id = report._get_model_info_from_id(ln['id']) if line_model == 'account.journal': ln['journal_id'] = res_id elif line_model == 'account.account': id_map = report._get_res_ids_from_line_id( ln['id'], ['account.journal', 'account.account'], ) ln['journal_id'] = id_map['account.journal'] ln['account_id'] = id_map['account.account'] ln['date'] = options['date'] jnl = self.env['account.journal'].browse(ln['journal_id']) is_last_acct = ( idx + 1 == len(lines) or report._get_model_info_from_id(lines[idx + 1]['id'])[0] != 'account.account' ) if is_last_acct and self._section_has_tax(options, jnl.id): enriched.append({ 'id': report._get_generic_line_id( False, False, parent_line_id=ln['parent_id'], markup='tax_report_section', ), 'name': '', 'parent_id': ln['parent_id'], 'journal_id': jnl.id, 'is_tax_section_line': True, 'columns': [], 'colspan': len(options['columns']) + 1, 'level': 4, **self._get_tax_summary_section( options, {'id': jnl.id, 'type': jnl.type}, ), }) if report._get_model_info_from_id(lines[0]['id'])[0] == 'account.report.line': if self._section_has_tax(options, False): enriched.append({ 'id': report._get_generic_line_id(False, False, markup='tax_report_section_heading'), 'name': _('Global Tax Summary'), 'level': 0, 'columns': [], 'unfoldable': False, 'colspan': len(options['columns']) + 1, }) enriched.append({ 'id': report._get_generic_line_id(False, False, markup='tax_report_section'), 'name': '', 'is_tax_section_line': True, 'columns': [], 'colspan': len(options['columns']) + 1, 'level': 4, 'class': 'o_account_reports_ja_subtable', **self._get_tax_summary_section(options), }) return enriched # ================================================================ # PDF EXPORT # ================================================================ def export_to_pdf(self, options): report = self.env['account.report'].browse(options['report_id']) base_url = report.get_base_url() print_opts = { **report.get_options(previous_options={**options, 'export_mode': 'print'}), 'css_custom_class': self._get_custom_display_config().get( 'pdf_css_custom_class', 'journal_report_pdf', ), } ctx = {'mode': 'print', 'base_url': base_url, 'company': self.env.company} footer_html = self.env['ir.actions.report']._render_template( 'fusion_accounting.internal_layout', values=ctx, ) footer_html = self.env['ir.actions.report']._render_template( 'web.minimal_layout', values=dict(ctx, subst=True, body=Markup(footer_html.decode())), ) doc_data = self._generate_document_data_for_export(report, print_opts, 'pdf') body_html = self.env['ir.qweb']._render( 'fusion_accounting.journal_report_pdf_export_main', {'report': report, 'options': print_opts, 'base_url': base_url, 'document_data': doc_data}, ) pdf_bytes = io.BytesIO( self.env['ir.actions.report']._run_wkhtmltopdf( [body_html], footer=footer_html.decode(), landscape=False, specific_paperformat_args={ 'data-report-margin-top': 10, 'data-report-header-spacing': 10, 'data-report-margin-bottom': 15, }, ) ) result = pdf_bytes.getvalue() pdf_bytes.close() return { 'file_name': report.get_default_report_filename(print_opts, 'pdf'), 'file_content': result, 'file_type': 'pdf', } # ================================================================ # XLSX EXPORT # ================================================================ def export_to_xlsx(self, options, response=None): wb_buffer = io.BytesIO() wb = xlsxwriter.Workbook(wb_buffer, {'in_memory': True, 'strings_to_formulas': False}) report = self.env['account.report'].search([('id', '=', options['report_id'])], limit=1) print_opts = report.get_options(previous_options={**options, 'export_mode': 'print'}) doc_data = self._generate_document_data_for_export(report, print_opts, 'xlsx') font_cache = {} for sz in (XLSX_FONT_SIZE_HEADING, XLSX_FONT_SIZE_DEFAULT): font_cache[sz] = defaultdict() for variant in ('Reg', 'Bol', 'RegIta', 'BolIta'): try: path = f'web/static/fonts/lato/Lato-{variant}-webfont.ttf' font_cache[sz][variant] = ImageFont.truetype(file_path(path), sz) except (OSError, FileNotFoundError): font_cache[sz][variant] = ImageFont.load_default() for jv in doc_data['journals_vals']: cx, cy = 0, 0 ws = wb.add_worksheet(jv['name'][:31]) cols = jv['columns'] for col in cols: alignment = 'right' if 'o_right_alignment' in col.get('class', '') else 'left' self._write_cell( cx, cy, col['name'], 1, False, report, font_cache, wb, ws, XLSX_FONT_SIZE_HEADING, True, XLSX_GRAY_200, alignment, 2, 2, ) cx += 1 cy += 1 cx = 0 for row in jv['lines'][:-1]: first_aml = False for col in cols: top_bdr = 1 if first_aml else 0 alignment = 'right' if 'o_right_alignment' in col.get('class', '') else 'left' if row.get(col['label'], {}).get('data'): val = row[col['label']]['data'] is_dt = isinstance(val, datetime.date) is_bold = False if row[col['label']].get('class') and 'o_bold' in row[col['label']]['class']: first_aml = True top_bdr = 1 is_bold = True self._write_cell( cx, cy, val, 1, is_dt, report, font_cache, wb, ws, XLSX_FONT_SIZE_DEFAULT, is_bold, 'white', alignment, 0, top_bdr, XLSX_BORDER_COLOR, ) else: self._write_cell( cx, cy, '', 1, False, report, font_cache, wb, ws, XLSX_FONT_SIZE_DEFAULT, False, 'white', alignment, 0, top_bdr, XLSX_BORDER_COLOR, ) cx += 1 cx = 0 cy += 1 # Total row total_row = jv['lines'][-1] for col in cols: val = total_row.get(col['label'], {}).get('data', '') alignment = 'right' if 'o_right_alignment' in col.get('class', '') else 'left' self._write_cell( cx, cy, val, 1, False, report, font_cache, wb, ws, XLSX_FONT_SIZE_DEFAULT, True, XLSX_GRAY_200, alignment, 2, 2, ) cx += 1 cx = 0 ws.set_default_row(20) ws.set_row(0, 30) if jv.get('tax_summary'): self._write_tax_summaries_to_sheet( report, wb, ws, font_cache, len(cols) + 1, 1, jv['tax_summary'], ) if doc_data.get('global_tax_summary'): self._write_tax_summaries_to_sheet( report, wb, wb.add_worksheet(_('Global Tax Summary')[:31]), font_cache, 0, 0, doc_data['global_tax_summary'], ) wb.close() wb_buffer.seek(0) xlsx_bytes = wb_buffer.read() wb_buffer.close() return { 'file_name': report.get_default_report_filename(options, 'xlsx'), 'file_content': xlsx_bytes, 'file_type': 'xlsx', } def _write_cell( self, x, y, value, colspan, is_datetime, report, fonts, workbook, sheet, font_size, bold=False, bg_color='white', align='left', border_bottom=0, border_top=0, border_color='0x000000', ): """Write a styled value to the specified worksheet cell.""" fmt = workbook.add_format({ 'font_name': 'Arial', 'font_size': font_size, 'bold': bold, 'bg_color': bg_color, 'align': align, 'bottom': border_bottom, 'top': border_top, 'border_color': border_color, }) if colspan == 1: if is_datetime: fmt.set_num_format('yyyy-mm-dd') sheet.write_datetime(y, x, value, fmt) else: if isinstance(value, str): value = value.replace('\n', ' ') report._set_xlsx_cell_sizes(sheet, fonts[font_size], x, y, value, fmt, colspan > 1) sheet.write(y, x, value, fmt) else: sheet.merge_range(y, x, y, x + colspan - 1, value, fmt) def _write_tax_summaries_to_sheet(self, report, workbook, sheet, fonts, start_x, start_y, tax_summary): cx, cy = start_x, start_y taxes = tax_summary.get('tax_report_lines') if taxes: ar_start = start_x + 1 cols = [] if len(taxes) > 1: ar_start += 1 cols.append(_('Country')) cols += [_('Name'), _('Base Amount'), _('Tax Amount')] if tax_summary.get('tax_non_deductible_column'): cols.append(_('Non-Deductible')) if tax_summary.get('tax_deductible_column'): cols.append(_('Deductible')) if tax_summary.get('tax_due_column'): cols.append(_('Due')) self._write_cell(cx, cy, _('Taxes Applied'), len(cols), False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_HEADING, True, 'white', 'left', 2) cy += 1 for c in cols: a = 'right' if cx >= ar_start else 'left' self._write_cell(cx, cy, c, 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, True, XLSX_GRAY_200, a, 2) cx += 1 cx = start_x cy += 1 for country in taxes: first_country_line = True for tax in taxes[country]: if len(taxes) > 1: if first_country_line: first_country_line = False self._write_cell(cx, cy, country, 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, True, 'white', 'left', 1, 0, XLSX_BORDER_COLOR) cx += 1 self._write_cell(cx, cy, tax['name'], 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, True, 'white', 'left', 1, 0, XLSX_BORDER_COLOR) self._write_cell(cx+1, cy, tax['base_amount'], 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) self._write_cell(cx+2, cy, tax['tax_amount'], 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) cx += 3 if tax_summary.get('tax_non_deductible_column'): self._write_cell(cx, cy, tax['tax_non_deductible'], 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) cx += 1 if tax_summary.get('tax_deductible_column'): self._write_cell(cx, cy, tax['tax_deductible'], 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) cx += 1 if tax_summary.get('tax_due_column'): self._write_cell(cx, cy, tax['tax_due'], 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) cx = start_x cy += 1 cx = start_x cy += 2 grids = tax_summary.get('tax_grid_summary_lines') if grids: ar_start = start_x + 1 gcols = [] if len(grids) > 1: ar_start += 1 gcols.append(_('Country')) gcols += [_('Grid'), _('+'), _('-'), _('Impact On Grid')] self._write_cell(cx, cy, _('Impact On Grid'), len(gcols), False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_HEADING, True, 'white', 'left', 2) cy += 1 for c in gcols: a = 'right' if cx >= ar_start else 'left' self._write_cell(cx, cy, c, 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, True, XLSX_GRAY_200, a, 2) cx += 1 cx = start_x cy += 1 for country in grids: first_line = True for grid_name in grids[country]: if len(grids) > 1: if first_line: first_line = False self._write_cell(cx, cy, country, 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, True, 'white', 'left', 1, 0, XLSX_BORDER_COLOR) cx += 1 self._write_cell(cx, cy, grid_name, 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, True, 'white', 'left', 1, 0, XLSX_BORDER_COLOR) self._write_cell(cx+1, cy, grids[country][grid_name].get('+', 0), 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) self._write_cell(cx+2, cy, grids[country][grid_name].get('-', 0), 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) self._write_cell(cx+3, cy, grids[country][grid_name]['impact'], 1, False, report, fonts, workbook, sheet, XLSX_FONT_SIZE_DEFAULT, False, 'white', 'right', 1, 0, XLSX_BORDER_COLOR) cx = start_x cy += 1 # ================================================================ # DOCUMENT DATA GENERATION # ================================================================ def _generate_document_data_for_export(self, report, options, export_type='pdf'): """Produce all data needed for journal report export (PDF or XLSX).""" self.env.flush_all() qry = report._get_report_query(options, 'strict_range') acct_alias = qry.join( lhs_alias='account_move_line', lhs_column='account_id', rhs_table='account_account', rhs_column='id', link='account_id', ) acct_code = self.env['account.account']._field_to_sql(acct_alias, 'code', qry) acct_name = self.env['account.account']._field_to_sql(acct_alias, 'name') stmt = SQL( """ SELECT account_move_line.id AS move_line_id, account_move_line.name, account_move_line.date, account_move_line.invoice_date, account_move_line.amount_currency, account_move_line.tax_base_amount, account_move_line.currency_id AS move_line_currency, am.id AS move_id, am.name AS move_name, am.journal_id, am.currency_id AS move_currency, am.amount_total_in_currency_signed AS amount_currency_total, am.currency_id != cp.currency_id AS is_multicurrency, p.name AS partner_name, %(acct_code)s AS account_code, %(acct_name)s AS account_name, %(acct_alias)s.account_type AS account_type, COALESCE(account_move_line.debit, 0) AS debit, COALESCE(account_move_line.credit, 0) AS credit, COALESCE(account_move_line.balance, 0) AS balance, %(j_name)s AS journal_name, j.code AS journal_code, j.type AS journal_type, cp.currency_id AS company_currency, CASE WHEN j.type = 'sale' THEN am.payment_reference WHEN j.type = 'purchase' THEN am.ref END AS reference, array_remove(array_agg(DISTINCT %(tax_name)s), NULL) AS taxes, array_remove(array_agg(DISTINCT %(tag_name)s), NULL) AS tax_grids FROM %(tbl)s JOIN account_move am ON am.id = account_move_line.move_id LEFT JOIN res_partner p ON p.id = account_move_line.partner_id JOIN account_journal j ON j.id = am.journal_id JOIN res_company cp ON cp.id = am.company_id LEFT JOIN account_move_line_account_tax_rel aml_at_rel ON aml_at_rel.account_move_line_id = account_move_line.id LEFT JOIN account_tax parent_tax ON parent_tax.id = aml_at_rel.account_tax_id and parent_tax.amount_type = 'group' LEFT JOIN account_tax_filiation_rel tax_filiation_rel ON tax_filiation_rel.parent_tax = parent_tax.id LEFT JOIN account_tax tax ON (tax.id = aml_at_rel.account_tax_id and tax.amount_type != 'group') or tax.id = tax_filiation_rel.child_tax LEFT JOIN account_account_tag_account_move_line_rel tag_rel ON tag_rel.account_move_line_id = account_move_line.id LEFT JOIN account_account_tag tag ON tag_rel.account_account_tag_id = tag.id LEFT JOIN res_currency journal_curr ON journal_curr.id = j.currency_id WHERE %(pmt_filter)s AND %(where)s GROUP BY "account_move_line".id, am.id, p.id, %(acct_alias)s.id, j.id, cp.id, journal_curr.id, account_code, account_name ORDER BY CASE j.type WHEN 'sale' THEN 1 WHEN 'purchase' THEN 2 WHEN 'general' THEN 3 WHEN 'bank' THEN 4 ELSE 5 END, j.sequence, CASE WHEN am.name = '/' THEN 1 ELSE 0 END, am.date, am.name, CASE %(acct_alias)s.account_type WHEN 'liability_payable' THEN 1 WHEN 'asset_receivable' THEN 1 WHEN 'liability_credit_card' THEN 5 WHEN 'asset_cash' THEN 5 ELSE 2 END, account_move_line.tax_line_id NULLS FIRST """, tbl=qry.from_clause, pmt_filter=self._get_payment_lines_filter_case_statement(options), where=qry.where_clause, acct_code=acct_code, acct_name=acct_name, acct_alias=SQL.identifier(acct_alias), j_name=self.env['account.journal']._field_to_sql('j', 'name'), tax_name=self.env['account.tax']._field_to_sql('tax', 'name'), tag_name=self.env['account.account.tag']._field_to_sql('tag', 'name'), ) self.env.cr.execute(stmt) by_journal = {} for row in self.env.cr.dictfetchall(): by_journal.setdefault(row['journal_id'], {}).setdefault(row['move_id'], []).append(row) journals_vals = [] any_has_taxes = False for jnl_moves in by_journal.values(): move_lists = list(jnl_moves.values()) first = move_lists[0][0] jv = { 'id': first['journal_id'], 'name': first['journal_name'], 'code': first['journal_code'], 'type': first['journal_type'], } if self._section_has_tax(options, jv['id']): jv['tax_summary'] = self._get_tax_summary_section(options, jv) any_has_taxes = True jv['lines'] = self._get_export_lines_for_journal(report, options, export_type, jv, move_lists) jv['columns'] = self._get_columns_for_journal(jv, export_type) journals_vals.append(jv) return { 'journals_vals': journals_vals, 'global_tax_summary': self._get_tax_summary_section(options) if any_has_taxes else False, } def _get_columns_for_journal(self, journal, export_type='pdf'): cols = [{'name': _('Document'), 'label': 'document'}] if export_type == 'pdf': cols.append({'name': _('Account'), 'label': 'account_label'}) else: cols.extend([ {'name': _('Account Code'), 'label': 'account_code'}, {'name': _('Account Label'), 'label': 'account_label'}, ]) cols.extend([ {'name': _('Name'), 'label': 'name'}, {'name': _('Debit'), 'label': 'debit', 'class': 'o_right_alignment '}, {'name': _('Credit'), 'label': 'credit', 'class': 'o_right_alignment '}, ]) if journal.get('tax_summary'): cols.append({'name': _('Taxes'), 'label': 'taxes'}) if journal['tax_summary'].get('tax_grid_summary_lines'): cols.append({'name': _('Tax Grids'), 'label': 'tax_grids'}) if journal['type'] == 'bank': cols.append({'name': _('Balance'), 'label': 'balance', 'class': 'o_right_alignment '}) if journal.get('multicurrency_column'): cols.append({'name': _('Amount Currency'), 'label': 'amount_currency', 'class': 'o_right_alignment '}) return cols def _get_export_lines_for_journal(self, report, options, export_type, journal_vals, move_lists): if journal_vals['type'] == 'bank': return self._get_export_lines_for_bank_journal(report, options, export_type, journal_vals, move_lists) sum_cr, sum_dr = 0, 0 rows = [] for i, aml_list in enumerate(move_lists): for j, entry in enumerate(aml_list): doc = False if j == 0: doc = entry['move_name'] elif j == 1: doc = entry['date'] row = self._get_base_line(report, options, export_type, doc, entry, j, i % 2 != 0, journal_vals.get('tax_summary')) sum_cr += entry['credit'] sum_dr += entry['debit'] rows.append(row) first_entry = aml_list[0] if first_entry['is_multicurrency']: mc_label = _('Amount in currency: %s', report._format_value(options, first_entry['amount_currency_total'], 'monetary', format_params={'currency_id': first_entry['move_currency']})) if len(aml_list) <= 2: rows.append({'document': {'data': mc_label}, 'line_class': 'o_even ' if i % 2 == 0 else 'o_odd ', 'amount': {'data': first_entry['amount_currency_total']}, 'currency_id': {'data': first_entry['move_currency']}}) else: rows[-1]['document'] = {'data': mc_label} rows[-1]['amount'] = {'data': first_entry['amount_currency_total']} rows[-1]['currency_id'] = {'data': first_entry['move_currency']} rows.append({}) rows.append({ 'name': {'data': _('Total')}, 'debit': {'data': report._format_value(options, sum_dr, 'monetary')}, 'credit': {'data': report._format_value(options, sum_cr, 'monetary')}, }) return rows def _get_export_lines_for_bank_journal(self, report, options, export_type, journal_vals, move_lists): rows = [] running_balance = self._query_bank_journal_initial_balance(options, journal_vals['id']) rows.append({'name': {'data': _('Starting Balance')}, 'balance': {'data': report._format_value(options, running_balance, 'monetary')}}) sum_cr, sum_dr = 0, 0 for i, aml_list in enumerate(move_lists): is_unreconciled = not any(ln for ln in aml_list if ln['account_type'] in ('liability_credit_card', 'asset_cash')) for j, entry in enumerate(aml_list): if entry['account_type'] not in ('liability_credit_card', 'asset_cash'): doc = '' if j == 0: doc = f'{entry["move_name"]} ({entry["date"]})' row = self._get_base_line(report, options, export_type, doc, entry, j, i % 2 != 0, journal_vals.get('tax_summary')) sum_cr += entry['credit'] sum_dr += entry['debit'] if not is_unreconciled: line_bal = -entry['balance'] running_balance += line_bal row['balance'] = { 'data': report._format_value(options, running_balance, 'monetary'), 'class': 'o_muted ' if self.env.company.currency_id.is_zero(line_bal) else '', } if self.env.user.has_group('base.group_multi_currency') and entry['move_line_currency'] != entry['company_currency']: journal_vals['multicurrency_column'] = True mc_amt = -entry['amount_currency'] if not is_unreconciled else entry['amount_currency'] mc_cur = self.env['res.currency'].browse(entry['move_line_currency']) row['amount_currency'] = { 'data': report._format_value(options, mc_amt, 'monetary', format_params={'currency_id': mc_cur.id}), 'class': 'o_muted ' if mc_cur.is_zero(mc_amt) else '', } rows.append(row) rows.append({}) rows.append({'name': {'data': _('Total')}, 'balance': {'data': report._format_value(options, running_balance, 'monetary')}}) return rows def _get_base_line(self, report, options, export_type, document, entry, line_idx, is_even, has_taxes): co_cur = self.env.company.currency_id label = entry['name'] or entry['reference'] acct_label = entry['partner_name'] or entry['account_name'] if entry['partner_name'] and entry['account_type'] == 'asset_receivable': fmt_label = _('AR %s', acct_label) elif entry['partner_name'] and entry['account_type'] == 'liability_payable': fmt_label = _('AP %s', acct_label) else: acct_label = entry['account_name'] fmt_label = _('G %s', entry["account_code"]) row = { 'line_class': 'o_even ' if is_even else 'o_odd ', 'document': {'data': document, 'class': 'o_bold ' if line_idx == 0 else ''}, 'account_code': {'data': entry['account_code']}, 'account_label': {'data': acct_label if export_type != 'pdf' else fmt_label}, 'name': {'data': label}, 'debit': {'data': report._format_value(options, entry['debit'], 'monetary'), 'class': 'o_muted ' if co_cur.is_zero(entry['debit']) else ''}, 'credit': {'data': report._format_value(options, entry['credit'], 'monetary'), 'class': 'o_muted ' if co_cur.is_zero(entry['credit']) else ''}, } if has_taxes: tax_display = '' if entry['taxes']: tax_display = _('T: %s', ', '.join(entry['taxes'])) elif entry['tax_base_amount'] is not None: tax_display = _('B: %s', report._format_value(options, entry['tax_base_amount'], 'monetary')) row['taxes'] = {'data': tax_display} row['tax_grids'] = {'data': ', '.join(entry['tax_grids'])} return row # ================================================================ # QUERY HELPERS # ================================================================ def _get_payment_lines_filter_case_statement(self, options): if not options.get('show_payment_lines'): return SQL(""" (j.type != 'bank' OR EXISTS( SELECT 1 FROM account_move_line JOIN account_account acc ON acc.id = account_move_line.account_id WHERE account_move_line.move_id = am.id AND acc.account_type IN ('liability_credit_card', 'asset_cash') )) """) return SQL('TRUE') def _query_bank_journal_initial_balance(self, options, journal_id): report = self.env.ref('fusion_accounting.journal_report') qry = report._get_report_query(options, 'to_beginning_of_period', domain=[('journal_id', '=', journal_id)]) stmt = SQL(""" SELECT COALESCE(SUM(account_move_line.balance), 0) AS balance FROM %(tbl)s JOIN account_journal journal ON journal.id = "account_move_line".journal_id AND account_move_line.account_id = journal.default_account_id WHERE %(where)s GROUP BY journal.id """, tbl=qry.from_clause, where=qry.where_clause) self.env.cr.execute(stmt) rows = self.env.cr.dictfetchall() return rows[0]['balance'] if rows else 0 # ================================================================ # TAX SUMMARIES # ================================================================ def _section_has_tax(self, options, journal_id): report = self.env['account.report'].browse(options.get('report_id')) domain = [('tax_ids', '!=', False)] if journal_id: domain.append(('journal_id', '=', journal_id)) domain += report._get_options_domain(options, 'strict_range') return bool(self.env['account.move.line'].search_count(domain, limit=1)) def _get_tax_summary_section(self, options, journal_vals=None): td = { 'date_from': options.get('date', {}).get('date_from'), 'date_to': options.get('date', {}).get('date_to'), } if journal_vals: td['journal_id'] = journal_vals['id'] td['journal_type'] = journal_vals['type'] tax_lines = self._get_generic_tax_summary_for_sections(options, td) nd_col = any(ln.get('tax_non_deductible_no_format') for vals in tax_lines.values() for ln in vals) ded_col = any(ln.get('tax_deductible_no_format') for vals in tax_lines.values() for ln in vals) due_col = any(ln.get('tax_due_no_format') for vals in tax_lines.values() for ln in vals) return { 'tax_report_lines': tax_lines, 'tax_non_deductible_column': nd_col, 'tax_deductible_column': ded_col, 'tax_due_column': due_col, 'extra_columns': int(nd_col) + int(ded_col) + int(due_col), 'tax_grid_summary_lines': self._get_tax_grids_summary(options, td), } def _get_generic_tax_report_options(self, options, data): generic_rpt = self.env.ref('account.generic_tax_report') prev = options.copy() prev.update({ 'selected_variant_id': generic_rpt.id, 'date_from': data.get('date_from'), 'date_to': data.get('date_to'), }) tax_opts = generic_rpt.get_options(prev) jnl_rpt = self.env['account.report'].browse(options['report_id']) tax_opts['forced_domain'] = tax_opts.get('forced_domain', []) + jnl_rpt._get_options_domain(options, 'strict_range') if data.get('journal_id') or data.get('journal_type'): tax_opts['journals'] = [{ 'id': data.get('journal_id'), 'model': 'account.journal', 'type': data.get('journal_type'), 'selected': True, }] return tax_opts def _get_tax_grids_summary(self, options, data): report = self.env.ref('fusion_accounting.journal_report') tax_opts = self._get_generic_tax_report_options(options, data) qry = report._get_report_query(tax_opts, 'strict_range') country_nm = self.env['res.country']._field_to_sql('country', 'name') tag_nm = self.env['account.account.tag']._field_to_sql('tag', 'name') stmt = SQL(""" WITH tag_info (country_name, tag_id, tag_name, tag_sign, balance) AS ( SELECT %(cn)s AS country_name, tag.id, %(tn)s AS name, CASE WHEN tag.tax_negate IS TRUE THEN '-' ELSE '+' END, SUM(COALESCE("account_move_line".balance, 0) * CASE WHEN "account_move_line".tax_tag_invert THEN -1 ELSE 1 END) AS balance FROM account_account_tag tag JOIN account_account_tag_account_move_line_rel rel ON tag.id = rel.account_account_tag_id JOIN res_country country ON country.id = tag.country_id , %(tbl)s WHERE %(where)s AND applicability = 'taxes' AND "account_move_line".id = rel.account_move_line_id GROUP BY country_name, tag.id ) SELECT country_name, tag_id, REGEXP_REPLACE(tag_name, '^[+-]', '') AS name, balance, tag_sign AS sign FROM tag_info ORDER BY country_name, name """, cn=country_nm, tn=tag_nm, tbl=qry.from_clause, where=qry.where_clause) self.env.cr.execute(stmt) rows = self.env.cr.fetchall() result = {} opp = {'+': '-', '-': '+'} for cname, _tid, gname, bal, sign in rows: result.setdefault(cname, {}).setdefault(gname, {}) result[cname][gname].setdefault('tag_ids', []).append(_tid) result[cname][gname][sign] = report._format_value(options, bal, 'monetary') if opp[sign] not in result[cname][gname]: result[cname][gname][opp[sign]] = report._format_value(options, 0, 'monetary') result[cname][gname][sign + '_no_format'] = bal result[cname][gname]['impact'] = report._format_value(options, result[cname][gname].get('+_no_format', 0) - result[cname][gname].get('-_no_format', 0), 'monetary') return result def _get_generic_tax_summary_for_sections(self, options, data): report = self.env['account.report'].browse(options['report_id']) tax_opts = self._get_generic_tax_report_options(options, data) tax_opts['account_journal_report_tax_deductibility_columns'] = True tax_rpt = self.env.ref('account.generic_tax_report') rpt_lines = tax_rpt._get_lines(tax_opts) tax_vals = {} for rln in rpt_lines: model, lid = report._parse_line_id(rln.get('id'))[-1][1:] if model == 'account.tax': tax_vals[lid] = { 'base_amount': rln['columns'][0]['no_format'], 'tax_amount': rln['columns'][1]['no_format'], 'tax_non_deductible': rln['columns'][2]['no_format'], 'tax_deductible': rln['columns'][3]['no_format'], 'tax_due': rln['columns'][4]['no_format'], } taxes = self.env['account.tax'].browse(tax_vals.keys()) grouped = {} for tx in taxes: grouped.setdefault(tx.country_id.name, []).append({ 'base_amount': report._format_value(options, tax_vals[tx.id]['base_amount'], 'monetary'), 'tax_amount': report._format_value(options, tax_vals[tx.id]['tax_amount'], 'monetary'), 'tax_non_deductible': report._format_value(options, tax_vals[tx.id]['tax_non_deductible'], 'monetary'), 'tax_non_deductible_no_format': tax_vals[tx.id]['tax_non_deductible'], 'tax_deductible': report._format_value(options, tax_vals[tx.id]['tax_deductible'], 'monetary'), 'tax_deductible_no_format': tax_vals[tx.id]['tax_deductible'], 'tax_due': report._format_value(options, tax_vals[tx.id]['tax_due'], 'monetary'), 'tax_due_no_format': tax_vals[tx.id]['tax_due'], 'name': tx.name, 'line_id': report._get_generic_line_id('account.tax', tx.id), }) return dict(sorted(grouped.items())) # ================================================================ # ACTIONS # ================================================================ def journal_report_tax_tag_template_open_aml(self, options, params=None): tag_ids = params.get('tag_ids') domain = ( self.env['account.report'].browse(options['report_id'])._get_options_domain(options, 'strict_range') + [('tax_tag_ids', 'in', [tag_ids])] + self.env['account.move.line']._get_tax_exigible_domain() ) return { 'type': 'ir.actions.act_window', 'name': _('Journal Items for Tax Audit'), 'res_model': 'account.move.line', 'views': [[self.env.ref('account.view_move_line_tax_audit_tree').id, 'list']], 'domain': domain, 'context': self.env.context, } def journal_report_action_dropdown_audit_default_tax_report(self, options, params): return self.env['account.generic.tax.report.handler'].caret_option_audit_tax(options, params) def journal_report_action_open_tax_journal_items(self, options, params): ctx = { 'search_default_posted': 0 if options.get('all_entries') else 1, 'search_default_date_between': 1, 'date_from': params and params.get('date_from') or options.get('date', {}).get('date_from'), 'date_to': params and params.get('date_to') or options.get('date', {}).get('date_to'), 'search_default_journal_id': params.get('journal_id'), 'expand': 1, } if params and params.get('tax_type') == 'tag': ctx.update({'search_default_group_by_tax_tags': 1, 'search_default_group_by_account': 2}) elif params and params.get('tax_type') == 'tax': ctx.update({'search_default_group_by_taxes': 1, 'search_default_group_by_account': 2}) if params and 'journal_id' in params: ctx['search_default_journal_id'] = [params['journal_id']] if options and options.get('journals') and not ctx.get('search_default_journal_id'): sel = [j['id'] for j in options['journals'] if j.get('selected') and j['model'] == 'account.journal'] if len(sel) == 1: ctx['search_default_journal_id'] = sel return { 'name': params.get('name'), 'view_mode': 'list,pivot,graph,kanban', 'res_model': 'account.move.line', 'views': [(self.env.ref('account.view_move_line_tree').id, 'list')], 'type': 'ir.actions.act_window', 'domain': [('display_type', 'not in', ('line_section', 'line_note'))], 'context': ctx, } def journal_report_action_open_account_move_lines_by_account(self, options, params): report = self.env['account.report'].browse(options['report_id']) jnl = self.env['account.journal'].browse(params['journal_id']) acct = self.env['account.account'].browse(params['account_id']) domain = [('journal_id.id', '=', jnl.id), ('account_id.id', '=', acct.id)] domain += report._get_options_domain(options, 'strict_range') return { 'type': 'ir.actions.act_window', 'name': _("%(journal)s - %(account)s", journal=jnl.name, account=acct.name), 'res_model': 'account.move.line', 'views': [[False, 'list']], 'domain': domain, } def journal_report_open_aml_by_move(self, options, params): report = self.env['account.report'].browse(options['report_id']) jnl = self.env['account.journal'].browse(params['journal_id']) ctx_extra = {'search_default_group_by_account': 0, 'show_more_partner_info': 1} if jnl.type in ('bank', 'credit'): params['view_ref'] = 'fusion_accounting.view_journal_report_audit_bank_move_line_tree' ctx_extra['search_default_exclude_bank_lines'] = 1 else: params['view_ref'] = 'fusion_accounting.view_journal_report_audit_move_line_tree' ctx_extra.update({'search_default_group_by_partner': 1, 'search_default_group_by_move': 2}) if jnl.type in ('sale', 'purchase'): ctx_extra['search_default_invoices_lines'] = 1 action = report.open_journal_items(options=options, params=params) action.get('context', {}).update(ctx_extra) return action