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 # Using context backdate_inventory_mode to allow primary valuation but suppress side-effect revaluations _logger.info(f"Validating {len(moves_to_process)} moves for {self.name} (Selective Valuation)") moves_to_process.with_context(backdate_inventory_mode=True)._action_done() # Step 4: Post-process all moves self._post_process_validated_moves(moves_to_process) self.write({'state': 'done'}) return True def _post_process_validated_moves(self, moves): """Handle backdating for a batch of moves, including valuation and accounting""" self.ensure_one() backdate = self.backdate_datetime # Flush all pending ORM operations to DB before running raw SQL self.env.flush_all() move_ids = tuple(moves.ids) # 1. Update stock move dates self.env.cr.execute( "UPDATE stock_move SET date = %s WHERE id IN %s", (backdate, move_ids) ) # 2. Update stock move line dates self.env.cr.execute( "UPDATE stock_move_line SET date = %s WHERE move_id IN %s", (backdate, move_ids) ) # 3. Update stock valuation layer dates self.env.cr.execute( "UPDATE stock_valuation_layer SET create_date = %s WHERE stock_move_id IN %s", (backdate, move_ids) ) # 4. Update account move dates (journal entries) # We find AMs linked to these moves via SVLs self.env.cr.execute(""" UPDATE account_move SET date = %s WHERE id IN ( SELECT account_move_id FROM stock_valuation_layer WHERE stock_move_id IN %s ) """, (backdate.date(), move_ids)) # 5. 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_print_pdf(self): """Print the PDF report""" self.ensure_one() return self.env.ref('stock_inventory_backdate.action_report_inventory_backdate').report_action(self) def action_export_xlsx(self): """Open XLSX export wizard""" self.ensure_one() return { 'type': 'ir.actions.act_window', 'name': _('Export Inventory Backdate'), 'res_model': 'stock.inventory.backdate.export.wizard', 'view_mode': 'form', 'target': 'new', 'context': {'default_inventory_id': self.id} } def action_draft(self): """Reset to draft""" self.ensure_one() self.write({'state': 'draft'}) return True class StockMove(models.Model): _inherit = 'stock.move' def _create_in_svl(self, forced_quantity=None): """Allow primary SVL creation for backdated adjustments, but handle backdating""" svls = super()._create_in_svl(forced_quantity=forced_quantity) if self.env.context.get('backdate_inventory_mode') and svls: # We will backdate SVLs in post-processing pass return svls def _create_out_svl(self, forced_quantity=None): """Allow primary SVL creation for backdated adjustments, but handle backdating""" svls = super()._create_out_svl(forced_quantity=forced_quantity) if self.env.context.get('backdate_inventory_mode') and svls: # We will backdate SVLs in post-processing pass return svls def product_price_update_before_done(self, forced_qty=None): """ In backdated adjustments, we allow the price update for the move itself, but we bypass the recursive revaluation of older moves (vacuuming). """ if self.env.context.get('backdate_inventory_mode'): # Call super but with a context that bypasses _run_fifo_vacuum return super(StockMove, self.with_context(skip_fifo_vacuum=True)).product_price_update_before_done(forced_qty=forced_qty) return super().product_price_update_before_done(forced_qty=forced_qty) class ProductProduct(models.Model): _inherit = 'product.product' def _run_fifo_vacuum(self, company=None): """Bypass revaluation side-effects during backdated adjustments""" if self.env.context.get('skip_fifo_vacuum') or self.env.context.get('backdate_inventory_mode'): return return super()._run_fifo_vacuum(company=company) 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