From 58f24c83c0916e7cf9e8a2796ae2149b6bb157cf Mon Sep 17 00:00:00 2001 From: Sudharsanan11 Date: Tue, 26 May 2026 18:21:40 +0530 Subject: [PATCH] fix(stock): add validation for work order seial nos and batch nos --- .../stock_entry_handler/manufacturing.py | 135 +++++++++++++++--- 1 file changed, 112 insertions(+), 23 deletions(-) diff --git a/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py b/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py index 231664f954d..79dafd69d9f 100644 --- a/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py +++ b/erpnext/stock/doctype/stock_entry/stock_entry_handler/manufacturing.py @@ -7,10 +7,13 @@ from frappe.query_builder.functions import Sum from frappe.utils import ceil, cint, flt, get_link_to_form from erpnext.manufacturing.doctype.bom.bom import add_additional_cost +from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos from erpnext.stock.serial_batch_bundle import ( SerialBatchCreation, get_batch_nos, + get_batches_from_bundle, get_empty_batches_based_work_order, + get_serial_nos_from_bundle, ) from .base import BaseStockEntry @@ -165,25 +168,30 @@ class BaseManufactureStockEntry(BaseStockEntry): else: self.doc.append("items", item_details) - def set_serial_nos_for_finished_good(self, item_details): + def set_serial_nos_for_finished_good(self, item_details, existing_row=None): serial_nos = self.get_available_serial_nos_for_fg(item_details.item_code) - if serial_nos: - row = frappe._dict({"serial_nos": serial_nos[0 : cint(item_details.qty)]}) + if not serial_nos: + return - _id = create_serial_and_batch_bundle( - self.doc, - row, - frappe._dict( - { - "item_code": item_details.item_code, - "warehouse": item_details.t_warehouse, - } - ), - ) + row = frappe._dict({"serial_nos": serial_nos[0 : cint(item_details.qty)]}) + _id = create_serial_and_batch_bundle( + self.doc, + row, + frappe._dict( + { + "item_code": item_details.item_code, + "warehouse": item_details.t_warehouse, + } + ), + ) + + if existing_row: + existing_row.serial_and_batch_bundle = _id + existing_row.use_serial_batch_fields = 0 + else: item_details.serial_and_batch_bundle = _id item_details.use_serial_batch_fields = 0 - self.doc.append("items", item_details) def get_available_serial_nos_for_fg(self, item_code) -> list[str]: @@ -199,22 +207,23 @@ class BaseManufactureStockEntry(BaseStockEntry): order_by="creation asc", ) - def set_batchwise_finished_goods(self, item_details): - batches = get_empty_batches_based_work_order(self.doc.work_order, self.doc.pro_doc.production_item) + def set_batchwise_finished_goods(self, item_details, existing_row=None): + batches = get_empty_batches_based_work_order(self.doc.work_order, self.wo_doc.production_item) if not batches: - self.doc.append("items", item_details) + if not existing_row: + self.doc.append("items", item_details) else: - self.add_batchwise_finished_good(batches, item_details) + self.add_batchwise_finished_good(batches, item_details, existing_row=existing_row) - def add_batchwise_finished_good(self, batches, item_details): + def add_batchwise_finished_good(self, batches, item_details, existing_row=None): qty = flt(self.doc.fg_completed_qty) row = frappe._dict({"batches_to_be_consume": defaultdict(float)}) self.update_batches_to_be_consume(batches, row, qty) if row.batches_to_be_consume: - self._link_fg_bundle_and_append(item_details, row) + self._link_fg_bundle_and_append(item_details, row, existing_row=existing_row) - def _link_fg_bundle_and_append(self, item_details, row): + def _link_fg_bundle_and_append(self, item_details, row, existing_row=None): _id = create_serial_and_batch_bundle( self.doc, row, @@ -222,8 +231,13 @@ class BaseManufactureStockEntry(BaseStockEntry): {"item_code": self.wo_doc.production_item, "warehouse": item_details.get("t_warehouse")} ), ) - item_details["serial_and_batch_bundle"] = _id - self.doc.append("items", item_details) + if existing_row: + existing_row.serial_and_batch_bundle = _id + existing_row.use_serial_batch_fields = 0 + else: + item_details["serial_and_batch_bundle"] = _id + item_details["use_serial_batch_fields"] = 0 + self.doc.append("items", item_details) def update_batches_to_be_consume(self, batches, row, qty): qty_to_be_consumed = qty @@ -253,6 +267,81 @@ class ManufactureStockEntry(BaseManufactureStockEntry): self.validate_warehouse() self.validate_raw_materials_exists() self.validate_component_and_quantities() + self.validate_finished_good_serial_batch_for_work_order() + + def validate_finished_good_serial_batch_for_work_order(self): + if not ( + self.doc.work_order + and self.wo_doc + and self.wo_doc.track_semi_finished_goods != 1 + and cint( + frappe.db.get_single_value( + "Manufacturing Settings", "make_serial_no_batch_from_work_order", cache=True + ) + ) + and (self.wo_doc.has_serial_no or self.wo_doc.has_batch_no) + ): + return + + for row in self.doc.items: + if not row.is_finished_item: + continue + + if self.check_invalid_serial_batch_nos_for_finished_good_item(row): + self.reset_serial_batch_on_fg_row(row) + frappe.msgprint( + _( + "Row {0}: Serial/Batch has been reset to values linked with Work Order {1}" + " because the previously selected serial/batch does not belong to this Work Order." + ).format(row.idx, frappe.bold(self.doc.work_order)) + ) + + def check_invalid_serial_batch_nos_for_finished_good_item(self, row) -> bool: + if self.wo_doc.has_serial_no: + serial_nos = get_serial_nos(row.serial_no) if row.serial_no else [] + if not serial_nos and row.serial_and_batch_bundle: + serial_nos = get_serial_nos_from_bundle(row.serial_and_batch_bundle) + if serial_nos: + valid_serial_nos = frappe.get_all( + "Serial No", + filters={"name": ("in", serial_nos), "work_order": self.doc.work_order}, + pluck="name", + ) + return bool(set(serial_nos) - set(valid_serial_nos)) + else: + return True + + if self.wo_doc.has_batch_no: + batch_nos = [row.batch_no] if row.batch_no else [] + if not batch_nos and row.serial_and_batch_bundle: + batch_nos = list(get_batches_from_bundle(row.serial_and_batch_bundle).keys()) + if batch_nos: + valid_batch_nos = frappe.get_all( + "Batch", + filters={"name": ("in", batch_nos), "reference_name": self.doc.work_order}, + pluck="name", + ) + return bool(set(batch_nos) - set(valid_batch_nos)) + else: + return True + + def reset_serial_batch_on_fg_row(self, row): + item_details = frappe._dict( + { + "item_code": row.item_code, + "t_warehouse": row.t_warehouse, + "qty": row.qty, + } + ) + + row.serial_no = None + row.batch_no = None + row.serial_and_batch_bundle = None + + if self.wo_doc.has_serial_no: + self.set_serial_nos_for_finished_good(item_details, existing_row=row) + elif self.wo_doc.has_batch_no: + self.set_batchwise_finished_goods(item_details, existing_row=row) def set_job_card_data(self): if self.doc.job_card and not self.doc.work_order: