schuetz3/schuetz3-erpnext
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: pick list validation didn't consider existing draft pick list

Browse Source 
(cherry picked from commit 3bce4d92f6)

# Conflicts:
#	erpnext/stock/doctype/pick_list/pick_list.py
This commit is contained in:
Rohit Waghchaure
2024-04-08 18:18:45 +05:30
committed by Mergify
parent 566d4fa76e
commit 722abf1b6b
2 changed files with 279 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

252
erpnext/stock/doctype/pick_list/pick_list.py
View File

@@ -13,7 +13,7 @@ from frappe.model.mapper import map_child_doc
from frappe.query_builder import Case
from frappe.query_builder.custom import GROUP_CONCAT
from frappe.query_builder.functions import Coalesce, Locate, Replace, Sum
from frappe.utils import ceil, cint, floor, flt
from frappe.utils import ceil, cint, floor, flt, get_link_to_form
from frappe.utils.nestedset import get_descendants_of
from erpnext.selling.doctype.sales_order.sales_order import (
@@ -24,7 +24,11 @@ from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle impor
get_picked_serial_nos,
)
from erpnext.stock.get_item_details import get_conversion_factor
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
from erpnext.stock.serial_batch_bundle import (
SerialBatchCreation,
get_batches_from_bundle,
get_serial_nos_from_bundle,
)
# TODO: Prioritize SO or WO group warehouse
@@ -202,10 +206,11 @@ class PickList(Document):
row.db_set("serial_and_batch_bundle", None)
def on_update(self):
self.linked_serial_and_batch_bundle()
if self.get("locations"):
self.linked_serial_and_batch_bundle()
def linked_serial_and_batch_bundle(self):
for row in self.locations:
for row in self.get("locations"):
if row.serial_and_batch_bundle:
frappe.get_doc(
"Serial and Batch Bundle", row.serial_and_batch_bundle
@@ -518,56 +523,87 @@ class PickList(Document):
def get_picked_items_details(self, items):
picked_items = frappe._dict()
if items:
pi = frappe.qb.DocType("Pick List")
pi_item = frappe.qb.DocType("Pick List Item")
query = (
frappe.qb.from_(pi)
.inner_join(pi_item)
.on(pi.name == pi_item.parent)
.select(
pi_item.item_code,
pi_item.warehouse,
pi_item.batch_no,
pi_item.serial_and_batch_bundle,
Sum(Case().when(pi_item.picked_qty > 0, pi_item.picked_qty).else_(pi_item.stock_qty)).as_(
"picked_qty"
),
Replace(GROUP_CONCAT(pi_item.serial_no), ",", "\n").as_("serial_no"),
)
.where(
(pi_item.item_code.isin([x.item_code for x in items]))
& ((pi_item.picked_qty > 0) | (pi_item.stock_qty > 0))
& (pi.status != "Completed")
& (pi.status != "Cancelled")
& (pi_item.docstatus != 2)
)
.groupby(
pi_item.item_code,
pi_item.warehouse,
pi_item.batch_no,
)
)
if not items:
return picked_items
if self.name:
query = query.where(pi_item.parent != self.name)
items_data = self._get_pick_list_items(items)
items_data = query.run(as_dict=True)
for item_data in items_data:
key = (item_data.warehouse, item_data.batch_no) if item_data.batch_no else item_data.warehouse
serial_no = [x for x in item_data.serial_no.split("\n") if x] if item_data.serial_no else None
for item_data in items_data:
key = (item_data.warehouse, item_data.batch_no) if item_data.batch_no else item_data.warehouse
serial_no = [x for x in item_data.serial_no.split("\n") if x] if item_data.serial_no else None
data = {"picked_qty": item_data.picked_qty}
if serial_no:
data["serial_no"] = serial_no
if item_data.item_code not in picked_items:
picked_items[item_data.item_code] = {key: data}
else:
picked_items[item_data.item_code][key] = data
if item_data.serial_and_batch_bundle:
if not serial_no:
serial_no = get_serial_nos_from_bundle(item_data.serial_and_batch_bundle)
if not item_data.batch_no and not serial_no:
bundle_batches = get_batches_from_bundle(item_data.serial_and_batch_bundle)
for batch_no, batch_qty in bundle_batches.items():
batch_qty = abs(batch_qty)
key = (item_data.warehouse, batch_no)
if item_data.item_code not in picked_items:
picked_items[item_data.item_code] = {key: {"picked_qty": batch_qty}}
else:
picked_items[item_data.item_code][key]["picked_qty"] += batch_qty
continue
if item_data.item_code not in picked_items:
picked_items[item_data.item_code] = {}
if key not in picked_items[item_data.item_code]:
picked_items[item_data.item_code][key] = frappe._dict(
{
"picked_qty": 0,
"serial_no": [],
"batch_no": item_data.batch_no or "",
"warehouse": item_data.warehouse,
}
)
picked_items[item_data.item_code][key]["picked_qty"] += item_data.picked_qty
if serial_no:
picked_items[item_data.item_code][key]["serial_no"].extend(serial_no)
return picked_items
<<<<<<< HEAD
def _get_product_bundles(self) -> Dict[str, str]:
=======
def _get_pick_list_items(self, items):
pi = frappe.qb.DocType("Pick List")
pi_item = frappe.qb.DocType("Pick List Item")
query = (
frappe.qb.from_(pi)
.inner_join(pi_item)
.on(pi.name == pi_item.parent)
.select(
pi_item.item_code,
pi_item.warehouse,
pi_item.batch_no,
pi_item.serial_and_batch_bundle,
pi_item.serial_no,
(Case().when(pi_item.picked_qty > 0, pi_item.picked_qty).else_(pi_item.stock_qty)).as_(
"picked_qty"
),
)
.where(
(pi_item.item_code.isin([x.item_code for x in items]))
& ((pi_item.picked_qty > 0) | (pi_item.stock_qty > 0))
& (pi.status != "Completed")
& (pi.status != "Cancelled")
& (pi_item.docstatus != 2)
)
)
if self.name:
query = query.where(pi_item.parent != self.name)
return query.run(as_dict=True)
def _get_product_bundles(self) -> dict[str, str]:
>>>>>>> 3bce4d92f6 (fix: pick list validation didn't consider existing draft pick list)
# Dict[so_item_row: item_code]
product_bundles = {}
for item in self.locations:
@@ -725,9 +761,7 @@ def get_available_item_locations(
consider_rejected_warehouses=False,
):
locations = []
total_picked_qty = (
sum([v.get("picked_qty") for k, v in picked_item_details.items()]) if picked_item_details else 0
)
has_serial_no = frappe.get_cached_value("Item", item_code, "has_serial_no")
has_batch_no = frappe.get_cached_value("Item", item_code, "has_batch_no")
@@ -737,63 +771,90 @@ def get_available_item_locations(
from_warehouses,
required_qty,
company,
total_picked_qty,
consider_rejected_warehouses=consider_rejected_warehouses,
)
elif has_serial_no:
locations = get_available_item_locations_for_serialized_item(
item_code,
from_warehouses,
required_qty,
company,
total_picked_qty,
consider_rejected_warehouses=consider_rejected_warehouses,
)
elif has_batch_no:
locations = get_available_item_locations_for_batched_item(
item_code,
from_warehouses,
required_qty,
company,
total_picked_qty,
consider_rejected_warehouses=consider_rejected_warehouses,
)
else:
locations = get_available_item_locations_for_other_item(
item_code,
from_warehouses,
required_qty,
company,
total_picked_qty,
consider_rejected_warehouses=consider_rejected_warehouses,
)
if picked_item_details:
locations = filter_locations_by_picked_materials(locations, picked_item_details)
if locations:
locations = get_locations_based_on_required_qty(locations, required_qty)
if not ignore_validation:
validate_picked_materials(item_code, required_qty, locations)
return locations
def get_locations_based_on_required_qty(locations, required_qty):
filtered_locations = []
for location in locations:
if location.qty >= required_qty:
location.qty = required_qty
filtered_locations.append(location)
break
required_qty -= location.qty
filtered_locations.append(location)
return filtered_locations
def validate_picked_materials(item_code, required_qty, locations):
for location in list(locations):
if location["qty"] < 0:
locations.remove(location)
total_qty_available = sum(location.get("qty") for location in locations)
remaining_qty = required_qty - total_qty_available
if remaining_qty > 0 and not ignore_validation:
if remaining_qty > 0:
frappe.msgprint(
_("{0} units of Item {1} is not available.").format(
remaining_qty, frappe.get_desk_link("Item", item_code)
_("{0} units of Item {1} is picked in another Pick List.").format(
remaining_qty, get_link_to_form("Item", item_code)
),
title=_("Insufficient Stock"),
title=_("Already Picked"),
)
if picked_item_details:
for location in list(locations):
if location["qty"] < 0:
locations.remove(location)
total_qty_available = sum(location.get("qty") for location in locations)
remaining_qty = required_qty - total_qty_available
def filter_locations_by_picked_materials(locations, picked_item_details) -> list[dict]:
for row in locations:
key = row.warehouse
if row.batch_no:
key = (row.warehouse, row.batch_no)
if remaining_qty > 0 and not ignore_validation:
frappe.msgprint(
_("{0} units of Item {1} is picked in another Pick List.").format(
remaining_qty, frappe.get_desk_link("Item", item_code)
),
title=_("Already Picked"),
)
picked_qty = picked_item_details.get(key, {}).get("picked_qty", 0)
if not picked_qty:
continue
if picked_qty > row.qty:
row.qty = 0
picked_item_details[key]["picked_qty"] -= row.qty
else:
row.qty -= picked_qty
picked_item_details[key]["picked_qty"] = 0.0
if row.serial_nos:
row.serial_nos = list(set(row.serial_nos) - set(picked_item_details[key].get("serial_no")))
return locations
@@ -803,15 +864,12 @@ def get_available_item_locations_for_serial_and_batched_item(
from_warehouses,
required_qty,
company,
total_picked_qty=0,
consider_rejected_warehouses=False,
):
# Get batch nos by FIFO
locations = get_available_item_locations_for_batched_item(
item_code,
from_warehouses,
required_qty,
company,
consider_rejected_warehouses=consider_rejected_warehouses,
)
@@ -831,7 +889,6 @@ def get_available_item_locations_for_serial_and_batched_item(
(conditions) & (sn.batch_no == location.batch_no) & (sn.warehouse == location.warehouse)
)
.orderby(sn.creation)
.limit(ceil(location.qty + total_picked_qty))
).run(as_dict=True)
serial_nos = [sn.name for sn in serial_nos]
@@ -844,18 +901,14 @@ def get_available_item_locations_for_serial_and_batched_item(
def get_available_item_locations_for_serialized_item(
item_code,
from_warehouses,
required_qty,
company,
total_picked_qty=0,
consider_rejected_warehouses=False,
):
picked_serial_nos = get_picked_serial_nos(item_code, from_warehouses)
sn = frappe.qb.DocType("Serial No")
query = (
frappe.qb.from_(sn)
.select(sn.name, sn.warehouse)
.where((sn.item_code == item_code) & (sn.company == company))
.where(sn.item_code == item_code)
.orderby(sn.creation)
)
@@ -863,6 +916,7 @@ def get_available_item_locations_for_serialized_item(
query = query.where(sn.warehouse.isin(from_warehouses))
else:
query = query.where(Coalesce(sn.warehouse, "") != "")
query = query.where(sn.company == company)
if not consider_rejected_warehouses:
if rejected_warehouses := get_rejected_warehouses():
@@ -871,16 +925,8 @@ def get_available_item_locations_for_serialized_item(
serial_nos = query.run(as_list=True)
warehouse_serial_nos_map = frappe._dict()
picked_qty = required_qty
for serial_no, warehouse in serial_nos:
if serial_no in picked_serial_nos:
continue
if picked_qty <= 0:
break
warehouse_serial_nos_map.setdefault(warehouse, []).append(serial_no)
picked_qty -= 1
locations = []
@@ -888,12 +934,14 @@ def get_available_item_locations_for_serialized_item(
qty = len(serial_nos)
locations.append(
{
"qty": qty,
"warehouse": warehouse,
"item_code": item_code,
"serial_nos": serial_nos,
}
frappe._dict(
{
"qty": qty,
"warehouse": warehouse,
"item_code": item_code,
"serial_nos": serial_nos,
}
)
)
return locations
@@ -902,9 +950,6 @@ def get_available_item_locations_for_serialized_item(
def get_available_item_locations_for_batched_item(
item_code,
from_warehouses,
required_qty,
company,
total_picked_qty=0,
consider_rejected_warehouses=False,
):
locations = []
@@ -913,8 +958,6 @@ def get_available_item_locations_for_batched_item(
{
"item_code": item_code,
"warehouse": from_warehouses,
"qty": required_qty,
"is_pick_list": True,
}
)
)
@@ -952,9 +995,7 @@ def get_available_item_locations_for_batched_item(
def get_available_item_locations_for_other_item(
item_code,
from_warehouses,
required_qty,
company,
total_picked_qty=0,
consider_rejected_warehouses=False,
):
bin = frappe.qb.DocType("Bin")
@@ -963,7 +1004,6 @@ def get_available_item_locations_for_other_item(
.select(bin.warehouse, bin.actual_qty.as_("qty"))
.where((bin.item_code == item_code) & (bin.actual_qty > 0))
.orderby(bin.creation)
.limit(cint(required_qty + total_picked_qty))
)
if from_warehouses:

134
erpnext/stock/doctype/pick_list/test_pick_list.py
View File

@@ -818,7 +818,7 @@ class TestPickList(FrappeTestCase):
def test_pick_list_status(self):
warehouse = "_Test Warehouse - _TC"
item = make_item(properties={"maintain_stock": 1}).name
item = make_item(properties={"is_stock_item": 1}).name
make_stock_entry(item=item, to_warehouse=warehouse, qty=10)
so = make_sales_order(item_code=item, qty=10, rate=100)
@@ -848,3 +848,135 @@ class TestPickList(FrappeTestCase):
pl.cancel()
pl.reload()
self.assertEqual(pl.status, "Cancelled")
def test_pick_list_validation(self):
warehouse = "_Test Warehouse - _TC"
item = make_item("Test Non Serialized Pick List Item", properties={"is_stock_item": 1}).name
make_stock_entry(item=item, to_warehouse=warehouse, qty=10)
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.save()
pl.submit()
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.save()
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=4, rate=100)
pl = create_pick_list(so.name)
self.assertFalse(hasattr(pl, "locations"))
def test_pick_list_validation_for_serial_no(self):
warehouse = "_Test Warehouse - _TC"
item = make_item(
"Test Serialized Pick List Item",
properties={"is_stock_item": 1, "has_serial_no": 1, "serial_no_series": "SN-SPLI-.####"},
).name
make_stock_entry(item=item, to_warehouse=warehouse, qty=10)
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.locations[0].qty = 5
pl.save()
pl.submit()
self.assertTrue(pl.locations[0].serial_no)
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.save()
self.assertTrue(pl.locations[0].serial_no)
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=4, rate=100)
pl = create_pick_list(so.name)
self.assertFalse(hasattr(pl, "locations"))
def test_pick_list_validation_for_batch_no(self):
warehouse = "_Test Warehouse - _TC"
item = make_item(
"Test Batch Pick List Item",
properties={
"is_stock_item": 1,
"has_batch_no": 1,
"batch_number_series": "BATCH-SPLI-.####",
"create_new_batch": 1,
},
).name
make_stock_entry(item=item, to_warehouse=warehouse, qty=10)
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.locations[0].qty = 5
pl.save()
pl.submit()
self.assertTrue(pl.locations[0].batch_no)
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.save()
self.assertTrue(pl.locations[0].batch_no)
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=4, rate=100)
pl = create_pick_list(so.name)
self.assertFalse(hasattr(pl, "locations"))
def test_pick_list_validation_for_batch_no_and_serial_item(self):
warehouse = "_Test Warehouse - _TC"
item = make_item(
"Test Serialized Batch Pick List Item",
properties={
"is_stock_item": 1,
"has_batch_no": 1,
"batch_number_series": "SN-BT-BATCH-SPLI-.####",
"create_new_batch": 1,
"has_serial_no": 1,
"serial_no_series": "SN-BT-SPLI-.####",
},
).name
make_stock_entry(item=item, to_warehouse=warehouse, qty=10)
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.locations[0].qty = 5
pl.save()
pl.submit()
self.assertTrue(pl.locations[0].batch_no)
self.assertTrue(pl.locations[0].serial_no)
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=5, rate=100)
pl = create_pick_list(so.name)
pl.save()
self.assertTrue(pl.locations[0].batch_no)
self.assertTrue(pl.locations[0].serial_no)
self.assertEqual(pl.locations[0].qty, 5.0)
self.assertTrue(hasattr(pl, "locations"))
so = make_sales_order(item_code=item, qty=4, rate=100)
pl = create_pick_list(so.name)
self.assertFalse(hasattr(pl, "locations"))