# Fusion Accounting - Journal Extensions for Bank Statement Import # File-based import pipeline: parse, validate, create, reconcile from odoo import models, tools, _ from odoo.addons.base.models.res_bank import sanitize_account_number from odoo.exceptions import UserError, RedirectWarning class FusionAccountJournal(models.Model): """Extends journals with a pluggable bank-statement file import pipeline. Sub-modules register parsers by overriding ``_parse_bank_statement_file``.""" _inherit = "account.journal" # ---- Available Import Formats ---- def _get_bank_statements_available_import_formats(self): """Return a list of supported file-format labels (e.g. 'OFX'). Override in sub-modules to register additional formats.""" return [] def __get_bank_statements_available_sources(self): """Append file-import option to the bank-statement source selector when at least one import format is registered.""" sources = super(FusionAccountJournal, self).__get_bank_statements_available_sources() known_formats = self._get_bank_statements_available_import_formats() if known_formats: known_formats.sort() fmt_label = ', '.join(known_formats) sources.append(( "file_import", _("Manual (or import %(import_formats)s)", import_formats=fmt_label), )) return sources # ---- Document Upload Entry Point ---- def create_document_from_attachment(self, attachment_ids=None): """Route attachment uploads to the bank-statement importer when the journal is of type bank, credit, or cash.""" target_journal = self or self.browse(self.env.context.get('default_journal_id')) if target_journal.type in ('bank', 'credit', 'cash'): uploaded_files = self.env['ir.attachment'].browse(attachment_ids) if not uploaded_files: raise UserError(_("No attachment was provided")) return target_journal._import_bank_statement(uploaded_files) return super().create_document_from_attachment(attachment_ids) # ---- Core Import Pipeline ---- def _import_bank_statement(self, attachments): """Orchestrate the full import pipeline: parse -> validate -> find journal -> complete values -> create statements -> reconcile. Returns an action opening the reconciliation widget for the newly imported lines.""" if any(not att.raw for att in attachments): raise UserError(_("You uploaded an invalid or empty file.")) created_statement_ids = [] import_notifications = {} import_errors = {} for att in attachments: try: currency_code, acct_number, parsed_stmts = self._parse_bank_statement_file(att) self._check_parsed_data(parsed_stmts, acct_number) target_journal = self._find_additional_data(currency_code, acct_number) if not target_journal.default_account_id: raise UserError( _('You must set a Default Account for the journal: %s', target_journal.name) ) parsed_stmts = self._complete_bank_statement_vals( parsed_stmts, target_journal, acct_number, att, ) stmt_ids, _line_ids, notifs = self._create_bank_statements(parsed_stmts) created_statement_ids.extend(stmt_ids) # Auto-set the import source on the journal if target_journal.bank_statements_source != 'file_import': target_journal.sudo().bank_statements_source = 'file_import' combined_msg = "" for n in notifs: combined_msg += f"{n['message']}" if notifs: import_notifications[att.name] = combined_msg except (UserError, RedirectWarning) as exc: import_errors[att.name] = exc.args[0] statements = self.env['account.bank.statement'].browse(created_statement_ids) lines_to_reconcile = statements.line_ids if lines_to_reconcile: cron_time_limit = tools.config['limit_time_real_cron'] or -1 effective_limit = cron_time_limit if 0 < cron_time_limit < 180 else 180 lines_to_reconcile._cron_try_auto_reconcile_statement_lines( limit_time=effective_limit, ) widget_action = self.env['account.bank.statement.line']._action_open_bank_reconciliation_widget( extra_domain=[('statement_id', 'in', statements.ids)], default_context={ 'search_default_not_matched': True, 'default_journal_id': statements[:1].journal_id.id, 'notifications': import_notifications, }, ) if import_errors: err_summary = _("The following files could not be imported:\n") err_summary += "\n".join( f"- {fname}: {msg}" for fname, msg in import_errors.items() ) if statements: self.env.cr.commit() raise RedirectWarning( err_summary, widget_action, _('View successfully imported statements'), ) else: raise UserError(err_summary) return widget_action # ---- Parsing (Chain of Responsibility) ---- def _parse_bank_statement_file(self, attachment) -> tuple: """Parse *attachment* into structured statement data. Each module that adds format support must extend this method and return ``super()`` if the format is not recognised. :returns: ``(currency_code, account_number, statements_data)`` :raises RedirectWarning: when no parser can handle the file. """ raise RedirectWarning( message=_("Could not interpret the uploaded file.\n" "Do you have the appropriate import module installed?"), action=self.env.ref('base.open_module_tree').id, button_text=_("Go to Apps"), additional_context={ 'search_default_name': 'account_bank_statement_import', 'search_default_extra': True, }, ) # ---- Validation ---- def _check_parsed_data(self, stmts_vals, acct_number): """Verify that the parsed data contains at least one statement with at least one transaction.""" if not stmts_vals: raise UserError(_( "This file contains no statement for account %s.\n" "If the file covers multiple accounts, import it on each one separately.", acct_number, )) has_transactions = any( sv.get('transactions') for sv in stmts_vals ) if not has_transactions: raise UserError(_( "This file contains no transaction for account %s.\n" "If the file covers multiple accounts, import it on each one separately.", acct_number, )) # ---- Bank Account Matching ---- def _statement_import_check_bank_account(self, acct_number): """Compare *acct_number* against the journal's bank account, accommodating special formats (CH, BNP France, LCL).""" sanitised = self.bank_account_id.sanitized_acc_number.split(" ")[0] # BNP France: 27-char IBAN vs 11-char local if len(sanitised) == 27 and len(acct_number) == 11 and sanitised[:2].upper() == "FR": return sanitised[14:-2] == acct_number # Credit Lyonnais (LCL): 27-char IBAN vs 7-char local if len(sanitised) == 27 and len(acct_number) == 7 and sanitised[:2].upper() == "FR": return sanitised[18:-2] == acct_number return sanitised == acct_number def _find_additional_data(self, currency_code, acct_number): """Locate the matching journal based on currency and account number, creating the bank account link if necessary.""" co_currency = self.env.company.currency_id stmt_currency = None normalised_acct = sanitize_account_number(acct_number) if currency_code: stmt_currency = self.env['res.currency'].search( [('name', '=ilike', currency_code)], limit=1, ) if not stmt_currency: raise UserError(_("No currency found matching '%s'.", currency_code)) if stmt_currency == co_currency: stmt_currency = False target_journal = self if acct_number: if target_journal and not target_journal.bank_account_id: target_journal.set_bank_account(acct_number) elif not target_journal: target_journal = self.search([ ('bank_account_id.sanitized_acc_number', '=', normalised_acct), ]) if not target_journal: partial = self.search([ ('bank_account_id.sanitized_acc_number', 'ilike', normalised_acct), ]) if len(partial) == 1: target_journal = partial else: if not self._statement_import_check_bank_account(normalised_acct): raise UserError(_( 'The statement account (%(account)s) does not match ' 'the journal account (%(journal)s).', account=acct_number, journal=target_journal.bank_account_id.acc_number, )) if target_journal: j_currency = target_journal.currency_id or target_journal.company_id.currency_id if stmt_currency is None: stmt_currency = j_currency if stmt_currency and stmt_currency != j_currency: raise UserError(_( 'Statement currency (%(code)s) differs from journal ' 'currency (%(journal)s).', code=(stmt_currency.name if stmt_currency else co_currency.name), journal=(j_currency.name if j_currency else co_currency.name), )) if not target_journal: raise UserError( _('Unable to determine the target journal. Please select one manually.') ) return target_journal # ---- Value Completion ---- def _complete_bank_statement_vals(self, stmts_vals, journal, acct_number, attachment): """Enrich raw parsed values with journal references, unique import IDs, and partner-bank associations.""" for sv in stmts_vals: if not sv.get('reference'): sv['reference'] = attachment.name for txn in sv['transactions']: txn['journal_id'] = journal.id uid = txn.get('unique_import_id') if uid: normalised = sanitize_account_number(acct_number) prefix = f"{normalised}-" if normalised else "" txn['unique_import_id'] = f"{prefix}{journal.id}-{uid}" if not txn.get('partner_bank_id'): ident_str = txn.get('account_number') if ident_str: if txn.get('partner_id'): bank_match = self.env['res.partner.bank'].search([ ('acc_number', '=', ident_str), ('partner_id', '=', txn['partner_id']), ]) else: bank_match = self.env['res.partner.bank'].search([ ('acc_number', '=', ident_str), ('company_id', 'in', (False, journal.company_id.id)), ]) if bank_match and len(bank_match) == 1: txn['partner_bank_id'] = bank_match.id txn['partner_id'] = bank_match.partner_id.id return stmts_vals # ---- Statement Creation ---- def _create_bank_statements(self, stmts_vals, raise_no_imported_file=True): """Create bank statements from the enriched values, skipping duplicate transactions and generating PDF attachments for complete statements. :returns: ``(statement_ids, line_ids, notifications)`` """ BankStmt = self.env['account.bank.statement'] BankStmtLine = self.env['account.bank.statement.line'] new_stmt_ids = [] new_line_ids = [] skipped_imports = [] for sv in stmts_vals: accepted_txns = [] for txn in sv['transactions']: uid = txn.get('unique_import_id') already_exists = ( uid and BankStmtLine.sudo().search( [('unique_import_id', '=', uid)], limit=1, ) ) if txn['amount'] != 0 and not already_exists: accepted_txns.append(txn) else: skipped_imports.append(txn) if sv.get('balance_start') is not None: sv['balance_start'] += float(txn['amount']) if accepted_txns: sv.pop('transactions', None) sv['line_ids'] = [[0, False, line] for line in accepted_txns] new_stmt = BankStmt.with_context( default_journal_id=self.id, ).create(sv) if not new_stmt.name: new_stmt.name = sv['reference'] new_stmt_ids.append(new_stmt.id) new_line_ids.extend(new_stmt.line_ids.ids) if new_stmt.is_complete and not self.env.context.get('skip_pdf_attachment_generation'): new_stmt.action_generate_attachment() if not new_line_ids and raise_no_imported_file: raise UserError(_('You already have imported that file.')) user_notifications = [] num_skipped = len(skipped_imports) if num_skipped: user_notifications.append({ 'type': 'warning', 'message': ( _("%d transactions had already been imported and were ignored.", num_skipped) if num_skipped > 1 else _("1 transaction had already been imported and was ignored.") ), }) return new_stmt_ids, new_line_ids, user_notifications