admin.suherdy/hr_expense_account_split
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: implement SQL-based detachment to bypass Odoo expense locking during payment synchronization

Browse Source
This commit is contained in:
Suherdy Yacob 2026-04-06 13:57:21 +07:00
parent ae46a2deb3
commit 624423dfae
1 changed files with 101 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

178
models/account_payment.py
View File

@ -17,133 +17,140 @@ class AccountPayment(models.Model):
def action_post(self): def action_post(self):
""" """
Confirmation bypass. Calls standard post with skip flag and FORCES a sync Confirmation bypass using SQL De-coupling.
to ensure deductions are applied to the Journal Entry. Temporarily hides the linked expense from Odoo to bypass all rigid locking logic.
""" """
self_bypass = self.with_context(skip_expense_lock=True) # Save links
for payment in self_bypass: sheet = self.expense_sheet_id
if payment.state == 'draft': move_lines = self.move_id.line_ids.filtered(lambda l: l.expense_id)
payment._synchronize_to_moves({'amount', 'deduction_line_ids', 'amount_substract'}) move_lines_map = {line.id: line.expense_id.id for line in move_lines}
# JUMPER: Skip hr_expense's action_post if it exists, or just ensure skip_expense_lock is used globally # SQL Detach
hr_expense_class = self._get_hr_expense_base_class() if sheet:
if hr_expense_class: self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,))
return super(hr_expense_class, self.with_context(skip_expense_lock=True)).action_post() self.env.cr.execute("UPDATE account_move SET expense_sheet_id = NULL WHERE id = %s", (self.id,))
return super(AccountPayment, self.with_context(skip_expense_lock=True)).action_post() 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
def _synchronize_to_moves(self, changed_fields): def _synchronize_to_moves(self, changed_fields):
# 1. Standard synchronization first # 1. Base Synchronization (forced via SQL bypass if already called from action_post)
# We ensure 'amount' is in changed_fields if deductions are involved to trigger base sync
if 'deduction_line_ids' in changed_fields or 'amount_substract' in changed_fields: if 'deduction_line_ids' in changed_fields or 'amount_substract' in changed_fields:
if 'amount' not in changed_fields: if 'amount' not in changed_fields:
changed_fields = set(changed_fields) | {'amount'} changed_fields = set(changed_fields) | {'amount'}
# SURGICAL JUMPER: Bypass the hr_expense method's locking check
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) super()._synchronize_to_moves(changed_fields)
# 2. Custom Deduction Synchronization # 2. Apply complex deductions
# After base sync, we "fix" the lines if deductions are present bypass_ctx = {'skip_account_move_synchronization': True}
# We use with_context(skip_expense_lock=True) for EVERY operation on lines
bypass_ctx = {'skip_expense_lock': True, 'skip_account_move_synchronization': True}
for payment in self.with_context(**bypass_ctx): for payment in self.with_context(**bypass_ctx):
if payment.amount_substract and payment.amount_substract > 0 and payment.move_id: if payment.amount_substract and payment.amount_substract > 0 and payment.move_id:
liquidity_lines, counterpart_lines, writeoff_lines = payment._seek_for_lines() liquidity_lines, counterpart_lines, writeoff_lines = payment._seek_for_lines()
# Adjust liquidity line to final_payment_amount # Update Bank Line to Net
for line in liquidity_lines.with_context(skip_expense_lock=True): for line in liquidity_lines:
final_amount = payment.final_payment_amount final_amount = payment.final_payment_amount
line_vals = { line.write({
'amount_currency': -final_amount if payment.payment_type == 'outbound' else final_amount, '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, '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, '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)
# Deduction lines management # Refresh Deduction Lines
existing_deduction_accounts = payment.deduction_line_ids.mapped('substract_account_id') existing_deduction_accounts = payment.deduction_line_ids.mapped('substract_account_id')
other_lines = payment.move_id.line_ids - liquidity_lines - counterpart_lines - writeoff_lines 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) to_delete = other_lines.filtered(lambda l: l.account_id.id in existing_deduction_accounts.ids)
if to_delete: if to_delete:
# Pass context to unlink as well to_delete.unlink()
to_delete.with_context(skip_expense_lock=True).unlink()
# Re-add deduction lines
for deduction in payment.deduction_line_ids: for deduction in payment.deduction_line_ids:
deduction_amount = deduction.amount_substract ded_amt = deduction.amount_substract
deduction_balance = payment.currency_id._convert(
deduction_amount,
payment.company_id.currency_id,
payment.company_id,
payment.date,
)
line_vals = { line_vals = {
'name': deduction.name or _('Payment Deduction: %s') % deduction.substract_account_id.name, 'name': deduction.name or _('Payment Deduction: %s') % deduction.substract_account_id.name,
'move_id': payment.move_id.id, 'move_id': payment.move_id.id,
'date_maturity': payment.date, 'date_maturity': payment.date,
'amount_currency': -deduction_amount if payment.payment_type == 'outbound' else deduction_amount, 'amount_currency': -ded_amt if payment.payment_type == 'outbound' else ded_amt,
'currency_id': payment.currency_id.id, 'currency_id': payment.currency_id.id,
'debit': deduction_balance if payment.payment_type == 'inbound' else 0.0, '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': deduction_balance if payment.payment_type == 'outbound' 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,
'account_id': deduction.substract_account_id.id, 'account_id': deduction.substract_account_id.id,
} }
payment.env['account.move.line'].with_context(skip_expense_lock=True, check_move_validity=False).create(line_vals) payment.env['account.move.line'].with_context(check_move_validity=False).create(line_vals)
# Force re-balance check payment.move_id._check_balanced()
payment.move_id.with_context(skip_expense_lock=True)._check_balanced()
def _synchronize_from_moves(self, changed_fields): def _synchronize_from_moves(self, changed_fields):
# 1. Standard sync with jumper support # Base restoration logic
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) super()._synchronize_from_moves(changed_fields)
# 2. Custom Deduction Restoration # Restore Gross from lines if Odoo accidentally set it to Net
for payment in self.with_context(skip_expense_lock=True, skip_account_move_synchronization=True): for payment in self.with_context(skip_account_move_synchronization=True):
if payment.amount_substract and payment.amount_substract > 0 and payment.move_id: if payment.amount_substract and payment.amount_substract > 0 and payment.move_id:
liquidity_lines, counterpart_lines, writeoff_lines = payment._seek_for_lines() liquidity_lines, counterpart_lines, writeoff_lines = payment._seek_for_lines()
non_liquidity_lines = counterpart_lines + writeoff_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)
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: if gross_lines:
correct_gross_amount = sum(abs(l.amount_currency) for l in gross_lines) correct_gross = sum(abs(l.amount_currency) for l in gross_lines)
if abs(payment.amount - correct_gross_amount) > 0.001: if abs(payment.amount - correct_gross) > 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, payment.id))
payment.env.cr.execute(
"UPDATE account_payment SET amount = %s WHERE id = %s",
(correct_gross_amount, payment.id)
)
payment.invalidate_recordset(['amount']) payment.invalidate_recordset(['amount'])
payment._compute_final_payment_amount() payment._compute_final_payment_amount()
def write(self, vals): def write(self, vals):
# Propagate bypass flag during writes # SQL Bypass for writes to avoid locked checks
# If we have the flag, we jump over hr_expense to avoid its 'linked to expense' error sheet = self.expense_sheet_id
hr_expense_class = self._get_hr_expense_base_class() if sheet and self._context.get('skip_expense_lock'):
if hr_expense_class and self._context.get('skip_expense_lock'): self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,))
return super(hr_expense_class, self).write(vals) 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
if self.expense_sheet_id and self.state == 'draft': if self.expense_sheet_id and self.state == 'draft':
return super(AccountPayment, self.with_context(skip_expense_lock=True)).write(vals) return super(AccountPayment, self.with_context(skip_expense_lock=True)).write(vals)
return super().write(vals) return super().write(vals)
def action_cancel(self): def action_cancel(self):
# Bypass locks for cancellations as well # SQL Bypass for cancel
hr_expense_class = self._get_hr_expense_base_class() sheet = self.expense_sheet_id
if hr_expense_class: if sheet:
res = super(hr_expense_class, self.with_context(skip_expense_lock=True)).action_cancel() self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,))
else: self.env.cr.execute("UPDATE account_move SET expense_sheet_id = NULL WHERE id = %s", (self.id,))
res = super().action_cancel() 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'])
for payment in self: for payment in self:
if payment.expense_sheet_id: if payment.expense_sheet_id:
@ -153,12 +160,20 @@ class AccountPayment(models.Model):
return res return res
def action_draft(self): def action_draft(self):
# Bypass locks for resetting to draft # SQL Bypass for reset to draft
hr_expense_class = self._get_hr_expense_base_class() sheet = self.expense_sheet_id
if hr_expense_class: if sheet:
res = super(hr_expense_class, self.with_context(skip_expense_lock=True)).action_draft() self.env.cr.execute("UPDATE account_payment SET expense_sheet_id = NULL WHERE id = %s", (self.id,))
else: self.env.cr.execute("UPDATE account_move SET expense_sheet_id = NULL WHERE id = %s", (self.id,))
res = super().action_draft() 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'])
for payment in self: for payment in self:
if payment.expense_sheet_id: if payment.expense_sheet_id:
@ -167,3 +182,4 @@ class AccountPayment(models.Model):
payment.realization_id.expense_sheet_id.invalidate_recordset(['state']) payment.realization_id.expense_sheet_id.invalidate_recordset(['state'])
return res return res