# Fusion Accounting - General Ledger Report Handler import json from odoo import models, fields, api, _ from odoo.tools.misc import format_date from odoo.tools import get_lang, SQL from odoo.exceptions import UserError from datetime import timedelta from collections import defaultdict class GeneralLedgerCustomHandler(models.AbstractModel): """Produces the General Ledger report. Aggregates journal items by account and period, handles initial balances, unaffected-earnings allocation, and optional tax-declaration sections. """ _name = 'account.general.ledger.report.handler' _inherit = 'account.report.custom.handler' _description = 'General Ledger Custom Handler' # ------------------------------------------------------------------ # Display configuration # ------------------------------------------------------------------ def _get_custom_display_config(self): return { 'templates': { 'AccountReportLineName': 'fusion_accounting.GeneralLedgerLineName', }, } # ------------------------------------------------------------------ # Options # ------------------------------------------------------------------ def _custom_options_initializer(self, report, options, previous_options): """Strip the multi-currency column when the user lacks the group, and auto-unfold when printing.""" super()._custom_options_initializer(report, options, previous_options=previous_options) if self.env.user.has_group('base.group_multi_currency'): options['multi_currency'] = True else: options['columns'] = [ c for c in options['columns'] if c['expression_label'] != 'amount_currency' ] # When printing the whole report, unfold everything unless the user # explicitly selected specific lines. options['unfold_all'] = ( (options['export_mode'] == 'print' and not options.get('unfolded_lines')) or options['unfold_all'] ) # ------------------------------------------------------------------ # Dynamic lines # ------------------------------------------------------------------ def _dynamic_lines_generator(self, report, options, all_column_groups_expression_totals, warnings=None): """Return ``[(seq, line_dict), ...]`` for every account row plus an optional tax-declaration block and a grand-total row.""" result_lines = [] period_start = fields.Date.from_string(options['date']['date_from']) comp_currency = self.env.company.currency_id running_totals = defaultdict(lambda: {'debit': 0, 'credit': 0, 'balance': 0}) for account_rec, col_grp_vals in self._aggregate_account_values(report, options): per_col = {} any_current = False for col_key, bucket in col_grp_vals.items(): main = bucket.get('sum', {}) unaff = bucket.get('unaffected_earnings', {}) dr = main.get('debit', 0.0) + unaff.get('debit', 0.0) cr = main.get('credit', 0.0) + unaff.get('credit', 0.0) bal = main.get('balance', 0.0) + unaff.get('balance', 0.0) per_col[col_key] = { 'amount_currency': main.get('amount_currency', 0.0) + unaff.get('amount_currency', 0.0), 'debit': dr, 'credit': cr, 'balance': bal, } latest_date = main.get('max_date') if latest_date and latest_date >= period_start: any_current = True running_totals[col_key]['debit'] += dr running_totals[col_key]['credit'] += cr running_totals[col_key]['balance'] += bal result_lines.append( self._build_account_header_line(report, options, account_rec, any_current, per_col) ) # Round the accumulated balance for totals in running_totals.values(): totals['balance'] = comp_currency.round(totals['balance']) # Tax-declaration section (single column group + single journal of sale/purchase type) active_journals = report._get_options_journals(options) if ( len(options['column_groups']) == 1 and len(active_journals) == 1 and active_journals[0]['type'] in ('sale', 'purchase') ): result_lines += self._produce_tax_declaration_lines( report, options, active_journals[0]['type'] ) # Grand total result_lines.append(self._build_grand_total_line(report, options, running_totals)) return [(0, ln) for ln in result_lines] # ------------------------------------------------------------------ # Batch unfold helper # ------------------------------------------------------------------ def _custom_unfold_all_batch_data_generator(self, report, options, lines_to_expand_by_function): """Pre-load data for all accounts that need unfolding so the engine does not issue per-account queries.""" target_acct_ids = [] for line_info in lines_to_expand_by_function.get('_report_expand_unfoldable_line_general_ledger', []): mdl, mdl_id = report._get_model_info_from_id(line_info['id']) if mdl == 'account.account': target_acct_ids.append(mdl_id) page_size = report.load_more_limit if report.load_more_limit and not options.get('export_mode') else None overflow_flags = {} full_aml_data = self._fetch_aml_data(report, options, target_acct_ids)[0] if page_size: trimmed_aml_data = {} for acct_id, acct_rows in full_aml_data.items(): page = {} for key, val in acct_rows.items(): if len(page) >= page_size: overflow_flags[acct_id] = True break page[key] = val trimmed_aml_data[acct_id] = page else: trimmed_aml_data = full_aml_data return { 'initial_balances': self._fetch_opening_balances(report, target_acct_ids, options), 'aml_results': trimmed_aml_data, 'has_more': overflow_flags, } # ------------------------------------------------------------------ # Tax declaration # ------------------------------------------------------------------ def _produce_tax_declaration_lines(self, report, options, tax_type): """Append a Tax Declaration section when viewing a single sale / purchase journal.""" header_labels = { 'debit': _("Base Amount"), 'credit': _("Tax Amount"), } output = [ { 'id': report._get_generic_line_id(None, None, markup='tax_decl_header_1'), 'name': _('Tax Declaration'), 'columns': [{} for _ in options['columns']], 'level': 1, 'unfoldable': False, 'unfolded': False, }, { 'id': report._get_generic_line_id(None, None, markup='tax_decl_header_2'), 'name': _('Name'), 'columns': [ {'name': header_labels.get(c['expression_label'], '')} for c in options['columns'] ], 'level': 3, 'unfoldable': False, 'unfolded': False, }, ] tax_report = self.env.ref('account.generic_tax_report') tax_opts = tax_report.get_options({ **options, 'selected_variant_id': tax_report.id, 'forced_domain': [('tax_line_id.type_tax_use', '=', tax_type)], }) tax_lines = tax_report._get_lines(tax_opts) parent_marker = tax_report._get_generic_line_id(None, None, markup=tax_type) for tl in tax_lines: if tl.get('parent_id') != parent_marker: continue src_cols = tl['columns'] mapped = { 'debit': src_cols[0], 'credit': src_cols[1], } tl['columns'] = [mapped.get(c['expression_label'], {}) for c in options['columns']] output.append(tl) return output # ------------------------------------------------------------------ # Core queries # ------------------------------------------------------------------ def _aggregate_account_values(self, report, options): """Execute summary queries and assign unaffected-earnings. Returns ``[(account_record, {col_group_key: {...}, ...}), ...]`` """ combined_sql = self._build_summary_query(report, options) if not combined_sql: return [] by_account = {} by_company = {} self.env.cr.execute(combined_sql) for row in self.env.cr.dictfetchall(): if row['groupby'] is None: continue cg = row['column_group_key'] bucket = row['key'] if bucket == 'sum': by_account.setdefault(row['groupby'], {k: {} for k in options['column_groups']}) by_account[row['groupby']][cg][bucket] = row elif bucket == 'initial_balance': by_account.setdefault(row['groupby'], {k: {} for k in options['column_groups']}) by_account[row['groupby']][cg][bucket] = row elif bucket == 'unaffected_earnings': by_company.setdefault(row['groupby'], {k: {} for k in options['column_groups']}) by_company[row['groupby']][cg] = row # Assign unaffected earnings to the equity_unaffected account if by_company: candidate_accounts = self.env['account.account'].search([ ('display_name', 'ilike', options.get('filter_search_bar')), *self.env['account.account']._check_company_domain(list(by_company.keys())), ('account_type', '=', 'equity_unaffected'), ]) for comp_id, comp_data in by_company.items(): target_acct = candidate_accounts.filtered( lambda a: self.env['res.company'].browse(comp_id).root_id in a.company_ids ) if not target_acct: continue for cg in options['column_groups']: by_account.setdefault( target_acct.id, {k: {'unaffected_earnings': {}} for k in options['column_groups']}, ) unaff = comp_data.get(cg) if not unaff: continue existing = by_account[target_acct.id][cg].get('unaffected_earnings') if existing: for fld in ('amount_currency', 'debit', 'credit', 'balance'): existing[fld] = existing.get(fld, 0.0) + unaff[fld] else: by_account[target_acct.id][cg]['unaffected_earnings'] = unaff if by_account: accounts = self.env['account.account'].search([('id', 'in', list(by_account.keys()))]) else: accounts = self.env['account.account'] return [(acct, by_account[acct.id]) for acct in accounts] def _build_summary_query(self, report, options) -> SQL: """Construct the UNION ALL query that retrieves period sums and unaffected-earnings sums for every account.""" per_col = report._split_options_per_column_group(options) parts = [] for col_key, grp_opts in per_col.items(): # Decide date scope scope = 'strict_range' if grp_opts.get('general_ledger_strict_range') else 'from_beginning' domain_extra = [] if not grp_opts.get('general_ledger_strict_range'): fy_start = fields.Date.from_string(grp_opts['date']['date_from']) fy_dates = self.env.company.compute_fiscalyear_dates(fy_start) domain_extra += [ '|', ('date', '>=', fy_dates['date_from']), ('account_id.include_initial_balance', '=', True), ] if grp_opts.get('export_mode') == 'print' and grp_opts.get('filter_search_bar'): domain_extra.append(('account_id', 'ilike', grp_opts['filter_search_bar'])) if grp_opts.get('include_current_year_in_unaff_earnings'): domain_extra += [('account_id.include_initial_balance', '=', True)] qry = report._get_report_query(grp_opts, scope, domain=domain_extra) parts.append(SQL( """ SELECT account_move_line.account_id AS groupby, 'sum' AS key, MAX(account_move_line.date) AS max_date, %(col_key)s AS column_group_key, COALESCE(SUM(account_move_line.amount_currency), 0.0) AS amount_currency, SUM(%(dr)s) AS debit, SUM(%(cr)s) AS credit, SUM(%(bal)s) AS balance FROM %(tbl)s %(fx)s WHERE %(cond)s GROUP BY account_move_line.account_id """, col_key=col_key, tbl=qry.from_clause, dr=report._currency_table_apply_rate(SQL("account_move_line.debit")), cr=report._currency_table_apply_rate(SQL("account_move_line.credit")), bal=report._currency_table_apply_rate(SQL("account_move_line.balance")), fx=report._currency_table_aml_join(grp_opts), cond=qry.where_clause, )) # Unaffected earnings sub-query if not grp_opts.get('general_ledger_strict_range'): unaff_opts = self._get_options_unaffected_earnings(grp_opts) unaff_domain = [('account_id.include_initial_balance', '=', False)] unaff_qry = report._get_report_query(unaff_opts, 'strict_range', domain=unaff_domain) parts.append(SQL( """ SELECT account_move_line.company_id AS groupby, 'unaffected_earnings' AS key, NULL AS max_date, %(col_key)s AS column_group_key, COALESCE(SUM(account_move_line.amount_currency), 0.0) AS amount_currency, SUM(%(dr)s) AS debit, SUM(%(cr)s) AS credit, SUM(%(bal)s) AS balance FROM %(tbl)s %(fx)s WHERE %(cond)s GROUP BY account_move_line.company_id """, col_key=col_key, tbl=unaff_qry.from_clause, dr=report._currency_table_apply_rate(SQL("account_move_line.debit")), cr=report._currency_table_apply_rate(SQL("account_move_line.credit")), bal=report._currency_table_apply_rate(SQL("account_move_line.balance")), fx=report._currency_table_aml_join(grp_opts), cond=unaff_qry.where_clause, )) return SQL(" UNION ALL ").join(parts) def _get_options_unaffected_earnings(self, options): """Return modified options for computing prior-year unaffected earnings (P&L accounts before the current fiscal year).""" modified = options.copy() modified.pop('filter_search_bar', None) fy = self.env.company.compute_fiscalyear_dates( fields.Date.from_string(options['date']['date_from']) ) cutoff = ( fields.Date.from_string(modified['date']['date_to']) if options.get('include_current_year_in_unaff_earnings') else fy['date_from'] - timedelta(days=1) ) modified['date'] = self.env['account.report']._get_dates_period(None, cutoff, 'single') return modified # ------------------------------------------------------------------ # AML detail queries # ------------------------------------------------------------------ def _fetch_aml_data(self, report, options, account_ids, offset=0, limit=None): """Load individual move lines for the given accounts. Returns ``({account_id: {(aml_id, date): {col_grp: row}}}, has_more)`` """ container = {aid: {} for aid in account_ids} raw_sql = self._build_aml_query(report, options, account_ids, offset=offset, limit=limit) self.env.cr.execute(raw_sql) row_count = 0 overflow = False for row in self.env.cr.dictfetchall(): row_count += 1 if row_count == limit: overflow = True break # Build a display-friendly communication field if row['ref'] and row['account_type'] != 'asset_receivable': row['communication'] = f"{row['ref']} - {row['name']}" else: row['communication'] = row['name'] composite_key = (row['id'], row['date']) acct_bucket = container[row['account_id']] if composite_key not in acct_bucket: acct_bucket[composite_key] = {cg: {} for cg in options['column_groups']} prior = acct_bucket[composite_key][row['column_group_key']] if prior: prior['debit'] += row['debit'] prior['credit'] += row['credit'] prior['balance'] += row['balance'] prior['amount_currency'] += row['amount_currency'] else: acct_bucket[composite_key][row['column_group_key']] = row return container, overflow def _build_aml_query(self, report, options, account_ids, offset=0, limit=None) -> SQL: """SQL for individual move lines within the strict period range.""" extra_domain = [('account_id', 'in', account_ids)] if account_ids is not None else None fragments = [] journal_label = self.env['account.journal']._field_to_sql('journal', 'name') for col_key, grp_opts in report._split_options_per_column_group(options).items(): qry = report._get_report_query(grp_opts, domain=extra_domain, date_scope='strict_range') acct_a = qry.join( lhs_alias='account_move_line', lhs_column='account_id', rhs_table='account_account', rhs_column='id', link='account_id', ) code_f = self.env['account.account']._field_to_sql(acct_a, 'code', qry) name_f = self.env['account.account']._field_to_sql(acct_a, 'name') type_f = self.env['account.account']._field_to_sql(acct_a, 'account_type') fragments.append(SQL( ''' SELECT account_move_line.id, account_move_line.date, account_move_line.date_maturity, account_move_line.name, account_move_line.ref, account_move_line.company_id, account_move_line.account_id, account_move_line.payment_id, account_move_line.partner_id, account_move_line.currency_id, account_move_line.amount_currency, COALESCE(account_move_line.invoice_date, account_move_line.date) AS invoice_date, account_move_line.date AS date, %(dr)s AS debit, %(cr)s AS credit, %(bal)s AS balance, mv.name AS move_name, co.currency_id AS company_currency_id, prt.name AS partner_name, mv.move_type AS move_type, %(code_f)s AS account_code, %(name_f)s AS account_name, %(type_f)s AS account_type, journal.code AS journal_code, %(journal_label)s AS journal_name, fr.id AS full_rec_name, %(col_key)s AS column_group_key FROM %(tbl)s JOIN account_move mv ON mv.id = account_move_line.move_id %(fx)s LEFT JOIN res_company co ON co.id = account_move_line.company_id LEFT JOIN res_partner prt ON prt.id = account_move_line.partner_id LEFT JOIN account_journal journal ON journal.id = account_move_line.journal_id LEFT JOIN account_full_reconcile fr ON fr.id = account_move_line.full_reconcile_id WHERE %(cond)s ORDER BY account_move_line.date, account_move_line.move_name, account_move_line.id ''', code_f=code_f, name_f=name_f, type_f=type_f, journal_label=journal_label, col_key=col_key, tbl=qry.from_clause, fx=report._currency_table_aml_join(grp_opts), dr=report._currency_table_apply_rate(SQL("account_move_line.debit")), cr=report._currency_table_apply_rate(SQL("account_move_line.credit")), bal=report._currency_table_apply_rate(SQL("account_move_line.balance")), cond=qry.where_clause, )) combined = SQL(" UNION ALL ").join(SQL("(%s)", f) for f in fragments) if offset: combined = SQL('%s OFFSET %s ', combined, offset) if limit: combined = SQL('%s LIMIT %s ', combined, limit) return combined # ------------------------------------------------------------------ # Initial balance # ------------------------------------------------------------------ def _fetch_opening_balances(self, report, account_ids, options): """Compute the opening balance per account at the start of the reporting period.""" parts = [] for col_key, grp_opts in report._split_options_per_column_group(options).items(): init_opts = self._get_options_initial_balance(grp_opts) domain = [('account_id', 'in', account_ids)] if not init_opts.get('general_ledger_strict_range'): domain += [ '|', ('date', '>=', init_opts['date']['date_from']), ('account_id.include_initial_balance', '=', True), ] if init_opts.get('include_current_year_in_unaff_earnings'): domain += [('account_id.include_initial_balance', '=', True)] qry = report._get_report_query(init_opts, 'from_beginning', domain=domain) parts.append(SQL( """ SELECT account_move_line.account_id AS groupby, 'initial_balance' AS key, NULL AS max_date, %(col_key)s AS column_group_key, COALESCE(SUM(account_move_line.amount_currency), 0.0) AS amount_currency, SUM(%(dr)s) AS debit, SUM(%(cr)s) AS credit, SUM(%(bal)s) AS balance FROM %(tbl)s %(fx)s WHERE %(cond)s GROUP BY account_move_line.account_id """, col_key=col_key, tbl=qry.from_clause, dr=report._currency_table_apply_rate(SQL("account_move_line.debit")), cr=report._currency_table_apply_rate(SQL("account_move_line.credit")), bal=report._currency_table_apply_rate(SQL("account_move_line.balance")), fx=report._currency_table_aml_join(grp_opts), cond=qry.where_clause, )) self.env.cr.execute(SQL(" UNION ALL ").join(parts)) init_map = { aid: {cg: {} for cg in options['column_groups']} for aid in account_ids } for row in self.env.cr.dictfetchall(): init_map[row['groupby']][row['column_group_key']] = row accts = self.env['account.account'].browse(account_ids) return {a.id: (a, init_map[a.id]) for a in accts} def _get_options_initial_balance(self, options): """Derive an options dict whose date range ends just before the report's ``date_from``, suitable for computing opening balances.""" derived = options.copy() # End date raw_to = ( derived['comparison']['periods'][-1]['date_from'] if derived.get('comparison', {}).get('periods') else derived['date']['date_from'] ) end_dt = fields.Date.from_string(raw_to) - timedelta(days=1) # Start date: if date_from aligns with a fiscal-year boundary take the # previous FY; otherwise use the current FY start. start_dt = fields.Date.from_string(derived['date']['date_from']) fy = self.env.company.compute_fiscalyear_dates(start_dt) if start_dt == fy['date_from']: prev_fy = self.env.company.compute_fiscalyear_dates(start_dt - timedelta(days=1)) begin_dt = prev_fy['date_from'] include_curr_yr = True else: begin_dt = fy['date_from'] include_curr_yr = False derived['date'] = self.env['account.report']._get_dates_period(begin_dt, end_dt, 'range') derived['include_current_year_in_unaff_earnings'] = include_curr_yr return derived # ------------------------------------------------------------------ # Line builders # ------------------------------------------------------------------ def _build_account_header_line(self, report, options, account, has_entries, col_data): """Produce the foldable account-level line.""" cols = [] for col_def in options['columns']: expr = col_def['expression_label'] raw = col_data.get(col_def['column_group_key'], {}).get(expr) display_val = ( None if raw is None or (expr == 'amount_currency' and not account.currency_id) else raw ) cols.append(report._build_column_dict( display_val, col_def, options=options, currency=account.currency_id if expr == 'amount_currency' else None, )) lid = report._get_generic_line_id('account.account', account.id) is_unfolded = any( report._get_res_id_from_line_id(ul, 'account.account') == account.id for ul in options.get('unfolded_lines') ) return { 'id': lid, 'name': account.display_name, 'columns': cols, 'level': 1, 'unfoldable': has_entries, 'unfolded': has_entries and (is_unfolded or options.get('unfold_all')), 'expand_function': '_report_expand_unfoldable_line_general_ledger', } def _get_aml_line(self, report, parent_line_id, options, col_dict, running_bal): """Build a single move-line row under a given account header.""" cols = [] for col_def in options['columns']: expr = col_def['expression_label'] raw = col_dict[col_def['column_group_key']].get(expr) cur = None if raw is not None: if expr == 'amount_currency': cur = self.env['res.currency'].browse(col_dict[col_def['column_group_key']]['currency_id']) raw = None if cur == self.env.company.currency_id else raw elif expr == 'balance': raw += (running_bal[col_def['column_group_key']] or 0) cols.append(report._build_column_dict(raw, col_def, options=options, currency=cur)) aml_id = None move_label = None caret = None row_date = None for grp_data in col_dict.values(): aml_id = grp_data.get('id', '') if aml_id: caret = 'account.payment' if grp_data.get('payment_id') else 'account.move.line' move_label = grp_data['move_name'] row_date = str(grp_data.get('date', '')) break return { 'id': report._get_generic_line_id( 'account.move.line', aml_id, parent_line_id=parent_line_id, markup=row_date, ), 'caret_options': caret, 'parent_id': parent_line_id, 'name': move_label, 'columns': cols, 'level': 3, } @api.model def _build_grand_total_line(self, report, options, col_totals): """Build the bottom total row.""" cols = [] for col_def in options['columns']: raw = col_totals[col_def['column_group_key']].get(col_def['expression_label']) cols.append(report._build_column_dict(raw if raw is not None else None, col_def, options=options)) return { 'id': report._get_generic_line_id(None, None, markup='total'), 'name': _('Total'), 'level': 1, 'columns': cols, } # ------------------------------------------------------------------ # Caret / expand handlers # ------------------------------------------------------------------ def caret_option_audit_tax(self, options, params): return self.env['account.generic.tax.report.handler'].caret_option_audit_tax(options, params) def _report_expand_unfoldable_line_general_ledger( self, line_dict_id, groupby, options, progress, offset, unfold_all_batch_data=None, ): """Called when an account line is unfolded. Returns initial-balance, individual AML lines, and load-more metadata.""" def _extract_running_balance(line_dict): return { c['column_group_key']: lc.get('no_format', 0) for c, lc in zip(options['columns'], line_dict['columns']) if c['expression_label'] == 'balance' } report = self.env.ref('fusion_accounting.general_ledger_report') mdl, mdl_id = report._get_model_info_from_id(line_dict_id) if mdl != 'account.account': raise UserError(_("Invalid line ID for general ledger expansion: %s", line_dict_id)) lines = [] # Opening balance (only on first page) if offset == 0: if unfold_all_batch_data: acct_rec, init_by_cg = unfold_all_batch_data['initial_balances'][mdl_id] else: acct_rec, init_by_cg = self._fetch_opening_balances(report, [mdl_id], options)[mdl_id] opening_line = report._get_partner_and_general_ledger_initial_balance_line( options, line_dict_id, init_by_cg, acct_rec.currency_id, ) if opening_line: lines.append(opening_line) progress = _extract_running_balance(opening_line) # Move lines page_size = report.load_more_limit + 1 if report.load_more_limit and options['export_mode'] != 'print' else None if unfold_all_batch_data: aml_rows = unfold_all_batch_data['aml_results'][mdl_id] has_more = unfold_all_batch_data['has_more'].get(mdl_id, False) else: aml_rows, has_more = self._fetch_aml_data(report, options, [mdl_id], offset=offset, limit=page_size) aml_rows = aml_rows[mdl_id] running = progress for entry in aml_rows.values(): row_line = self._get_aml_line(report, line_dict_id, options, entry, running) lines.append(row_line) running = _extract_running_balance(row_line) return { 'lines': lines, 'offset_increment': report.load_more_limit, 'has_more': has_more, 'progress': running, }