87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
# The COPYRIGHT file at the top level of this repository contains the full
|
|
# copyright notices and license terms.
|
|
from trytond.pool import Pool, PoolMeta
|
|
from trytond.transaction import Transaction
|
|
|
|
|
|
class Lot(metaclass=PoolMeta):
|
|
__name__ = 'stock.lot'
|
|
|
|
@staticmethod
|
|
def _get_fifo_search_order_by():
|
|
pool = Pool()
|
|
Lot = pool.get('stock.lot')
|
|
|
|
order = []
|
|
if hasattr(Lot, 'shelf_life_expiration_date'):
|
|
order.append(('shelf_life_expiration_date', 'ASC'))
|
|
order.append(('expiration_date', 'ASC'))
|
|
if hasattr(Lot, 'lot_date'):
|
|
order.append(('lot_date', 'ASC'))
|
|
order.append(('create_date', 'ASC'))
|
|
return order
|
|
|
|
|
|
class Move(metaclass=PoolMeta):
|
|
__name__ = 'stock.move'
|
|
|
|
def check_lot(self):
|
|
ShipmentIn = Pool().get('stock.shipment.in')
|
|
if (not self.product.lot_force_assign
|
|
or (
|
|
(not self.origin or isinstance(self.shipment, ShipmentIn))
|
|
and self.from_location.type == 'supplier')):
|
|
return super().check_lot()
|
|
|
|
@classmethod
|
|
def assign_try(cls, moves, with_childs=True, grouping=('product',)):
|
|
not_lot_moves = []
|
|
date2lot_moves = {}
|
|
for move in moves:
|
|
if move.lot or move.product.lot_is_required(move.from_location,
|
|
move.to_location):
|
|
date2lot_moves.setdefault(
|
|
move.effective_date or move.planned_date, []).append(move)
|
|
else:
|
|
not_lot_moves.append(move)
|
|
|
|
not_lot_success, lot_success = True, True
|
|
if not_lot_moves:
|
|
not_lot_success = super().assign_try(
|
|
not_lot_moves, with_childs=with_childs, grouping=grouping)
|
|
|
|
grouping = grouping + ('lot',)
|
|
# TODO 6.2: Grouping not necessary in context.
|
|
for date_, lot_moves in date2lot_moves.items():
|
|
with Transaction().set_context(
|
|
assign_grouping=grouping, fifo_assign_date=date_):
|
|
lot_success &= super().assign_try(lot_moves,
|
|
with_childs=with_childs, grouping=grouping)
|
|
|
|
return not_lot_success & lot_success
|
|
|
|
def pick_product(self, quantities):
|
|
# TODO 6.2: Move this sorting to the method created for it
|
|
# def sort_quantities(self, quantities, locations, grouping):
|
|
pool = Pool()
|
|
Lot = pool.get('stock.lot')
|
|
|
|
transaction = Transaction()
|
|
grouping = transaction.context.get('assign_grouping')
|
|
if not quantities or not grouping:
|
|
return super().pick_product(quantities)
|
|
|
|
lot_ids = set([key[2] for key, _ in quantities if key[2]])
|
|
lot_id2order = {None: len(lot_ids)}
|
|
if lot_ids:
|
|
cursor = transaction.connection.cursor()
|
|
cursor.execute(*Lot.search([
|
|
('id', 'in', lot_ids)
|
|
], order=Lot._get_fifo_search_order_by(), query=True))
|
|
lot_id2order.update({row[0]: order
|
|
for order, row in enumerate(list(cursor.fetchall()))})
|
|
quantities = sorted(
|
|
quantities, key=lambda key_qty: lot_id2order.get(key_qty[0][2]))
|
|
|
|
return super().pick_product(quantities)
|