481 lines
18 KiB
Python
Executable File
481 lines
18 KiB
Python
Executable File
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. Update account move line dates
|
|
self.env.cr.execute("""
|
|
UPDATE account_move_line
|
|
SET date = %s
|
|
WHERE move_id IN (
|
|
SELECT account_move_id FROM stock_valuation_layer WHERE stock_move_id IN %s
|
|
)
|
|
""", (backdate.date(), move_ids))
|
|
|
|
# 6. Update account analytic line dates
|
|
self.env.cr.execute("""
|
|
UPDATE account_analytic_line
|
|
SET date = %s
|
|
WHERE move_line_id IN (
|
|
SELECT id FROM account_move_line WHERE move_id IN (
|
|
SELECT account_move_id FROM stock_valuation_layer WHERE stock_move_id IN %s
|
|
)
|
|
)
|
|
""", (backdate.date(), move_ids))
|
|
|
|
# 7. 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
|