From 12c4edb2213e078a4bed0842fcb94481d3250a69 Mon Sep 17 00:00:00 2001 From: Suherdy Yacob Date: Mon, 6 Apr 2026 14:03:24 +0700 Subject: [PATCH] refactor: replace SQL-based payment bypass with MRO-based surgical jumpers to safely modify expense-linked journal entries --- models/__init__.py | 1 + models/account_move.py | 16 ++++ models/account_move_line.py | 19 ++++ models/account_payment.py | 181 +++++++++++++++++------------------- 4 files changed, 119 insertions(+), 98 deletions(-) create mode 100644 models/account_move.py diff --git a/models/__init__.py b/models/__init__.py index b9bb15e..4b9255e 100644 --- a/models/__init__.py +++ b/models/__init__.py @@ -2,6 +2,7 @@ from . import product_template from . import hr_expense from . import hr_expense_sheet from . import account_move_line +from . import account_move from . import hr_expense_realization from . import account_payment from . import res_company diff --git a/models/account_move.py b/models/account_move.py new file mode 100644 index 0000000..2e14c06 --- /dev/null +++ b/models/account_move.py @@ -0,0 +1,16 @@ +from odoo import models, api + +class AccountMove(models.Model): + _inherit = 'account.move' + + def _get_hr_expense_base_class(self): + """ Returns the hr_expense class in the MRO to jump over it. """ + mro = type(self).mro() + return next((c for c in mro if 'hr_expense' in c.__module__), None) + + def write(self, vals): + # Surgical Jumper to bypass hr_expense's account.move lock + hr_expense_class = self._get_hr_expense_base_class() + if hr_expense_class and self._context.get('skip_expense_lock'): + return super(hr_expense_class, self).write(vals) + return super().write(vals) diff --git a/models/account_move_line.py b/models/account_move_line.py index b182b94..5c0b4a7 100644 --- a/models/account_move_line.py +++ b/models/account_move_line.py @@ -4,6 +4,11 @@ from odoo.tools import float_round class AccountMoveLine(models.Model): _inherit = 'account.move.line' + def _get_hr_expense_base_class(self): + """ Returns the hr_expense class in the MRO to jump over it. """ + mro = type(self).mro() + return next((c for c in mro if 'hr_expense' in c.__module__), None) + @api.model_create_multi def create(self, vals_list): for vals in vals_list: @@ -14,9 +19,23 @@ class AccountMoveLine(models.Model): return super().create(vals_list) def write(self, vals): + # 1. Rounding Logic if 'price_unit' in vals: for line in self: if line.expense_id: currency = line.currency_id or self.env.company.currency_id vals['price_unit'] = float_round(vals['price_unit'], precision_digits=currency.decimal_places or 2) + + # 2. Surgical Jumper to bypass hr_expense's account.move.line lock + hr_expense_class = self._get_hr_expense_base_class() + if hr_expense_class and self._context.get('skip_expense_lock'): + return super(hr_expense_class, self).write(vals) + return super().write(vals) + + def unlink(self): + # Surgical Jumper to bypass hr_expense's account.move.line lock + hr_expense_class = self._get_hr_expense_base_class() + if hr_expense_class and self._context.get('skip_expense_lock'): + return super(hr_expense_class, self).unlink() + return super().unlink() diff --git a/models/account_payment.py b/models/account_payment.py index d93070e..40a0e77 100644 --- a/models/account_payment.py +++ b/models/account_payment.py @@ -17,140 +17,133 @@ class AccountPayment(models.Model): def action_post(self): """ - Confirmation bypass using SQL De-coupling. - Temporarily hides the linked expense from Odoo to bypass all rigid locking logic. + Confirmation bypass. Calls standard post with skip flag and FORCES a sync + to ensure deductions are applied to the Journal Entry. """ - # Save links - sheet = self.expense_sheet_id - move_lines = self.move_id.line_ids.filtered(lambda l: l.expense_id) - move_lines_map = {line.id: line.expense_id.id for line in move_lines} - - # SQL Detach - if sheet: - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.invalidate_recordset(['expense_sheet_id']) - if move_lines_map: - self.env.cr.execute("UPDATE account_move_line SET expense_id = NULL WHERE id IN %s", (tuple(move_lines_map.keys()),)) - self.env['account.move.line'].browse(move_lines_map.keys()).invalidate_recordset(['expense_id']) - - try: - # Force apply deductions with custom labels and amounts - self.with_context(skip_account_move_synchronization=True)._synchronize_to_moves({'amount', 'deduction_line_ids', 'amount_substract'}) - - # Standard posting - res = super(AccountPayment, self).action_post() - finally: - # SQL Restore - if sheet: - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.invalidate_recordset(['expense_sheet_id']) - if move_lines_map: - for line_id, exp_id in move_lines_map.items(): - self.env.cr.execute("UPDATE account_move_line SET expense_id = %s WHERE id = %s", (exp_id, line_id)) - self.env['account.move.line'].browse(move_lines_map.keys()).invalidate_recordset(['expense_id']) - - return res + # 1. Force a synchronization of the moves right before posting. + # We pass 'skip_expense_lock' to all layers (Move, Line) via context. + self_bypass = self.with_context(skip_expense_lock=True) + for payment in self_bypass: + if payment.state == 'draft': + payment._synchronize_to_moves({'amount', 'deduction_line_ids', 'amount_substract'}) + + # 2. Standard confirmation logic with bypass flag for internal writes + hr_expense_class = self._get_hr_expense_base_class() + if hr_expense_class: + return super(hr_expense_class, self.with_context(skip_expense_lock=True)).action_post() + + return super(AccountPayment, self.with_context(skip_expense_lock=True)).action_post() def _synchronize_to_moves(self, changed_fields): - # 1. Base Synchronization (forced via SQL bypass if already called from action_post) + # 1. Standard synchronization first if 'deduction_line_ids' in changed_fields or 'amount_substract' in changed_fields: if 'amount' not in changed_fields: changed_fields = set(changed_fields) | {'amount'} - super()._synchronize_to_moves(changed_fields) + # Triple Jumper: If we have the bypass flag, we jump over the hr_expense method. + # This works in tandem with our Model-level jumpers in account_move and account_move_line. + hr_expense_class = self._get_hr_expense_base_class() + if self._context.get('skip_expense_lock') and hr_expense_class: + super(hr_expense_class, self)._synchronize_to_moves(changed_fields) + else: + super()._synchronize_to_moves(changed_fields) - # 2. Apply complex deductions - bypass_ctx = {'skip_account_move_synchronization': True} - for payment in self.with_context(**bypass_ctx): + # 2. Custom Deduction Synchronization + # After base sync, we "fix" the lines if deductions are present + for payment in self.with_context(skip_expense_lock=True, skip_account_move_synchronization=True): if payment.amount_substract and payment.amount_substract > 0 and payment.move_id: liquidity_lines, counterpart_lines, writeoff_lines = payment._seek_for_lines() - # Update Bank Line to Net + # Adjust liquidity line to final_payment_amount for line in liquidity_lines: final_amount = payment.final_payment_amount - line.write({ + line_vals = { 'amount_currency': -final_amount if payment.payment_type == 'outbound' else final_amount, 'debit': payment.currency_id._convert(final_amount, payment.company_id.currency_id, payment.company_id, payment.date) if payment.payment_type == 'inbound' else 0.0, 'credit': payment.currency_id._convert(final_amount, payment.company_id.currency_id, payment.company_id, payment.date) if payment.payment_type == 'outbound' else 0.0, - }) + } + line.write(line_vals) - # Refresh Deduction Lines + # Deduction lines management existing_deduction_accounts = payment.deduction_line_ids.mapped('substract_account_id') other_lines = payment.move_id.line_ids - liquidity_lines - counterpart_lines - writeoff_lines to_delete = other_lines.filtered(lambda l: l.account_id.id in existing_deduction_accounts.ids) if to_delete: to_delete.unlink() + # Re-add deduction lines for deduction in payment.deduction_line_ids: - ded_amt = deduction.amount_substract + deduction_amount = deduction.amount_substract + deduction_balance = payment.currency_id._convert( + deduction_amount, + payment.company_id.currency_id, + payment.company_id, + payment.date, + ) + line_vals = { 'name': deduction.name or _('Payment Deduction: %s') % deduction.substract_account_id.name, 'move_id': payment.move_id.id, 'date_maturity': payment.date, - 'amount_currency': -ded_amt if payment.payment_type == 'outbound' else ded_amt, + 'amount_currency': -deduction_amount if payment.payment_type == 'outbound' else deduction_amount, 'currency_id': payment.currency_id.id, - 'debit': payment.currency_id._convert(ded_amt, payment.company_id.currency_id, payment.company_id, payment.date) if payment.payment_type == 'inbound' else 0.0, - 'credit': payment.currency_id._convert(ded_amt, payment.company_id.currency_id, payment.company_id, payment.date) if payment.payment_type == 'outbound' else 0.0, + 'debit': deduction_balance if payment.payment_type == 'inbound' else 0.0, + 'credit': deduction_balance if payment.payment_type == 'outbound' else 0.0, 'account_id': deduction.substract_account_id.id, } - payment.env['account.move.line'].with_context(check_move_validity=False).create(line_vals) + payment.env['account.move.line'].with_context(skip_expense_lock=True, check_move_validity=False).create(line_vals) + # Force re-balance check payment.move_id._check_balanced() def _synchronize_from_moves(self, changed_fields): - # Base restoration logic - super()._synchronize_from_moves(changed_fields) + # 1. Standard sync with jumper support + hr_expense_class = self._get_hr_expense_base_class() + if self._context.get('skip_expense_lock') and hr_expense_class: + super(hr_expense_class, self)._synchronize_from_moves(changed_fields) + else: + super()._synchronize_from_moves(changed_fields) - # Restore Gross from lines if Odoo accidentally set it to Net - for payment in self.with_context(skip_account_move_synchronization=True): + # 2. Custom Deduction Restoration + for payment in self.with_context(skip_expense_lock=True, skip_account_move_synchronization=True): if payment.amount_substract and payment.amount_substract > 0 and payment.move_id: liquidity_lines, counterpart_lines, writeoff_lines = payment._seek_for_lines() non_liquidity_lines = counterpart_lines + writeoff_lines - gross_lines = non_liquidity_lines.filtered(lambda l: l.debit > 0 if payment.payment_type == 'outbound' else l.credit > 0) + + # Gross amount is sum of counterparts that balance the liquidity + if payment.payment_type == 'outbound': + gross_lines = non_liquidity_lines.filtered(lambda l: l.debit > 0) + else: + gross_lines = non_liquidity_lines.filtered(lambda l: l.credit > 0) if gross_lines: - correct_gross = sum(abs(l.amount_currency) for l in gross_lines) - if abs(payment.amount - correct_gross) > 0.001: - payment.env.cr.execute("UPDATE account_payment SET amount = %s WHERE id = %s", (correct_gross, payment.id)) + correct_gross_amount = sum(abs(l.amount_currency) for l in gross_lines) + if abs(payment.amount - correct_gross_amount) > 0.001: + # Use SQL to avoid triggering more syncs, then invalidate cache + payment.env.cr.execute( + "UPDATE account_payment SET amount = %s WHERE id = %s", + (correct_gross_amount, payment.id) + ) payment.invalidate_recordset(['amount']) payment._compute_final_payment_amount() def write(self, vals): - # SQL Bypass for writes to avoid locked checks - sheet = self.expense_sheet_id - if sheet and self._context.get('skip_expense_lock'): - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.invalidate_recordset(['expense_sheet_id']) - try: - return super(AccountPayment, self).write(vals) - finally: - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.invalidate_recordset(['expense_sheet_id']) - - # Standard auto-bypass for internal draft writes + # Propagate bypass flag during writes to avoid locked checks + hr_expense_class = self._get_hr_expense_base_class() + if hr_expense_class and self._context.get('skip_expense_lock'): + return super(hr_expense_class, self).write(vals) + if self.expense_sheet_id and self.state == 'draft': return super(AccountPayment, self.with_context(skip_expense_lock=True)).write(vals) return super().write(vals) def action_cancel(self): - # SQL Bypass for cancel - sheet = self.expense_sheet_id - if sheet: - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.invalidate_recordset(['expense_sheet_id']) - - try: - res = super(AccountPayment, self).action_cancel() - finally: - if sheet: - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.invalidate_recordset(['expense_sheet_id']) + # Propagate bypass flag during cancel + hr_expense_class = self._get_hr_expense_base_class() + if hr_expense_class: + res = super(hr_expense_class, self.with_context(skip_expense_lock=True)).action_cancel() + else: + res = super().action_cancel() for payment in self: if payment.expense_sheet_id: @@ -160,20 +153,12 @@ class AccountPayment(models.Model): return res def action_draft(self): - # SQL Bypass for reset to draft - sheet = self.expense_sheet_id - if sheet: - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = NULL WHERE id = %s", (self.id,)) - self.invalidate_recordset(['expense_sheet_id']) - - try: - res = super(AccountPayment, self).action_draft() - finally: - if sheet: - self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.env.cr.execute("UPDATE account_move SET expense_sheet_id = %s WHERE id = %s", (sheet.id, self.id)) - self.invalidate_recordset(['expense_sheet_id']) + # Propagate bypass flag during reset to draft + hr_expense_class = self._get_hr_expense_base_class() + if hr_expense_class: + res = super(hr_expense_class, self.with_context(skip_expense_lock=True)).action_draft() + else: + res = super().action_draft() for payment in self: if payment.expense_sheet_id: