3e83d8d7d5
Includes: - Odoo 19 framework (odoo/) - 27 custom EnCoach addons (new_project/custom_addons/) - encoach_core, encoach_api, encoach_lms_api, encoach_adaptive_api - encoach_exam, encoach_taxonomy, encoach_adaptive, encoach_assignment - encoach_ai, encoach_ai_grading, encoach_ai_generation, encoach_ai_media - encoach_courseware, encoach_communication, encoach_subscription - encoach_notification, encoach_approval, encoach_branding - encoach_classroom, encoach_registration, encoach_stats - encoach_faq, encoach_ticket, encoach_training, encoach_resources - encoach_adaptive_ai, encoach_sis - 21 OpenEduCat Enterprise modules (new_project/enterprise-19/) - 14 OpenEduCat Community modules (new_project/openeducat_erp-19.0/) - Configuration: odoo.conf, requirements.txt, scripts - 200+ REST API endpoints with JWT authentication - SRS and test documentation Made-with: Cursor
412 lines
20 KiB
Python
412 lines
20 KiB
Python
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
|
|
|
import operator as py_operator
|
|
from collections.abc import Iterable
|
|
from re import findall as regex_findall, split as regex_split
|
|
from collections import defaultdict
|
|
|
|
from odoo import _, api, fields, models
|
|
from odoo.exceptions import UserError, ValidationError
|
|
from odoo.fields import Domain
|
|
|
|
PY_OPERATORS = {
|
|
'<': py_operator.lt,
|
|
'>': py_operator.gt,
|
|
'<=': py_operator.le,
|
|
'>=': py_operator.ge,
|
|
'=': py_operator.eq,
|
|
'!=': py_operator.ne,
|
|
'in': lambda elem, container: elem in container,
|
|
'not in': lambda elem, container: elem not in container,
|
|
}
|
|
|
|
|
|
class StockLot(models.Model):
|
|
_name = 'stock.lot'
|
|
_inherit = ['mail.thread', 'mail.activity.mixin']
|
|
_description = 'Lot/Serial'
|
|
_check_company_auto = True
|
|
_order = 'name, id'
|
|
|
|
@api.model
|
|
def default_get(self, fields):
|
|
context = dict(self.env.context)
|
|
# We always want the company_id to be computed, regardless of where it's been created.
|
|
context.pop('default_company_id', False)
|
|
return super(StockLot, self.with_context(context)).default_get(fields)
|
|
|
|
def _read_group_location_id(self, locations, domain):
|
|
partner_locations = locations.search([('usage', 'in', ('customer', 'supplier'))])
|
|
return partner_locations + locations.warehouse_id.search([]).lot_stock_id
|
|
|
|
name = fields.Char('Lot/Serial Number', required=True, compute='_compute_name', store=True, readonly=False, help="Unique Lot/Serial Number", index='trigram', precompute=True)
|
|
ref = fields.Char('Internal Reference', help="Internal reference number in case it differs from the manufacturer's lot/serial number")
|
|
product_id = fields.Many2one(
|
|
'product.product', 'Product', index=True,
|
|
domain=("[('tracking', '!=', 'none'), ('is_storable', '=', True)] +"
|
|
" ([('product_tmpl_id', '=', context['default_product_tmpl_id'])] if context.get('default_product_tmpl_id') else [])"),
|
|
required=True, check_company=True, tracking=True)
|
|
product_uom_id = fields.Many2one(
|
|
'uom.uom', 'Unit',
|
|
related='product_id.uom_id')
|
|
quant_ids = fields.One2many('stock.quant', 'lot_id', 'Quants', readonly=True)
|
|
product_qty = fields.Float('On Hand Quantity', compute='_product_qty', search='_search_product_qty')
|
|
note = fields.Html(string='Description')
|
|
display_complete = fields.Boolean(compute='_compute_display_complete')
|
|
company_id = fields.Many2one('res.company', 'Company', index=True, store=True, readonly=False, compute='_compute_company_id')
|
|
delivery_ids = fields.Many2many('stock.picking', compute='_compute_delivery_ids', string='Transfers')
|
|
delivery_count = fields.Integer('Delivery order count', compute='_compute_delivery_ids')
|
|
partner_ids = fields.Many2many('res.partner', compute='_compute_partner_ids', search='_search_partner_ids')
|
|
lot_properties = fields.Properties('Properties', definition='product_id.lot_properties_definition', copy=True)
|
|
location_id = fields.Many2one(
|
|
'stock.location', 'Location', compute='_compute_single_location', store=True, readonly=False,
|
|
inverse='_set_single_location', domain="[('usage', '!=', 'view')]", group_expand='_read_group_location_id')
|
|
|
|
@api.depends('product_id')
|
|
def _compute_name(self):
|
|
for lot in self:
|
|
if not lot.name:
|
|
lot.name = lot.product_id.lot_sequence_id.next_by_id() if lot.product_id.lot_sequence_id else False
|
|
|
|
@api.model
|
|
def generate_lot_names(self, first_lot, count):
|
|
"""Generate `lot_names` from a string."""
|
|
# We look if the first lot contains at least one digit.
|
|
caught_initial_number = regex_findall(r"\d+", first_lot)
|
|
if not caught_initial_number:
|
|
return self.generate_lot_names(first_lot + "0", count)
|
|
# We base the series on the last number found in the base lot.
|
|
initial_number = caught_initial_number[-1]
|
|
padding = len(initial_number)
|
|
# We split the lot name to get the prefix and suffix.
|
|
splitted = regex_split(initial_number, first_lot)
|
|
# initial_number could appear several times, e.g. BAV023B00001S00001
|
|
prefix = initial_number.join(splitted[:-1])
|
|
suffix = splitted[-1]
|
|
initial_number = int(initial_number)
|
|
|
|
return [{
|
|
'lot_name': '%s%s%s' % (prefix, str(initial_number + i).zfill(padding), suffix),
|
|
} for i in range(0, count)]
|
|
|
|
@api.model
|
|
def _get_next_serial(self, company, product):
|
|
"""Return the next serial number to be attributed to the product."""
|
|
if product.tracking != "none":
|
|
last_serial = self.env['stock.lot'].search(
|
|
['|', ('company_id', '=', company.id), ('company_id', '=', False), ('product_id', '=', product.id)],
|
|
limit=1, order='id DESC')
|
|
if last_serial:
|
|
return self.env['stock.lot'].generate_lot_names(last_serial.name, 2)[1]['lot_name']
|
|
return False
|
|
|
|
@api.constrains('name', 'product_id', 'company_id')
|
|
def _check_unique_lot(self):
|
|
domain = [('product_id', 'in', self.product_id.ids),
|
|
('name', 'in', self.mapped('name'))]
|
|
groupby = ['company_id', 'product_id', 'name']
|
|
if any(not lot.company_id for lot in self):
|
|
# We need to check across other companies to not have duplicates between 'no-company' and a company.
|
|
self = self.sudo()
|
|
records = self.with_context(skip_preprocess_gs1=True)._read_group(domain, groupby, ['__count'], order='company_id DESC')
|
|
error_message_lines = set()
|
|
cross_lots = {}
|
|
for company, product, name, count in records:
|
|
if not company:
|
|
cross_lots[(product, name)] = count
|
|
# For company-specific lots, we check that there is no duplicate with 'no-company' lots, but NOT between specific-company ones.
|
|
if (company and (cross_lots.get((product, name), 0) + count) > 1) or count > 1:
|
|
error_message_lines.add(_(" - Product: %(product)s, Lot/Serial Number: %(lot)s", product=product.display_name, lot=name))
|
|
if error_message_lines:
|
|
raise ValidationError(
|
|
_(
|
|
"The combination of lot/serial number and product must be unique within a company including when no company is defined.\nThe following combinations contain duplicates:\n%(error_lines)s",
|
|
error_lines="\n".join(error_message_lines),
|
|
),
|
|
)
|
|
|
|
def _check_create(self):
|
|
active_picking_id = self.env.context.get('active_picking_id', False)
|
|
if active_picking_id:
|
|
picking_id = self.env['stock.picking'].browse(active_picking_id)
|
|
if picking_id and not picking_id.picking_type_id.use_create_lots:
|
|
raise UserError(_('You are not allowed to create a lot or serial number with this operation type. To change this, go on the operation type and tick the box "Create New Lots/Serial Numbers".'))
|
|
|
|
@api.depends('product_id.company_id')
|
|
def _compute_company_id(self):
|
|
for lot in self:
|
|
if self.env.company in lot.product_id.company_id.all_child_ids and lot.product_id.company_id not in self.env.companies:
|
|
lot.company_id = self.env.company
|
|
else:
|
|
lot.company_id = lot.product_id.company_id
|
|
|
|
@api.depends('name')
|
|
def _compute_display_complete(self):
|
|
""" Defines if we want to display all fields in the stock.production.lot form view.
|
|
It will if the record exists (`id` set) or if we precised it into the context.
|
|
This compute depends on field `name` because as it has always a default value, it'll be
|
|
always triggered.
|
|
"""
|
|
for prod_lot in self:
|
|
prod_lot.display_complete = prod_lot.id or self.env.context.get('display_complete')
|
|
|
|
def _compute_delivery_ids(self):
|
|
delivery_ids_by_lot = self._find_delivery_ids_by_lot_iterative()
|
|
for lot in self:
|
|
lot.delivery_ids = delivery_ids_by_lot.get(lot.id, [])
|
|
lot.delivery_count = len(lot.delivery_ids)
|
|
|
|
def _compute_partner_ids(self):
|
|
delivery_ids_by_lot = self._find_delivery_ids_by_lot_iterative()
|
|
for lot in self:
|
|
if delivery_ids_by_lot.get(lot.id, []):
|
|
lot.partner_ids = self.env['stock.picking'].browse(delivery_ids_by_lot[lot.id]).sorted(key='date_done', reverse=True).partner_id
|
|
else:
|
|
lot.partner_ids = False
|
|
|
|
@api.depends('quant_ids', 'quant_ids.quantity')
|
|
def _compute_single_location(self):
|
|
for lot in self:
|
|
quants = lot.quant_ids.filtered(lambda q: q.quantity > 0)
|
|
lot.location_id = quants.location_id if len(quants.location_id) == 1 else False
|
|
|
|
def _set_single_location(self):
|
|
quants = self.quant_ids.filtered(lambda q: q.quantity > 0)
|
|
if len(quants.location_id) == 1:
|
|
unpack = len(quants.package_id.quant_ids) > 1
|
|
quants.move_quants(location_dest_id=self.location_id, message=_("Lot/Serial Number Relocated"), unpack=unpack)
|
|
elif len(quants.location_id) > 1:
|
|
raise UserError(_('You can only move a lot/serial to a new location if it exists in a single location.'))
|
|
|
|
@api.model_create_multi
|
|
def create(self, vals_list):
|
|
lot_product_ids = {val.get('product_id') for val in vals_list} | {self.env.context.get('default_product_id')}
|
|
self.with_context(lot_product_ids=lot_product_ids)._check_create()
|
|
return super(StockLot, self.with_context(mail_create_nosubscribe=True)).create(vals_list)
|
|
|
|
def write(self, vals):
|
|
if 'company_id' in vals:
|
|
for lot in self:
|
|
if lot.location_id.company_id and vals['company_id'] and lot.location_id.company_id.id != vals['company_id']:
|
|
raise UserError(_("You cannot change the company of a lot/serial number currently in a location belonging to another company."))
|
|
if 'product_id' in vals and any(vals['product_id'] != lot.product_id.id for lot in self):
|
|
move_lines = self.env['stock.move.line'].search([('lot_id', 'in', self.ids), ('product_id', '!=', vals['product_id'])])
|
|
if move_lines:
|
|
raise UserError(_(
|
|
'You are not allowed to change the product linked to a serial or lot number '
|
|
'if some stock moves have already been created with that number. '
|
|
'This would lead to inconsistencies in your stock.'
|
|
))
|
|
return super().write(vals)
|
|
|
|
def copy_data(self, default=None):
|
|
default = dict(default or {})
|
|
vals_list = super().copy_data(default=default)
|
|
if 'name' not in default:
|
|
for lot, vals in zip(self, vals_list):
|
|
vals['name'] = _("(copy of) %s", lot.name)
|
|
return vals_list
|
|
|
|
@api.depends('quant_ids', 'quant_ids.quantity')
|
|
def _product_qty(self):
|
|
for lot in self:
|
|
# We only care for the quants in internal or transit locations.
|
|
quants = lot.quant_ids.filtered(lambda q: q.location_id.usage == 'internal' or (q.location_id.usage == 'transit' and q.location_id.company_id))
|
|
lot.product_qty = sum(quants.mapped('quantity'))
|
|
|
|
def _search_product_qty(self, operator, value):
|
|
op = PY_OPERATORS.get(operator)
|
|
if not op:
|
|
return NotImplemented
|
|
if isinstance(value, Iterable) and not isinstance(value, str):
|
|
value = {float(v) for v in value}
|
|
else:
|
|
value = float(value)
|
|
domain = [
|
|
('lot_id', '!=', False),
|
|
'|', ('location_id.usage', '=', 'internal'),
|
|
'&', ('location_id.usage', '=', 'transit'), ('location_id.company_id', '!=', False)
|
|
]
|
|
lots_w_qty = self.env['stock.quant']._read_group(domain=domain, groupby=['lot_id'], aggregates=['quantity:sum'], having=[('quantity:sum', '!=', 0)])
|
|
ids = []
|
|
lot_ids_w_qty = []
|
|
for lot, quantity_sum in lots_w_qty:
|
|
lot_id = lot.id
|
|
lot_ids_w_qty.append(lot_id)
|
|
if op(quantity_sum, value):
|
|
ids.append(lot_id)
|
|
|
|
# check if we need include zero values in result
|
|
include_zero = op(0.0, value)
|
|
if include_zero:
|
|
return ['|', ('id', 'in', ids), ('id', 'not in', lot_ids_w_qty)]
|
|
return [('id', 'in', ids)]
|
|
|
|
def _search_partner_ids(self, operator, value):
|
|
""" returns partner_ids that are directly delivered the product of the lot/SN, i.e. not
|
|
lots/SNs that are consumed within a MO. This means this search is NOT symmetric with the
|
|
partner_ids field within the form view since it uses different logic that isn't efficient
|
|
enough for this search due to it being usable within the list view.
|
|
"""
|
|
if operator in Domain.NEGATIVE_OPERATORS or not isinstance(value, (Iterable)):
|
|
return NotImplemented
|
|
is_no_partner = operator == 'in' and list(value) == [False]
|
|
domain = Domain([
|
|
('lot_id', '!=', False),
|
|
('state', '=', 'done'),
|
|
])
|
|
if is_no_partner:
|
|
# reverse the search, get all lots sent to partner so we can return all lots NOT sent
|
|
domain &= Domain('picking_partner_id', 'not in', value)
|
|
else:
|
|
domain &= Domain.OR([
|
|
Domain('picking_partner_id', operator, value),
|
|
Domain('move_partner_id', operator, value),
|
|
])
|
|
domain &= Domain(self._get_outgoing_domain())
|
|
move_lines = self.env['stock.move.line'].search(domain)
|
|
|
|
if is_no_partner:
|
|
return [('id', 'not in', move_lines.lot_id.ids)]
|
|
return [('id', 'in', move_lines.lot_id.ids)]
|
|
|
|
def action_lot_open_quants(self):
|
|
self = self.with_context(search_default_lot_id=self.id, create=False)
|
|
if self.env.user.has_group('stock.group_stock_manager'):
|
|
self = self.with_context(inventory_mode=True)
|
|
return self.env['stock.quant'].action_view_quants()
|
|
|
|
def action_lot_open_transfers(self):
|
|
self.ensure_one()
|
|
|
|
action = {
|
|
'res_model': 'stock.picking',
|
|
'type': 'ir.actions.act_window'
|
|
}
|
|
if len(self.delivery_ids) == 1:
|
|
action.update({
|
|
'view_mode': 'form',
|
|
'res_id': self.delivery_ids[0].id
|
|
})
|
|
else:
|
|
action.update({
|
|
'name': _("Delivery orders of %s", self.display_name),
|
|
'domain': [('id', 'in', self.delivery_ids.ids)],
|
|
'view_mode': 'list,form'
|
|
})
|
|
return action
|
|
|
|
@api.model
|
|
def _get_outgoing_domain(self):
|
|
return [
|
|
'|',
|
|
'|', ('picking_code', '=', 'outgoing'), ('move_id.picking_code', '=', 'outgoing'),
|
|
('produce_line_ids', '!=', False),
|
|
]
|
|
|
|
def _find_delivery_ids_by_lot(self, lot_path=None, delivery_by_lot=None):
|
|
if lot_path is None:
|
|
lot_path = set()
|
|
domain = Domain([
|
|
('lot_id', 'in', self.ids),
|
|
('state', '=', 'done'),
|
|
]) & Domain(self._get_outgoing_domain())
|
|
move_lines = self.env['stock.move.line'].search(domain)
|
|
moves_by_lot = {
|
|
lot_id: {'producing_lines': set(), 'barren_lines': set()}
|
|
for lot_id in move_lines.lot_id.ids
|
|
}
|
|
for line in move_lines:
|
|
if line.produce_line_ids:
|
|
moves_by_lot[line.lot_id.id]['producing_lines'].add(line.id)
|
|
else:
|
|
moves_by_lot[line.lot_id.id]['barren_lines'].add(line.id)
|
|
if delivery_by_lot is None:
|
|
delivery_by_lot = dict()
|
|
for lot in self:
|
|
delivery_ids = set()
|
|
|
|
if moves_by_lot.get(lot.id):
|
|
producing_move_lines = self.env['stock.move.line'].browse(moves_by_lot[lot.id]['producing_lines'])
|
|
barren_move_lines = self.env['stock.move.line'].browse(moves_by_lot[lot.id]['barren_lines'])
|
|
|
|
if producing_move_lines:
|
|
lot_path.add(lot.id)
|
|
next_lots = producing_move_lines.produce_line_ids.lot_id.filtered(lambda l: l.id not in lot_path)
|
|
next_lots_ids = set(next_lots.ids)
|
|
# If some producing lots are in lot_path, it means that they have been previously processed.
|
|
# Their results are therefore already in delivery_by_lot and we add them to delivery_ids directly.
|
|
delivery_ids.update(*(delivery_by_lot.get(lot_id, []) for lot_id in (producing_move_lines.produce_line_ids.lot_id - next_lots).ids))
|
|
|
|
for lot_id, delivery_ids_set in next_lots._find_delivery_ids_by_lot(lot_path=lot_path, delivery_by_lot=delivery_by_lot).items():
|
|
if lot_id in next_lots_ids:
|
|
delivery_ids.update(delivery_ids_set)
|
|
delivery_ids.update(barren_move_lines.picking_id.ids)
|
|
|
|
delivery_by_lot[lot.id] = list(delivery_ids)
|
|
return delivery_by_lot
|
|
|
|
def _find_delivery_ids_by_lot_iterative(self):
|
|
""" Retrieve all delivery IDs (outgoing picking) linked to the lots
|
|
in self and all the lots found when parcouring the produce lines.
|
|
:return: A dictionary where keys are the IDs of the original 'stock.lot'
|
|
records (self) and values are lists of associated 'stock.picking' IDs.
|
|
:rtype: dict
|
|
"""
|
|
|
|
all_lot_ids = set(self.ids)
|
|
barren_lines = defaultdict(set)
|
|
parent_map = defaultdict(set)
|
|
|
|
# Prefetch the lines linked to lots and split them between producing lines
|
|
# and barren lines (lines that have `produce_line_ids` and lines that don't
|
|
# have them respectively) and build the map of the parents of each lot (so we
|
|
# can browse the tree from the leaves to the root and propagate the pickings)
|
|
queue = list(self.ids)
|
|
while queue:
|
|
domain = Domain([
|
|
('lot_id', 'in', queue),
|
|
('state', '=', 'done'),
|
|
]) & Domain(self._get_outgoing_domain())
|
|
|
|
queue = []
|
|
move_lines = self.env['stock.move.line'].search(domain)
|
|
for line in move_lines:
|
|
lot_id = line.lot_id.id
|
|
|
|
produce_line_lot_ids = line.produce_line_ids.lot_id.ids
|
|
if produce_line_lot_ids:
|
|
for child_lot_id in produce_line_lot_ids:
|
|
parent_map[child_lot_id].add(lot_id)
|
|
else:
|
|
barren_lines[lot_id].add(line.id)
|
|
|
|
next_lots = set(produce_line_lot_ids) - all_lot_ids
|
|
all_lot_ids.update(next_lots)
|
|
queue.extend(next_lots)
|
|
|
|
# Initialize delivery_by_lot with barren lines (i.e. the leaves of the lot tree)
|
|
lots_to_propagate = set()
|
|
delivery_by_lot = {lot_id: set() for lot_id in all_lot_ids}
|
|
for lot_id in barren_lines:
|
|
barren_line_ids = barren_lines[lot_id]
|
|
if barren_line_ids:
|
|
barren_move_lines = self.env['stock.move.line'].browse(barren_line_ids)
|
|
delivery_by_lot[lot_id].update(barren_move_lines.picking_id.ids)
|
|
lots_to_propagate.add(lot_id)
|
|
|
|
# Propagate the deliveries from the children to their parent lots.
|
|
# This loop processes lots whose delivery sets have just been updated,
|
|
# ensuring the new results are merged upward through the parent graph until
|
|
# all deliveries are propagated
|
|
while lots_to_propagate:
|
|
lot_id = lots_to_propagate.pop()
|
|
|
|
parent_ids = parent_map[lot_id]
|
|
for parent_id in parent_ids:
|
|
if not delivery_by_lot[lot_id].issubset(delivery_by_lot[parent_id]):
|
|
delivery_by_lot[parent_id].update(delivery_by_lot[lot_id])
|
|
lots_to_propagate.add(parent_id)
|
|
|
|
return {lot_id: list(delivery_by_lot[lot_id]) for lot_id in delivery_by_lot}
|