nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-07-20 11:22:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: auto fetch serial no from previous operation output (backport to v16) (#56861)

Browse Source
This commit is contained in:
rohitwaghchaure
2026-07-03 23:02:05 +05:30
committed by GitHub
parent 1da28f2278
commit 17f2de42f3
5 changed files with 450 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
erpnext/manufacturing/doctype/job_card/job_card.py
View File

@@ -1720,7 +1720,7 @@ def make_material_request(source_name, target_doc=None):
@frappe.whitelist()
def make_stock_entry(source_name, target_doc=None):
def make_stock_entry(source_name: str, target_doc: Document | str | None = None):
def update_item(source, target, source_parent):
target.t_warehouse = source_parent.wip_warehouse
@@ -1752,6 +1752,8 @@ def make_stock_entry(source_name, target_doc=None):
target.set_missing_values()
target.set_stock_entry_type()
from erpnext.stock.doctype.stock_entry.stock_entry import set_previous_operation_serial_batch
wo_allows_alternate_item = frappe.db.get_value(
"Work Order", target.work_order, "allow_alternative_item"
)
@@ -1760,6 +1762,7 @@ def make_stock_entry(source_name, target_doc=None):
wo_allows_alternate_item
and frappe.get_cached_value("Item", item.item_code, "allow_alternative_item")
)
set_previous_operation_serial_batch(target, item)
doclist = get_mapped_doc(
"Job Card",

291
erpnext/manufacturing/doctype/job_card/test_job_card.py
View File

@@ -1057,6 +1057,9 @@ class TestJobCard(ERPNextTestSuite):
job_card.submit()
for row in fg_bom.items:
if row.item_code == sfg.name:
continue
make_stock_entry(
item_code=row.item_code,
target="Stores - _TC",
@@ -1067,9 +1070,295 @@ class TestJobCard(ERPNextTestSuite):
manufacturing_entry = frappe.get_doc(job_card.make_stock_entry_for_semi_fg_item())
manufacturing_entry.submit()
sfg_row = next(row for row in manufacturing_entry.items if row.item_code == sfg.name)
self.assertEqual(flt(sfg_row.basic_rate, 3), 95.0)
self.assertEqual(manufacturing_entry.items[2].item_code, scrap2.name)
self.assertEqual(manufacturing_entry.items[2].qty, 9)
self.assertEqual(flt(manufacturing_entry.items[2].basic_rate, 3), 5.556)
self.assertEqual(flt(manufacturing_entry.items[2].basic_rate, 3), 5.278)
def test_semi_fg_batch_auto_pull_on_manufacture(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
from erpnext.stock.serial_batch_bundle import get_batches_from_bundle
frappe.db.set_value("UOM", "Nos", "must_be_whole_number", 0)
frappe.db.set_single_value("Manufacturing Settings", "make_serial_no_batch_from_work_order", 0)
warehouse = "Stores - _TC"
rm1 = make_item("Auto Pull RM 1", {"is_stock_item": 1}).name
rm2 = make_item("Auto Pull RM 2", {"is_stock_item": 1}).name
fg1 = make_item("Auto Pull FG 1", {"is_stock_item": 1}).name
sfg = make_item(
"Auto Pull SFG 1",
{
"is_stock_item": 1,
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "AP-SFG-.#####",
},
).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm1, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg1,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
fg_bom.append("items", {"item_code": rm2, "qty": 1})
operation1 = {
"operation": "Auto Pull Op A",
"workstation": "_Test Workstation A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"finished_good_qty": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
operation2 = {
"operation": "Auto Pull Op B",
"workstation": "_Test Workstation A",
"finished_good": fg1,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 2,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation1)
make_operation(operation1)
make_operation(operation2)
fg_bom.append("operations", operation1)
fg_bom.append("operations", operation2)
fg_bom.append("items", {"item_code": sfg, "qty": 1, "uom": "Nos", "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg1,
qty=5,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
work_order.operations[0].time_in_mins = 60
work_order.operations[1].time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm1, target=warehouse, qty=10, basic_rate=100)
make_stock_entry(item_code=rm2, target=warehouse, qty=10, basic_rate=100)
jc_a = frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card", {"work_order": work_order.name, "operation": "Auto Pull Op A"}, "name"
),
)
jc_a.append(
"time_logs",
{
"from_time": "2024-01-01 08:00:00",
"to_time": "2024-01-01 09:00:00",
"completed_qty": jc_a.for_quantity,
},
)
jc_a.submit()
me_a = frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item())
me_a.submit()
me_a.reload()
sfg_fg_row = next(r for r in me_a.items if r.is_finished_item and r.item_code == sfg)
self.assertTrue(sfg_fg_row.serial_and_batch_bundle)
produced_batches = get_batches_from_bundle(sfg_fg_row.serial_and_batch_bundle)
jc_b = frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card", {"work_order": work_order.name, "operation": "Auto Pull Op B"}, "name"
),
)
jc_b.append(
"time_logs",
{
"from_time": "2024-02-01 08:00:00",
"to_time": "2024-02-01 09:00:00",
"completed_qty": jc_b.for_quantity,
},
)
jc_b.submit()
me_b = frappe.get_doc(jc_b.make_stock_entry_for_semi_fg_item())
sfg_consume_row = next(r for r in me_b.items if r.item_code == sfg and r.s_warehouse)
self.assertTrue(
sfg_consume_row.serial_and_batch_bundle,
"Previous operation's batch was not auto-pulled into the semi-finished consumption row",
)
consumed_batches = get_batches_from_bundle(sfg_consume_row.serial_and_batch_bundle)
self.assertEqual(set(consumed_batches.keys()), set(produced_batches.keys()))
def test_semi_fg_auto_pull_with_uom_conversion(self):
from erpnext.manufacturing.doctype.operation.test_operation import make_operation
from erpnext.stock.doctype.item.test_item import make_item
from erpnext.stock.doctype.stock_entry.stock_entry import set_previous_operation_serial_batch
from erpnext.stock.serial_batch_bundle import get_batches_from_bundle
frappe.db.set_value("UOM", "Nos", "must_be_whole_number", 0)
frappe.db.set_single_value("Manufacturing Settings", "make_serial_no_batch_from_work_order", 0)
warehouse = "Stores - _TC"
rm1 = make_item("UOM Pull RM 1", {"is_stock_item": 1}).name
rm2 = make_item("UOM Pull RM 2", {"is_stock_item": 1}).name
fg1 = make_item("UOM Pull FG 1", {"is_stock_item": 1}).name
sfg = make_item(
"UOM Pull SFG 1",
{
"is_stock_item": 1,
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "UP-SFG-.#####",
"uoms": [{"uom": "Box", "conversion_factor": 5}],
},
).name
sfg_bom = frappe.new_doc("BOM", company="_Test Company", item=sfg, quantity=1)
sfg_bom.append("items", {"item_code": rm1, "qty": 1})
sfg_bom.insert()
sfg_bom.submit()
fg_bom = frappe.new_doc(
"BOM",
company="_Test Company",
item=fg1,
quantity=1,
with_operations=1,
track_semi_finished_goods=1,
)
fg_bom.append("items", {"item_code": rm2, "qty": 1})
operation1 = {
"operation": "UOM Pull Op A",
"workstation": "_Test Workstation A",
"finished_good": sfg,
"bom_no": sfg_bom.name,
"finished_good_qty": 1,
"sequence_id": 1,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
operation2 = {
"operation": "UOM Pull Op B",
"workstation": "_Test Workstation A",
"finished_good": fg1,
"finished_good_qty": 1,
"is_final_finished_good": 1,
"sequence_id": 2,
"time_in_mins": 60,
"source_warehouse": warehouse,
"fg_warehouse": warehouse,
"skip_material_transfer": 1,
}
make_workstation(operation1)
make_operation(operation1)
make_operation(operation2)
fg_bom.append("operations", operation1)
fg_bom.append("operations", operation2)
fg_bom.append("items", {"item_code": sfg, "qty": 1, "uom": "Nos", "operation_row_id": 2})
fg_bom.insert()
fg_bom.submit()
work_order = make_wo_order_test_record(
item=fg1,
qty=5,
source_warehouse=warehouse,
fg_warehouse=warehouse,
bom_no=fg_bom.name,
skip_transfer=1,
do_not_save=True,
)
work_order.operations[0].time_in_mins = 60
work_order.operations[1].time_in_mins = 60
work_order.save()
work_order.submit()
make_stock_entry(item_code=rm1, target=warehouse, qty=10, basic_rate=100)
make_stock_entry(item_code=sfg, target=warehouse, qty=5, basic_rate=100, posting_date="2024-01-01")
jc_a = frappe.get_doc(
"Job Card",
frappe.db.get_value(
"Job Card", {"work_order": work_order.name, "operation": "UOM Pull Op A"}, "name"
),
)
jc_a.append(
"time_logs",
{
"from_time": "2024-02-01 08:00:00",
"to_time": "2024-02-01 09:00:00",
"completed_qty": jc_a.for_quantity,
},
)
jc_a.submit()
me_a = frappe.get_doc(jc_a.make_stock_entry_for_semi_fg_item())
me_a.submit()
me_a.reload()
sfg_fg_row = next(r for r in me_a.items if r.is_finished_item and r.item_code == sfg)
produced_batches = get_batches_from_bundle(sfg_fg_row.serial_and_batch_bundle)
se = frappe.new_doc("Stock Entry")
se.company = "_Test Company"
se.purpose = "Material Transfer"
se.work_order = work_order.name
se.set_stock_entry_type()
row = se.append(
"items",
{
"item_code": sfg,
"qty": 1,
"uom": "Box",
"conversion_factor": 5,
"s_warehouse": warehouse,
"t_warehouse": "_Test Warehouse - _TC",
},
)
set_previous_operation_serial_batch(se, row)
self.assertTrue(row.serial_and_batch_bundle)
self.assertEqual(
abs(frappe.db.get_value("Serial and Batch Bundle", row.serial_and_batch_bundle, "total_qty")),
5.0,
)
se.save()
se.submit()
se.reload()
row = se.items[0]
consumed_batches = get_batches_from_bundle(row.serial_and_batch_bundle)
self.assertEqual(set(consumed_batches.keys()), set(produced_batches.keys()))
self.assertEqual(abs(sum(consumed_batches.values())), 5.0)
def test_secondary_items_without_sfg(self):
for row in frappe.get_doc("BOM", self.work_order.bom_no).items:

18
erpnext/manufacturing/doctype/work_order/work_order.py
View File

@@ -266,6 +266,7 @@ class WorkOrder(Document):
self.validate_sales_order()
self.set_default_warehouse()
self.set_operation_warehouses()
self.validate_warehouse_belongs_to_company()
self.check_wip_warehouse_skip()
self.calculate_operating_cost()
@@ -1410,6 +1411,23 @@ class WorkOrder(Document):
self.set("operations", operations)
self.calculate_time()
self.set_operation_warehouses()
def set_operation_warehouses(self):
if not self.track_semi_finished_goods or not self.operations:
return
operations = self.operations
last_idx = len(operations) - 1
for idx, op in enumerate(operations):
if not op.source_warehouse:
op.source_warehouse = self.source_warehouse
if not op.fg_warehouse:
op.fg_warehouse = self.fg_warehouse if idx == last_idx else self.source_warehouse
if not op.wip_warehouse:
op.wip_warehouse = self.wip_warehouse
def calculate_time(self):
for d in self.get("operations"):

133
erpnext/stock/doctype/stock_entry/stock_entry.py
View File

@@ -4727,3 +4727,136 @@ def get_transferred_qty(material_request):
).run(as_dict=True)
return query[0]
def get_previous_operation_output_sn_batch(work_order, item_code, warehouse):
"""Serial nos / batches that an earlier operation produced for ``item_code`` (a
semi-finished good) and are still available in ``warehouse`` -- i.e. produced by a
prior operation's Manufacture entry minus whatever later entries already pulled out
of that warehouse. Returns an empty result for ordinary raw materials."""
result = frappe._dict(serial_nos=[], batches=defaultdict(float))
if not (work_order and item_code and warehouse):
return result
if not frappe.db.exists("Work Order Operation", {"parent": work_order, "finished_good": item_code}):
return result
item_details = frappe.get_cached_value("Item", item_code, ["has_serial_no", "has_batch_no"], as_dict=1)
if not item_details or not (item_details.has_serial_no or item_details.has_batch_no):
return result
produced = _get_operation_sn_batch(work_order, item_code, warehouse, produced=True)
consumed = _get_operation_sn_batch(work_order, item_code, warehouse, produced=False)
for serial_no in produced.serial_nos:
if serial_no not in consumed.serial_nos:
result.serial_nos.append(serial_no)
for batch_no, qty in produced.batches.items():
available = flt(qty) - flt(consumed.batches.get(batch_no))
if available > 0:
result.batches[batch_no] = available
return result
def _get_operation_sn_batch(work_order, item_code, warehouse, produced=True):
bundles = _get_operation_bundles(work_order, item_code, warehouse, produced)
result = frappe._dict(serial_nos=[], batches=defaultdict(float))
if not bundles:
return result
sbe = frappe.qb.DocType("Serial and Batch Entry")
entries = (
frappe.qb.from_(sbe)
.select(sbe.serial_no, sbe.batch_no, sbe.qty)
.where((sbe.parent.isin(bundles)) & (sbe.is_cancelled == 0))
.orderby(sbe.parent)
.orderby(sbe.idx)
).run(as_dict=True)
for row in entries:
if row.serial_no:
result.serial_nos.append(row.serial_no)
if row.batch_no:
result.batches[row.batch_no] += abs(flt(row.qty))
return result
def _get_operation_bundles(work_order, item_code, warehouse, produced):
se = frappe.qb.DocType("Stock Entry")
sed = frappe.qb.DocType("Stock Entry Detail")
warehouse_field = sed.t_warehouse if produced else sed.s_warehouse
query = (
frappe.qb.from_(se)
.inner_join(sed)
.on(sed.parent == se.name)
.select(sed.serial_and_batch_bundle)
.where(
(se.work_order == work_order)
& (se.docstatus == 1)
& (sed.item_code == item_code)
& (warehouse_field == warehouse)
& (sed.serial_and_batch_bundle.isnotnull())
)
)
if produced:
query = query.where((se.purpose == "Manufacture") & (sed.is_finished_item == 1))
return [row[0] for row in query.run()]
def _cap_pool_to_qty(pool, qty):
"""Trim the available serial/batch pool to at most ``qty`` (fill what's available)."""
serial_nos, batches = [], frappe._dict()
if pool.serial_nos:
serial_nos = pool.serial_nos[: cint(qty)]
elif pool.batches:
remaining = flt(qty)
for batch_no, batch_qty in pool.batches.items():
if remaining <= 0:
break
use = min(flt(batch_qty), remaining)
batches[batch_no] = use
remaining -= use
return serial_nos, batches
def set_previous_operation_serial_batch(parent_doc, row):
"""Auto-pull serial nos / batches produced by a previous operation onto a
consumption / transfer-out ``row`` of a Stock Entry, filling what is available and
leaving any shortfall blank for the user. No-op for ordinary raw materials or when
the row already carries serial/batch."""
warehouse = row.get("s_warehouse") or row.get("from_warehouse")
qty = flt(row.get("qty")) * flt(row.get("conversion_factor") or 1)
if not parent_doc.get("work_order") or not warehouse or qty <= 0:
return
if row.get("serial_and_batch_bundle") or row.get("serial_no") or row.get("batch_no"):
return
pool = get_previous_operation_output_sn_batch(parent_doc.work_order, row.item_code, warehouse)
serial_nos, batches = _cap_pool_to_qty(pool, qty)
if not serial_nos and not batches:
return
bundle = SerialBatchCreation(
{
"item_code": row.item_code,
"warehouse": warehouse,
"posting_datetime": get_combine_datetime(parent_doc.posting_date, parent_doc.posting_time),
"voucher_type": "Stock Entry",
"company": parent_doc.company,
"type_of_transaction": "Outward",
"qty": flt(qty),
"serial_nos": serial_nos,
"batches": batches,
"do_not_submit": True,
}
).make_serial_and_batch_bundle()
if bundle and bundle.get("name"):
row.serial_and_batch_bundle = bundle.name
row.use_serial_batch_fields = 0

8
erpnext/stock/doctype/stock_entry_type/stock_entry_type.py
View File

@@ -106,6 +106,8 @@ class ManufactureEntry:
)
def add_raw_materials(self):
from erpnext.stock.doctype.stock_entry.stock_entry import set_previous_operation_serial_batch
if self.job_card:
item_dict = {}
if not item_dict:
@@ -127,9 +129,7 @@ class ManufactureEntry:
_dict.from_warehouse = self.source_wh.get(item_code) or self.wip_warehouse
_dict.to_warehouse = ""
if backflush_based_on != "BOM" and not frappe.db.get_value(
"Job Card", self.job_card, "skip_material_transfer"
):
if backflush_based_on != "BOM" and not self.skip_material_transfer:
calculated_qty = flt(_dict.transferred_qty) - flt(_dict.consumed_qty)
if calculated_qty < 0:
frappe.throw(
@@ -138,6 +138,8 @@ class ManufactureEntry:
_dict.qty = calculated_qty
self.update_available_serial_batches(_dict, available_serial_batches)
elif self.skip_material_transfer:
set_previous_operation_serial_batch(self.stock_entry, _dict)
self.stock_entry.add_to_stock_entry_detail(item_dict)