Files
odoo_source/addons/stock/models/stock_inventory.py
T
Simon Lejeune b6f053b25e [REF] stock: multi company
Ensure proper domains are applied and enforced on relation fields thanks
to the `check_company` attributes.

product.template
- make responsible a property field in order to ensure proper next activities when a product
  is used between multiple companies
stock.putaway.rule
- added a company_id field
stock.move.line
- company_id is not related anymore since a move line can exist without a move until its validation
stock.package_level
- added a company_id field
stock.picking.type
- company_id is now required
stock.production.lot
- added a company_id field, adapted the constraint accordingly
stock.quant
- check the consistency only in inventory mode
stock.quant.package
- company_id is now empty if the package is empty
stock.picking
- company_id is now related to the one of its picking type

Added some tests.
Moved stock_traceability in the `report` directory.
Removed useless /tests/tours/route.js.

task-1985992
2019-09-03 15:05:03 +00:00

531 lines
25 KiB
Python

# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import api, fields, models, _
from odoo.exceptions import UserError, ValidationError
from odoo.osv import expression
from odoo.tools import float_compare, float_is_zero
class Inventory(models.Model):
_name = "stock.inventory"
_description = "Inventory"
_order = "date desc, id desc"
name = fields.Char(
'Inventory Reference', default="Inventory",
readonly=True, required=True,
states={'draft': [('readonly', False)]})
date = fields.Datetime(
'Inventory Date',
readonly=True, required=True,
default=fields.Datetime.now,
help="If the inventory adjustment is not validated, date at which the theoritical quantities have been checked.\n"
"If the inventory adjustment is validated, date at which the inventory adjustment has been validated.")
line_ids = fields.One2many(
'stock.inventory.line', 'inventory_id', string='Inventories',
copy=False, readonly=False,
states={'done': [('readonly', True)]})
move_ids = fields.One2many(
'stock.move', 'inventory_id', string='Created Moves',
states={'done': [('readonly', True)]})
state = fields.Selection(string='Status', selection=[
('draft', 'Draft'),
('cancel', 'Cancelled'),
('confirm', 'In Progress'),
('done', 'Validated')],
copy=False, index=True, readonly=True,
default='draft')
company_id = fields.Many2one(
'res.company', 'Company',
readonly=True, index=True, required=True,
states={'draft': [('readonly', False)]},
default=lambda self: self.env.company)
location_ids = fields.Many2many(
'stock.location', string='Locations',
readonly=True, check_company=True,
states={'draft': [('readonly', False)]},
domain="[('company_id', '=', company_id), ('usage', 'in', ['internal', 'transit'])]")
product_ids = fields.Many2many(
'product.product', string='Products', check_company=True,
domain="[('type', '=', 'product'), '|', ('company_id', '=', False), ('company_id', '=', company_id)]", readonly=True,
states={'draft': [('readonly', False)]},
help="Specify Products to focus your inventory on particular Products.")
start_empty = fields.Boolean('Empty Inventory',
help="Allows to start with an empty inventory.")
prefill_counted_quantity = fields.Selection(string='Counted Quantities',
help="Allows to start with prefill counted quantity for each lines or "
"with all counted quantity set to zero.", default='counted',
selection=[('counted', 'Default to stock on hand'), ('zero', 'Default to zero')])
@api.onchange('company_id')
def _onchange_company_id(self):
# If the multilocation group is not active, default the location to the one of the main
# warehouse.
if not self.user_has_groups('stock.group_stock_multi_locations'):
warehouse = self.env['stock.warehouse'].search([('company_id', '=', self.company_id.id)], limit=1)
if warehouse:
self.location_ids = warehouse.lot_stock_id
def copy_data(self, default=None):
name = _("%s (copy)") % (self.name)
default = dict(default or {}, name=name)
return super(Inventory, self).copy_data(default)
def unlink(self):
for inventory in self:
if inventory.state not in ('draft', 'cancel'):
raise UserError(_('You can only delete a draft inventory adjustment. If the inventory adjustment is not done, you can cancel it.'))
return super(Inventory, self).unlink()
def action_validate(self):
if not self.exists():
return
self.ensure_one()
if not self.user_has_groups('stock.group_stock_manager'):
raise UserError(_("Only a stock manager can validate an inventory adjustment."))
if self.state != 'confirm':
raise UserError(_(
"You can't validate the inventory '%s', maybe this inventory " +
"has been already validated or isn't ready.") % (self.name))
inventory_lines = self.line_ids.filtered(lambda l: l.product_id.tracking in ['lot', 'serial'] and not l.prod_lot_id and l.theoretical_qty != l.product_qty)
lines = self.line_ids.filtered(lambda l: float_compare(l.product_qty, 1, precision_rounding=l.product_uom_id.rounding) > 0 and l.product_id.tracking == 'serial' and l.prod_lot_id)
if inventory_lines and not lines:
wiz_lines = [(0, 0, {'product_id': product.id, 'tracking': product.tracking}) for product in inventory_lines.mapped('product_id')]
wiz = self.env['stock.track.confirmation'].create({'inventory_id': self.id, 'tracking_line_ids': wiz_lines})
return {
'name': _('Tracked Products in Inventory Adjustment'),
'type': 'ir.actions.act_window',
'view_mode': 'form',
'views': [(False, 'form')],
'res_model': 'stock.track.confirmation',
'target': 'new',
'res_id': wiz.id,
}
self._action_done()
self.line_ids._check_company()
self._check_company()
return True
def _action_done(self):
negative = next((line for line in self.mapped('line_ids') if line.product_qty < 0 and line.product_qty != line.theoretical_qty), False)
if negative:
raise UserError(_('You cannot set a negative product quantity in an inventory line:\n\t%s - qty: %s') % (negative.product_id.name, negative.product_qty))
self.action_check()
self.write({'state': 'done'})
self.post_inventory()
return True
def post_inventory(self):
# The inventory is posted as a single step which means quants cannot be moved from an internal location to another using an inventory
# as they will be moved to inventory loss, and other quants will be created to the encoded quant location. This is a normal behavior
# as quants cannot be reuse from inventory location (users can still manually move the products before/after the inventory if they want).
self.mapped('move_ids').filtered(lambda move: move.state != 'done')._action_done()
def action_check(self):
""" Checks the inventory and computes the stock move to do """
# tde todo: clean after _generate_moves
for inventory in self.filtered(lambda x: x.state not in ('done','cancel')):
# first remove the existing stock moves linked to this inventory
inventory.mapped('move_ids').unlink()
inventory.line_ids._generate_moves()
def action_cancel_draft(self):
self.mapped('move_ids')._action_cancel()
self.line_ids.unlink()
self.write({'state': 'draft'})
def action_start(self):
self.ensure_one()
self._action_start()
self._check_company()
return self.action_open_inventory_lines()
def _action_start(self):
""" Confirms the Inventory Adjustment and generates its inventory lines
if its state is draft and don't have already inventory lines (can happen
with demo data or tests).
"""
for inventory in self:
if inventory.state != 'draft':
continue
vals = {
'state': 'confirm',
'date': fields.Datetime.now()
}
if not inventory.line_ids and not inventory.start_empty:
self.env['stock.inventory.line'].create(inventory._get_inventory_lines_values())
inventory.write(vals)
def action_open_inventory_lines(self):
self.ensure_one()
action = {
'type': 'ir.actions.act_window',
'views': [(self.env.ref('stock.stock_inventory_line_tree2').id, 'tree')],
'view_mode': 'tree',
'name': _('Inventory Lines'),
'res_model': 'stock.inventory.line',
}
context = {
'default_is_editable': True,
'default_inventory_id': self.id,
'default_company_id': self.company_id.id,
}
# Define domains and context
domain = [
('inventory_id', '=', self.id),
('location_id.usage', 'in', ['internal', 'transit'])
]
if self.location_ids:
context['default_location_id'] = self.location_ids[0].id
if len(self.location_ids) == 1:
if not self.location_ids[0].child_ids:
context['readonly_location_id'] = True
if self.product_ids:
if len(self.product_ids) == 1:
context['default_product_id'] = self.product_ids[0].id
action['context'] = context
action['domain'] = domain
return action
def action_view_related_move_lines(self):
self.ensure_one()
domain = [('move_id', 'in', self.move_ids.ids)]
action = {
'name': _('Product Moves'),
'type': 'ir.actions.act_window',
'res_model': 'stock.move.line',
'view_type': 'list',
'view_mode': 'list,form',
'domain': domain,
}
return action
def _get_inventory_lines_values(self):
# TDE CLEANME: is sql really necessary ? I don't think so
locations = self.env['stock.location']
if self.location_ids:
locations = self.env['stock.location'].search([('id', 'child_of', self.location_ids.ids)])
else:
locations = self.env['stock.location'].search([('company_id', '=', self.company_id.id), ('usage', 'in', ['internal', 'transit'])])
domain = ' location_id in %s AND quantity != 0 AND active = TRUE'
args = (tuple(locations.ids),)
vals = []
Product = self.env['product.product']
# Empty recordset of products available in stock_quants
quant_products = self.env['product.product']
# If inventory by company
if self.company_id:
domain += ' AND company_id = %s'
args += (self.company_id.id,)
if self.product_ids:
domain += ' AND product_id in %s'
args += (tuple(self.product_ids.ids),)
self.env['stock.quant'].flush(['company_id', 'product_id', 'quantity', 'location_id', 'lot_id', 'package_id', 'owner_id'])
self.env['product.product'].flush(['active'])
self.env.cr.execute("""SELECT product_id, sum(quantity) as product_qty, location_id, lot_id as prod_lot_id, package_id, owner_id as partner_id
FROM stock_quant
LEFT JOIN product_product
ON product_product.id = stock_quant.product_id
WHERE %s
GROUP BY product_id, location_id, lot_id, package_id, partner_id """ % domain, args)
for product_data in self.env.cr.dictfetchall():
product_data['company_id'] = self.company_id.id
product_data['inventory_id'] = self.id
# replace the None the dictionary by False, because falsy values are tested later on
for void_field in [item[0] for item in product_data.items() if item[1] is None]:
product_data[void_field] = False
product_data['theoretical_qty'] = product_data['product_qty']
if self.prefill_counted_quantity == 'zero':
product_data['product_qty'] = 0
if product_data['product_id']:
product_data['product_uom_id'] = Product.browse(product_data['product_id']).uom_id.id
quant_products |= Product.browse(product_data['product_id'])
vals.append(product_data)
return vals
class InventoryLine(models.Model):
_name = "stock.inventory.line"
_description = "Inventory Line"
_order = "product_id, inventory_id, location_id, prod_lot_id"
@api.model
def _domain_location_id(self):
if self.env.context.get('active_model') == 'stock.inventory':
inventory = self.env['stock.inventory'].browse(self.env.context.get('active_id'))
if inventory.exists() and inventory.location_ids:
return "[('company_id', '=', company_id), ('usage', 'in', ['internal', 'transit']), ('id', 'child_of', %s)]" % inventory.location_ids.ids
return "[('company_id', '=', company_id), ('usage', 'in', ['internal', 'transit'])]"
@api.model
def _domain_product_id(self):
if self.env.context.get('active_model') == 'stock.inventory':
inventory = self.env['stock.inventory'].browse(self.env.context.get('active_id'))
if inventory.exists() and len(inventory.product_ids) > 1:
return "[('type', '=', 'product'), '|', ('company_id', '=', False), ('company_id', '=', company_id), ('id', 'in', %s)]" % inventory.product_ids.ids
return "[('type', '=', 'product'), '|', ('company_id', '=', False), ('company_id', '=', company_id)]"
is_editable = fields.Boolean(help="Technical field to restrict the edition.")
inventory_id = fields.Many2one(
'stock.inventory', 'Inventory', check_company=True,
index=True, ondelete='cascade')
partner_id = fields.Many2one('res.partner', 'Owner', check_company=True)
product_id = fields.Many2one(
'product.product', 'Product', check_company=True,
domain=lambda self: self._domain_product_id(),
index=True, required=True)
product_uom_id = fields.Many2one(
'uom.uom', 'Product Unit of Measure',
required=True, readonly=True)
product_qty = fields.Float(
'Counted Quantity',
digits='Product Unit of Measure', default=0)
categ_id = fields.Many2one(related='product_id.categ_id', store=True)
location_id = fields.Many2one(
'stock.location', 'Location', check_company=True,
domain=lambda self: self._domain_location_id(),
index=True, required=True)
package_id = fields.Many2one(
'stock.quant.package', 'Pack', index=True, check_company=True,
domain="[('location_id', '=', location_id)]",
)
prod_lot_id = fields.Many2one(
'stock.production.lot', 'Lot/Serial Number', check_company=True,
domain="[('product_id','=',product_id), ('company_id', '=', company_id)]")
company_id = fields.Many2one(
'res.company', 'Company', related='inventory_id.company_id',
index=True, readonly=True, store=True)
state = fields.Selection('Status', related='inventory_id.state')
theoretical_qty = fields.Float(
'Theoretical Quantity',
digits='Product Unit of Measure', readonly=True)
difference_qty = fields.Float('Difference', compute='_compute_difference',
help="Indicates the gap between the product's theoretical quantity and its newest quantity.",
readonly=True, digits='Product Unit of Measure', search="_search_difference_qty")
inventory_date = fields.Datetime('Inventory Date', readonly=True,
default=fields.Datetime.now,
help="Last date at which the On Hand Quantity has been computed.")
outdated = fields.Boolean(string='Quantity outdated',
compute='_compute_outdated', search='_search_outdated')
product_tracking = fields.Selection('Tracking', related='product_id.tracking', readonly=True)
@api.depends('product_qty', 'theoretical_qty')
def _compute_difference(self):
for line in self:
line.difference_qty = line.product_qty - line.theoretical_qty
@api.depends('inventory_date', 'product_id.stock_move_ids', 'theoretical_qty', 'product_uom_id.rounding')
def _compute_outdated(self):
grouped_quants = self.env['stock.quant'].read_group(
[('product_id', 'in', self.product_id.ids), ('location_id', 'in', self.location_id.ids)],
['product_id', 'location_id', 'lot_id', 'package_id', 'owner_id', 'quantity:sum'],
['product_id', 'location_id', 'lot_id', 'package_id', 'owner_id'],
lazy=False)
quants = {
(quant['product_id'][0],
quant['location_id'][0],
quant['lot_id'] and quant['lot_id'][0],
quant['package_id'] and quant['package_id'][0],
quant['owner_id'] and quant['owner_id'][0]): quant['quantity']
for quant in grouped_quants
}
for line in self:
if line.state == 'done' or not line.id:
line.outdated = False
continue
qty = quants.get((
line.product_id.id,
line.location_id.id,
line.prod_lot_id.id,
line.package_id.id,
line.partner_id.id,
), 0
)
if float_compare(qty, line.theoretical_qty, precision_rounding=line.product_uom_id.rounding) != 0:
line.outdated = True
else:
line.outdated = False
@api.onchange('product_id', 'location_id', 'product_uom_id', 'prod_lot_id', 'partner_id', 'package_id')
def _onchange_quantity_context(self):
product_qty = False
if self.product_id:
self.product_uom_id = self.product_id.uom_id
if self.product_id and self.location_id and self.product_id.uom_id.category_id == self.product_uom_id.category_id: # TDE FIXME: last part added because crash
theoretical_qty = self.product_id.get_theoretical_quantity(
self.product_id.id,
self.location_id.id,
lot_id=self.prod_lot_id.id,
package_id=self.package_id.id,
owner_id=self.partner_id.id,
to_uom=self.product_uom_id.id,
)
else:
theoretical_qty = 0
# Sanity check on the lot.
if self.prod_lot_id:
if self.product_id.tracking == 'none' or self.product_id != self.prod_lot_id.product_id:
self.prod_lot_id = False
if self.prod_lot_id and self.product_id.tracking == 'serial':
# We force `product_qty` to 1 for SN tracked product because it's
# the only relevant value aside 0 for this kind of product.
self.product_qty = 1
elif self.product_id and float_compare(self.product_qty, self.theoretical_qty, precision_rounding=self.product_uom_id.rounding) == 0:
# We update `product_qty` only if it equals to `theoretical_qty` to
# avoid to reset quantity when user manually set it.
self.product_qty = theoretical_qty
self.theoretical_qty = theoretical_qty
@api.model_create_multi
def create(self, vals_list):
""" Override to handle the case we create inventory line without
`theoretical_qty` because this field is usually computed, but in some
case (typicaly in tests), we create inventory line without trigger the
onchange, so in this case, we set `theoretical_qty` depending of the
product's theoretical quantity.
Handles the same problem with `product_uom_id` as this field is normally
set in an onchange of `product_id`.
Finally, this override checks we don't try to create a duplicated line.
"""
for values in vals_list:
if 'theoretical_qty' not in values:
theoretical_qty = self.env['product.product'].get_theoretical_quantity(
values['product_id'],
values['location_id'],
lot_id=values.get('prod_lot_id'),
package_id=values.get('package_id'),
owner_id=values.get('partner_id'),
to_uom=values.get('product_uom_id'),
)
values['theoretical_qty'] = theoretical_qty
if 'product_id' in values and 'product_uom_id' not in values:
values['product_uom_id'] = self.env['product.product'].browse(values['product_id']).uom_id.id
res = super(InventoryLine, self).create(vals_list)
res._check_no_duplicate_line()
return res
def write(self, vals):
res = super(InventoryLine, self).write(vals)
self._check_no_duplicate_line()
return res
def _check_no_duplicate_line(self):
for line in self:
domain = [
('id', '!=', line.id),
('product_id', '=', line.product_id.id),
('location_id', '=', line.location_id.id),
('partner_id', '=', line.partner_id.id),
('package_id', '=', line.package_id.id),
('prod_lot_id', '=', line.prod_lot_id.id),
('inventory_id', '=', line.inventory_id.id)]
existings = self.search_count(domain)
if existings:
raise UserError(_("There is already one inventory adjustment line for this product,"
" you should rather modify this one instead of creating a new one."))
@api.constrains('product_id')
def _check_product_id(self):
""" As no quants are created for consumable products, it should not be possible do adjust
their quantity.
"""
for line in self:
if line.product_id.type != 'product':
raise ValidationError(_("You can only adjust storable products.") + '\n\n%s -> %s' % (line.product_id.display_name, line.product_id.type))
def _get_move_values(self, qty, location_id, location_dest_id, out):
self.ensure_one()
return {
'name': _('INV:') + (self.inventory_id.name or ''),
'product_id': self.product_id.id,
'product_uom': self.product_uom_id.id,
'product_uom_qty': qty,
'date': self.inventory_id.date,
'company_id': self.inventory_id.company_id.id,
'inventory_id': self.inventory_id.id,
'state': 'confirmed',
'restrict_partner_id': self.partner_id.id,
'location_id': location_id,
'location_dest_id': location_dest_id,
'move_line_ids': [(0, 0, {
'product_id': self.product_id.id,
'lot_id': self.prod_lot_id.id,
'product_uom_qty': 0, # bypass reservation here
'product_uom_id': self.product_uom_id.id,
'qty_done': qty,
'package_id': out and self.package_id.id or False,
'result_package_id': (not out) and self.package_id.id or False,
'location_id': location_id,
'location_dest_id': location_dest_id,
'owner_id': self.partner_id.id,
})]
}
def _generate_moves(self):
vals_list = []
for line in self:
rounding = line.product_id.uom_id.rounding
if float_is_zero(line.difference_qty, precision_rounding=rounding):
continue
if line.difference_qty > 0: # found more than expected
vals = line._get_move_values(line.difference_qty, line.product_id.with_context(force_company=line.company_id.id).property_stock_inventory.id, line.location_id.id, False)
else:
vals = line._get_move_values(abs(line.difference_qty), line.location_id.id, line.product_id.with_context(force_company=line.company_id.id).property_stock_inventory.id, True)
vals_list.append(vals)
return self.env['stock.move'].create(vals_list)
def _refresh_inventory(self):
return self[0].inventory_id.action_open_inventory_lines()
def action_refresh_quantity(self):
filtered_lines = self.filtered(lambda l: l.state != 'done')
for line in filtered_lines:
if line.outdated:
quants = self.env['stock.quant']._gather(line.product_id, line.location_id, lot_id=line.prod_lot_id, package_id=line.package_id, owner_id=line.partner_id, strict=True)
if quants.exists():
quantity = sum(quants.mapped('quantity'))
if line.theoretical_qty != quantity:
line.theoretical_qty = quantity
else:
line.theoretical_qty = 0
line.inventory_date = fields.Datetime.now()
def action_reset_product_qty(self):
""" Write `product_qty` to zero on the selected records. """
impacted_lines = self.env['stock.inventory.line']
for line in self:
if line.state == 'done':
continue
impacted_lines |= line
impacted_lines.write({'product_qty': 0})
def _search_difference_qty(self, operator, value):
if operator == '=':
result = True
elif operator == '!=':
result = False
else:
raise NotImplementedError()
lines = self.search([('inventory_id', '=', self.env.context.get('default_inventory_id'))])
line_ids = lines.filtered(lambda line: float_is_zero(line.difference_qty, line.product_id.uom_id.rounding) == result).ids
return [('id', 'in', line_ids)]
def _search_outdated(self, operator, value):
if operator != '=':
if operator == '!=' and isinstance(value, bool):
value = not value
else:
raise NotImplementedError()
lines = self.search([('inventory_id', '=', self.env.context.get('default_inventory_id'))])
line_ids = lines.filtered(lambda line: line.outdated == value).ids
return [('id', 'in', line_ids)]