from odoo import api, fields, models, _ from odoo.exceptions import UserError, ValidationError from odoo.tools import float_is_zero, float_compare import logging _logger = logging.getLogger(__name__) class StockInventoryBackdate(models.Model): _name = 'stock.inventory.backdate' _description = 'Backdated Inventory Adjustment' _inherit = ['mail.thread', 'mail.activity.mixin'] _order = 'backdate_datetime desc, id desc' name = fields.Char(string='Reference', required=True, default='New', readonly=True, tracking=True) backdate_datetime = fields.Datetime( string='Adjustment Date & Time', required=True, default=fields.Datetime.now, help="The date and time for the backdated inventory adjustment", tracking=True ) location_id = fields.Many2one( 'stock.location', string='Location', required=True, domain="[('usage', '=', 'internal')]", tracking=True ) company_id = fields.Many2one( 'res.company', string='Company', required=True, default=lambda self: self.env.company ) state = fields.Selection([ ('draft', 'Draft'), ('done', 'Done'), ('cancel', 'Cancelled') ], string='Status', default='draft', readonly=True, tracking=True) line_ids = fields.One2many( 'stock.inventory.backdate.line', 'inventory_id', string='Inventory Lines' ) notes = fields.Text(string='Notes') @api.model def create(self, vals): if vals.get('name', 'New') == 'New': vals['name'] = self.env['ir.sequence'].next_by_code('stock.inventory.backdate') or 'New' return super(StockInventoryBackdate, self).create(vals) def action_load_products(self): """Load products with current inventory at the location""" self.ensure_one() if self.state != 'draft': raise UserError(_('You can only load products in draft state.')) if not self.location_id: raise UserError(_('Please select a location first.')) # Get all quants at this location quants = self.env['stock.quant'].search([ ('location_id', '=', self.location_id.id), ('company_id', '=', self.company_id.id), ('quantity', '!=', 0) ]) if not quants: raise UserError(_('No products found at this location. You can add products manually.')) # Get existing product IDs to avoid duplicates existing_product_ids = self.line_ids.mapped('product_id').ids lines = [] for quant in quants: # Skip if already in lines if quant.product_id.id in existing_product_ids: continue # Get inventory position at the backdate historical_qty = self._get_historical_quantity( quant.product_id, quant.location_id, quant.lot_id, quant.package_id, quant.owner_id, self.backdate_datetime ) lines.append((0, 0, { 'product_id': quant.product_id.id, 'lot_id': quant.lot_id.id, 'package_id': quant.package_id.id, 'owner_id': quant.owner_id.id, 'theoretical_qty': historical_qty, 'counted_qty': historical_qty, 'difference_qty': 0.0, })) if lines: self.line_ids = [(0, 0, line[2]) for line in lines] return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _('Products Loaded'), 'message': _('%s product(s) loaded successfully.') % len(lines), 'type': 'success', 'sticky': False, } } else: return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _('No New Products'), 'message': _('All products are already in the list.'), 'type': 'info', 'sticky': False, } } def _get_historical_quantity(self, product, location, lot, package, owner, date): """Calculate inventory quantity at a specific date""" base_domain = [ ('product_id', '=', product.id), ('state', '=', 'done'), ('date', '<=', date), ] if lot: base_domain.append(('lot_id', '=', lot.id)) if package: base_domain.append(('package_id', '=', package.id)) if owner: base_domain.append(('owner_id', '=', owner.id)) # Get all incoming moves (destination = our location) domain_in = base_domain + [('location_dest_id', '=', location.id)] moves_in = self.env['stock.move.line'].search(domain_in) # Get all outgoing moves (source = our location) domain_out = base_domain + [('location_id', '=', location.id)] moves_out = self.env['stock.move.line'].search(domain_out) qty_in = sum(moves_in.mapped('quantity')) qty_out = sum(moves_out.mapped('quantity')) return qty_in - qty_out def action_validate(self): """Validate and create backdated stock moves in batch""" self.ensure_one() if self.state != 'draft': raise UserError(_('Only draft adjustments can be validated.')) if not self.line_ids: raise UserError(_('Please add at least one inventory line.')) # Find inventory adjustment location inventory_location = self.env['stock.location'].search([ ('usage', '=', 'inventory'), ('company_id', 'in', [self.company_id.id, False]) ], limit=1) if not inventory_location: raise UserError(_('Inventory adjustment location not found. Please check your stock configuration.')) moves_to_process = self.env['stock.move'] move_line_data = [] # To store move line info linked to moves # Step 1: Create all stock moves for line in self.line_ids: if line.difference_qty == 0: continue # Determine source and destination based on difference if line.difference_qty > 0: location_id = inventory_location.id location_dest_id = self.location_id.id qty = line.difference_qty else: location_id = self.location_id.id location_dest_id = inventory_location.id qty = abs(line.difference_qty) backdate = self.backdate_datetime move_vals = { 'name': _('Backdated Inventory Adjustment: %s') % self.name, 'product_id': line.product_id.id, 'product_uom': line.product_uom_id.id, 'product_uom_qty': qty, 'location_id': location_id, 'location_dest_id': location_dest_id, 'company_id': self.company_id.id, 'is_inventory': True, 'origin': self.name, 'date': backdate, 'picked': True, # Mark as picked for Odoo 17 } move = self.env['stock.move'].create(move_vals) moves_to_process |= move # Store data for move line creation move_line_data.append({ 'move': move, 'line': line, 'qty': qty, 'location_id': location_id, 'location_dest_id': location_dest_id, 'backdate': backdate }) if not moves_to_process: self.write({'state': 'done'}) return True # Step 2: Confirm moves and handle move lines moves_to_process._action_confirm() for data in move_line_data: move = data['move'] line = data['line'] # Check if move line was already created by _action_confirm (e.g. reservation) existing_ml = move.move_line_ids.filtered(lambda ml: ml.product_id.id == line.product_id.id) ml_vals = { 'product_id': line.product_id.id, 'product_uom_id': line.product_uom_id.id, 'quantity': data['qty'], 'location_id': data['location_id'], 'location_dest_id': data['location_dest_id'], 'lot_id': line.lot_id.id if line.lot_id else False, 'package_id': line.package_id.id if line.package_id else False, 'owner_id': line.owner_id.id if line.owner_id else False, 'date': data['backdate'], 'picked': True, } if existing_ml: existing_ml[0].write(ml_vals) else: ml_vals['move_id'] = move.id self.env['stock.move.line'].create(ml_vals) # Step 3: Action Done on all moves at once _logger.info(f"Validating {len(moves_to_process)} moves for {self.name}") moves_to_process._action_done() # Step 4: Post-process all moves (backdating and accounting) self._post_process_validated_moves(moves_to_process) self.write({'state': 'done'}) return True def _post_process_validated_moves(self, moves): """Handle backdating and accounting adjustments for a batch of moves""" self.ensure_one() backdate = self.backdate_datetime account_date = backdate.date() # Flush all pending ORM operations to DB before running raw SQL self.env.flush_all() # 1. Update stock move dates self.env.cr.execute( "UPDATE stock_move SET date = %s WHERE id IN %s", (backdate, tuple(moves.ids)) ) # 2. Update stock move line dates self.env.cr.execute( "UPDATE stock_move_line SET date = %s WHERE move_id IN %s", (backdate, tuple(moves.ids)) ) # 3. Update stock valuation layers svls = self.env['stock.valuation.layer'].search([('stock_move_id', 'in', moves.ids)]) if svls: self.env.cr.execute( "UPDATE stock_valuation_layer SET create_date = %s WHERE id IN %s", (backdate, tuple(svls.ids)) ) # 4. Update account moves account_moves = moves.account_move_ids if account_moves: # Update account move dates self.env.cr.execute( "UPDATE account_move SET date = %s WHERE id IN %s", (account_date, tuple(account_moves.ids)) ) # Update account move line dates self.env.cr.execute( "UPDATE account_move_line SET date = %s WHERE move_id IN %s", (account_date, tuple(account_moves.ids)) ) # Update account for non-valuation lines if target account 521301 exists target_account = self.env['account.account'].search([ ('code', '=', '521301'), ('company_id', '=', self.company_id.id) ], limit=1) if target_account: for move in moves: product = move.product_id valuation_account = product.categ_id.property_stock_valuation_account_id if valuation_account and move.account_move_ids: # Update the line that is NOT the stock valuation account self.env.cr.execute( """UPDATE account_move_line SET account_id = %s WHERE move_id IN %s AND account_id != %s""", (target_account.id, tuple(move.account_move_ids.ids), valuation_account.id) ) # Clear cache to reflect changes self.env.invalidate_all() def action_cancel(self): """Cancel the adjustment""" self.ensure_one() if self.state == 'done': raise UserError(_('Cannot cancel a validated adjustment.')) self.write({'state': 'cancel'}) return True def action_draft(self): """Reset to draft""" self.ensure_one() self.write({'state': 'draft'}) return True class StockInventoryBackdateLine(models.Model): _name = 'stock.inventory.backdate.line' _description = 'Backdated Inventory Adjustment Line' _sql_constraints = [ ('unique_product_per_inventory', 'unique(inventory_id, product_id, lot_id, package_id, owner_id)', 'You cannot have duplicate products with the same lot/package/owner in the same adjustment!') ] inventory_id = fields.Many2one( 'stock.inventory.backdate', string='Inventory Adjustment', required=True, ondelete='cascade' ) product_id = fields.Many2one( 'product.product', string='Product', required=True, domain="[('type', '=', 'product')]" ) lot_id = fields.Many2one('stock.lot', string='Lot/Serial Number') package_id = fields.Many2one('stock.quant.package', string='Package') owner_id = fields.Many2one('res.partner', string='Owner') theoretical_qty = fields.Float( string='Theoretical Quantity', readonly=True, help="Quantity at the backdated time (can be negative if there was negative stock)" ) counted_qty = fields.Float( string='Counted Quantity', required=True, default=0.0 ) difference_qty = fields.Float( string='Adjustment Qty (+/-)', default=0.0, help="Positive value adds stock, negative value removes stock." ) product_uom_id = fields.Many2one( 'uom.uom', string='Unit of Measure', related='product_id.uom_id', readonly=True ) state = fields.Selection(related='inventory_id.state', string='Status') has_negative_theoretical = fields.Boolean( string='Has Negative Theoretical', compute='_compute_has_negative_theoretical', help="Indicates if theoretical quantity is negative" ) @api.onchange('counted_qty') def _onchange_counted_qty(self): for line in self: line.difference_qty = line.counted_qty - line.theoretical_qty @api.onchange('difference_qty') def _onchange_difference_qty(self): for line in self: line.counted_qty = line.theoretical_qty + line.difference_qty @api.depends('theoretical_qty') def _compute_has_negative_theoretical(self): for line in self: line.has_negative_theoretical = line.theoretical_qty < 0 @api.onchange('product_id', 'lot_id', 'package_id', 'owner_id') def _onchange_product_id(self): """Auto-calculate theoretical quantity when product is selected""" if self.product_id and self.inventory_id.location_id and self.inventory_id.backdate_datetime: self.theoretical_qty = self.inventory_id._get_historical_quantity( self.product_id, self.inventory_id.location_id, self.lot_id, self.package_id, self.owner_id, self.inventory_id.backdate_datetime ) # Set counted_qty to theoretical_qty by default (Adjustment 0) if not self.counted_qty and not self.difference_qty: self.counted_qty = self.theoretical_qty self.difference_qty = 0.0