mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-18 08:58:43 +00:00
refactor(stock): extract SerialBatchBundleService from StockController
Move serial & batch bundle handling (creation, validation, return bundles, teardown) out of StockController into erpnext/stock/services/serial_batch_bundle.py as a delegating service. The controller keeps thin delegators for the 10 methods reached from other doctypes or run_method; the 12 internal-only helpers live in the service. make_bundle_for_material_transfer stays a module fn in stock_controller (imported by stock/serial_batch_bundle.py). Behaviour-preserving: ledger characterization snapshots and the full Serial and Batch Bundle test suite stay green.
This commit is contained in:
Nabin Hait
@@ -77,43 +77,9 @@ class StockController(AccountsController):
|
||||
self.check_zero_rate()
|
||||
|
||||
def validate_warehouse_of_sabb(self):
|
||||
if self.is_internal_transfer():
|
||||
return
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
doc_before_save = self.get_doc_before_save()
|
||||
|
||||
for row in self.items:
|
||||
if not row.get("serial_and_batch_bundle"):
|
||||
continue
|
||||
|
||||
sabb_details = frappe.db.get_value(
|
||||
"Serial and Batch Bundle",
|
||||
row.serial_and_batch_bundle,
|
||||
["type_of_transaction", "warehouse", "has_serial_no"],
|
||||
as_dict=True,
|
||||
)
|
||||
if not sabb_details:
|
||||
continue
|
||||
|
||||
if sabb_details.type_of_transaction != "Outward":
|
||||
continue
|
||||
|
||||
warehouse = row.get("warehouse") or row.get("s_warehouse")
|
||||
if sabb_details.warehouse != warehouse:
|
||||
frappe.throw(
|
||||
_(
|
||||
"Row #{0}: Warehouse {1} does not match with the warehouse {2} in Serial and Batch Bundle {3}."
|
||||
).format(row.idx, warehouse, sabb_details.warehouse, row.serial_and_batch_bundle)
|
||||
)
|
||||
|
||||
if self.doctype == "Stock Reconciliation":
|
||||
continue
|
||||
|
||||
if sabb_details.has_serial_no and doc_before_save and doc_before_save.get("items"):
|
||||
prev_row = doc_before_save.get("items", {"idx": row.idx})
|
||||
if prev_row and prev_row[0].serial_and_batch_bundle != row.serial_and_batch_bundle:
|
||||
sabb_doc = frappe.get_doc("Serial and Batch Bundle", row.serial_and_batch_bundle)
|
||||
sabb_doc.validate_serial_no_status()
|
||||
return SerialBatchBundleService(self).validate_warehouse_of_sabb()
|
||||
|
||||
def reset_conversion_factor(self):
|
||||
for row in self.get("items"):
|
||||
@@ -170,37 +136,9 @@ class StockController(AccountsController):
|
||||
frappe.throw(_("Items {0} do not exist in the Item master.").format(", ".join(non_exists_items)))
|
||||
|
||||
def validate_duplicate_serial_and_batch_bundle(self, table_name):
|
||||
if not self.get(table_name):
|
||||
return
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
sbb_list = []
|
||||
for item in self.get(table_name):
|
||||
if item.get("serial_and_batch_bundle"):
|
||||
sbb_list.append(item.get("serial_and_batch_bundle"))
|
||||
|
||||
if item.get("rejected_serial_and_batch_bundle"):
|
||||
sbb_list.append(item.get("rejected_serial_and_batch_bundle"))
|
||||
|
||||
if sbb_list:
|
||||
SLE = frappe.qb.DocType("Stock Ledger Entry")
|
||||
data = (
|
||||
frappe.qb.from_(SLE)
|
||||
.select(SLE.voucher_type, SLE.voucher_no, SLE.serial_and_batch_bundle)
|
||||
.where(
|
||||
(SLE.docstatus == 1)
|
||||
& (SLE.serial_and_batch_bundle.notnull())
|
||||
& (SLE.serial_and_batch_bundle.isin(sbb_list))
|
||||
)
|
||||
.limit(1)
|
||||
).run(as_dict=True)
|
||||
|
||||
if data:
|
||||
data = data[0]
|
||||
frappe.throw(
|
||||
_("Serial and Batch Bundle {0} is already used in {1} {2}.").format(
|
||||
frappe.bold(data.serial_and_batch_bundle), data.voucher_type, data.voucher_no
|
||||
)
|
||||
)
|
||||
return SerialBatchBundleService(self).validate_duplicate_serial_and_batch_bundle(table_name)
|
||||
|
||||
def get_item_wise_inventory_account_map(self):
|
||||
inventory_account_map = frappe._dict()
|
||||
@@ -287,405 +225,31 @@ class StockController(AccountsController):
|
||||
make_gl_entries(gl_entries, from_repost=from_repost)
|
||||
|
||||
def validate_serialized_batch(self):
|
||||
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
is_material_issue = False
|
||||
if self.doctype == "Stock Entry" and self.purpose in ["Material Issue", "Material Transfer"]:
|
||||
is_material_issue = True
|
||||
|
||||
for d in self.get("items"):
|
||||
if hasattr(d, "serial_no") and hasattr(d, "batch_no") and d.serial_no and d.batch_no:
|
||||
serial_nos = frappe.get_all(
|
||||
"Serial No",
|
||||
fields=["batch_no", "name", "warehouse"],
|
||||
filters={"name": ("in", get_serial_nos(d.serial_no))},
|
||||
)
|
||||
|
||||
for row in serial_nos:
|
||||
if row.warehouse and row.batch_no != d.batch_no:
|
||||
frappe.throw(
|
||||
_("Row #{0}: Serial No {1} does not belong to Batch {2}").format(
|
||||
d.idx, row.name, d.batch_no
|
||||
)
|
||||
)
|
||||
|
||||
if is_material_issue:
|
||||
continue
|
||||
|
||||
if flt(d.qty) > 0.0 and d.get("batch_no") and self.get("posting_date") and self.docstatus < 2:
|
||||
expiry_date = frappe.get_cached_value("Batch", d.get("batch_no"), "expiry_date")
|
||||
|
||||
if expiry_date and getdate(expiry_date) < getdate(self.posting_date):
|
||||
frappe.throw(
|
||||
_("Row #{0}: The batch {1} has already expired.").format(
|
||||
d.idx, get_link_to_form("Batch", d.get("batch_no"))
|
||||
),
|
||||
BatchExpiredError,
|
||||
)
|
||||
return SerialBatchBundleService(self).validate_serialized_batch()
|
||||
|
||||
def clean_serial_nos(self):
|
||||
from erpnext.stock.doctype.serial_no.serial_no import clean_serial_no_string
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
for row in self.get("items"):
|
||||
if hasattr(row, "serial_no") and row.serial_no:
|
||||
# remove extra whitespace and store one serial no on each line
|
||||
row.serial_no = clean_serial_no_string(row.serial_no)
|
||||
|
||||
for row in self.get("packed_items") or []:
|
||||
if hasattr(row, "serial_no") and row.serial_no:
|
||||
# remove extra whitespace and store one serial no on each line
|
||||
row.serial_no = clean_serial_no_string(row.serial_no)
|
||||
return SerialBatchBundleService(self).clean_serial_nos()
|
||||
|
||||
def make_bundle_using_old_serial_batch_fields(self, table_name=None, via_landed_cost_voucher=False):
|
||||
if self.get("_action") == "update_after_submit":
|
||||
return
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
# To handle test cases
|
||||
if frappe.in_test and frappe.flags.use_serial_and_batch_fields:
|
||||
return
|
||||
|
||||
if not table_name:
|
||||
table_name = "items"
|
||||
|
||||
if self.doctype == "Asset Capitalization":
|
||||
table_name = "stock_items"
|
||||
|
||||
parent_details = frappe._dict()
|
||||
if table_name == "packed_items":
|
||||
parent_details = self.get_parent_details_for_packed_items()
|
||||
|
||||
for row in self.get(table_name):
|
||||
if (
|
||||
not via_landed_cost_voucher
|
||||
and row.serial_and_batch_bundle
|
||||
and (row.serial_no or row.batch_no)
|
||||
):
|
||||
self.validate_serial_nos_and_batches_with_bundle(row)
|
||||
|
||||
if not row.serial_no and not row.batch_no and not row.get("rejected_serial_no"):
|
||||
continue
|
||||
|
||||
if not row.use_serial_batch_fields and (
|
||||
row.serial_no or row.batch_no or row.get("rejected_serial_no")
|
||||
):
|
||||
row.use_serial_batch_fields = 1
|
||||
|
||||
if row.use_serial_batch_fields and (
|
||||
not row.serial_and_batch_bundle and not row.get("rejected_serial_and_batch_bundle")
|
||||
):
|
||||
bundle_details = {
|
||||
"item_code": row.get("rm_item_code") or row.item_code,
|
||||
"posting_datetime": combine_datetime(self.posting_date, self.posting_time),
|
||||
"voucher_type": self.doctype,
|
||||
"voucher_no": self.name,
|
||||
"voucher_detail_no": row.name,
|
||||
"company": self.company,
|
||||
"is_rejected": 1 if row.get("rejected_warehouse") else 0,
|
||||
"use_serial_batch_fields": row.use_serial_batch_fields,
|
||||
"via_landed_cost_voucher": via_landed_cost_voucher,
|
||||
"do_not_submit": True if not via_landed_cost_voucher else False,
|
||||
}
|
||||
|
||||
if self.is_internal_transfer() and row.get("from_warehouse") and not self.is_return:
|
||||
self.update_bundle_details(bundle_details, table_name, row)
|
||||
bundle_details["type_of_transaction"] = "Outward"
|
||||
bundle_details["warehouse"] = row.get("from_warehouse")
|
||||
bundle_details["qty"] = row.get("stock_qty") or row.get("qty")
|
||||
self.create_serial_batch_bundle(bundle_details, row)
|
||||
continue
|
||||
|
||||
if row.get("qty") or row.get("consumed_qty") or row.get("stock_qty"):
|
||||
self.update_bundle_details(bundle_details, table_name, row, parent_details=parent_details)
|
||||
self.create_serial_batch_bundle(bundle_details, row)
|
||||
|
||||
if row.get("rejected_qty"):
|
||||
self.update_bundle_details(bundle_details, table_name, row, is_rejected=True)
|
||||
self.create_serial_batch_bundle(bundle_details, row)
|
||||
|
||||
def get_parent_details_for_packed_items(self):
|
||||
parent_details = frappe._dict()
|
||||
for row in self.get("items"):
|
||||
parent_details[row.name] = row
|
||||
|
||||
return parent_details
|
||||
return SerialBatchBundleService(self).make_bundle_using_old_serial_batch_fields(
|
||||
table_name, via_landed_cost_voucher
|
||||
)
|
||||
|
||||
def make_bundle_for_sales_purchase_return(self, table_name=None):
|
||||
if not self.get("is_return"):
|
||||
return
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
if not table_name:
|
||||
table_name = "items"
|
||||
|
||||
self.make_bundle_for_non_rejected_qty(table_name)
|
||||
|
||||
if self.doctype in ["Purchase Invoice", "Purchase Receipt"]:
|
||||
self.make_bundle_for_rejected_qty(table_name)
|
||||
|
||||
def make_bundle_for_rejected_qty(self, table_name=None):
|
||||
field, reference_ids = self.get_reference_ids(
|
||||
table_name, "rejected_qty", "rejected_serial_and_batch_bundle"
|
||||
)
|
||||
|
||||
if not reference_ids:
|
||||
return
|
||||
|
||||
child_doctype = self.doctype + " Item"
|
||||
available_dict = available_serial_batch_for_return(
|
||||
field, child_doctype, reference_ids, is_rejected=True
|
||||
)
|
||||
|
||||
for row in self.get(table_name):
|
||||
if data := available_dict.get(row.get(field)):
|
||||
qty_field = "rejected_qty"
|
||||
warehouse_field = "rejected_warehouse"
|
||||
if row.get("return_qty_from_rejected_warehouse"):
|
||||
qty_field = "qty"
|
||||
warehouse_field = "warehouse"
|
||||
|
||||
if not data.get("qty"):
|
||||
frappe.throw(
|
||||
_("For the {0}, no stock is available for the return in the warehouse {1}.").format(
|
||||
frappe.bold(row.item_code), row.get(warehouse_field)
|
||||
)
|
||||
)
|
||||
|
||||
data = filter_serial_batches(
|
||||
self, data, row, warehouse_field=warehouse_field, qty_field=qty_field
|
||||
)
|
||||
bundle = make_serial_batch_bundle_for_return(data, row, self, warehouse_field, qty_field)
|
||||
if row.get("return_qty_from_rejected_warehouse"):
|
||||
row.db_set(
|
||||
{
|
||||
"serial_and_batch_bundle": bundle,
|
||||
"batch_no": "",
|
||||
"serial_no": "",
|
||||
}
|
||||
)
|
||||
else:
|
||||
row.db_set(
|
||||
{
|
||||
"rejected_serial_and_batch_bundle": bundle,
|
||||
"batch_no": "",
|
||||
"rejected_serial_no": "",
|
||||
}
|
||||
)
|
||||
|
||||
def make_bundle_for_non_rejected_qty(self, table_name):
|
||||
field, reference_ids = self.get_reference_ids(table_name)
|
||||
if not reference_ids:
|
||||
return
|
||||
|
||||
child_doctype = self.doctype + " Item"
|
||||
if table_name == "packed_items":
|
||||
field = "parent_detail_docname"
|
||||
child_doctype = "Packed Item"
|
||||
|
||||
available_dict = available_serial_batch_for_return(field, child_doctype, reference_ids)
|
||||
|
||||
for row in self.get(table_name):
|
||||
value = row.get(field)
|
||||
if table_name == "packed_items" and row.get("parent_detail_docname"):
|
||||
value = self.get_value_for_packed_item(row)
|
||||
if not value:
|
||||
continue
|
||||
|
||||
if data := available_dict.get(value):
|
||||
data = filter_serial_batches(self, data, row)
|
||||
bundle = make_serial_batch_bundle_for_return(data, row, self)
|
||||
row.db_set(
|
||||
{
|
||||
"serial_and_batch_bundle": bundle,
|
||||
"batch_no": "",
|
||||
"serial_no": "",
|
||||
}
|
||||
)
|
||||
|
||||
if self.doctype in ["Sales Invoice", "Delivery Note"]:
|
||||
row.db_set(
|
||||
"incoming_rate", frappe.db.get_value("Serial and Batch Bundle", bundle, "avg_rate")
|
||||
)
|
||||
|
||||
def get_value_for_packed_item(self, row):
|
||||
parent_items = self.get("items", {"name": row.parent_detail_docname})
|
||||
if parent_items:
|
||||
ref = parent_items[0].get("dn_detail")
|
||||
return (row.item_code, ref)
|
||||
|
||||
return None
|
||||
|
||||
def get_reference_ids(self, table_name, qty_field=None, bundle_field=None) -> tuple[str, list[str]]:
|
||||
field = {
|
||||
"Sales Invoice": "sales_invoice_item",
|
||||
"Delivery Note": "dn_detail",
|
||||
"Purchase Receipt": "purchase_receipt_item",
|
||||
"Purchase Invoice": "purchase_invoice_item",
|
||||
"POS Invoice": "pos_invoice_item",
|
||||
}.get(self.doctype)
|
||||
|
||||
if not bundle_field:
|
||||
bundle_field = "serial_and_batch_bundle"
|
||||
|
||||
if not qty_field:
|
||||
qty_field = "qty"
|
||||
|
||||
reference_ids = []
|
||||
|
||||
for row in self.get(table_name):
|
||||
if not self.is_serial_batch_item(row.item_code):
|
||||
continue
|
||||
|
||||
if (
|
||||
row.get(field)
|
||||
and (
|
||||
qty_field == "qty"
|
||||
and not row.get("return_qty_from_rejected_warehouse")
|
||||
or qty_field == "rejected_qty"
|
||||
and (row.get("return_qty_from_rejected_warehouse") or row.get("rejected_warehouse"))
|
||||
)
|
||||
and not row.get("use_serial_batch_fields")
|
||||
and not row.get(bundle_field)
|
||||
):
|
||||
reference_ids.append(row.get(field))
|
||||
|
||||
if table_name == "packed_items" and row.get("parent_detail_docname"):
|
||||
parent_rows = self.get("items", {"name": row.parent_detail_docname}) or []
|
||||
for d in parent_rows:
|
||||
if d.get(field) and not d.get(bundle_field):
|
||||
reference_ids.append(d.get(field))
|
||||
|
||||
return field, reference_ids
|
||||
|
||||
@frappe.request_cache
|
||||
def is_serial_batch_item(self, item_code) -> bool:
|
||||
if not frappe.db.exists("Item", item_code):
|
||||
frappe.throw(_("Item {0} does not exist.").format(bold(item_code)))
|
||||
|
||||
item_details = frappe.db.get_value("Item", item_code, ["has_serial_no", "has_batch_no"], as_dict=1)
|
||||
|
||||
if item_details.has_serial_no or item_details.has_batch_no:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def update_bundle_details(self, bundle_details, table_name, row, is_rejected=False, parent_details=None):
|
||||
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
|
||||
|
||||
# Since qty field is different for different doctypes
|
||||
qty = row.get("qty")
|
||||
warehouse = row.get("warehouse")
|
||||
|
||||
if table_name == "packed_items":
|
||||
type_of_transaction = "Inward"
|
||||
if not self.is_return:
|
||||
type_of_transaction = "Outward"
|
||||
elif table_name == "supplied_items":
|
||||
qty = row.consumed_qty
|
||||
warehouse = self.supplier_warehouse
|
||||
type_of_transaction = "Outward"
|
||||
if self.is_return:
|
||||
type_of_transaction = "Inward"
|
||||
else:
|
||||
type_of_transaction = get_type_of_transaction(self, row)
|
||||
|
||||
if hasattr(row, "stock_qty"):
|
||||
qty = row.stock_qty
|
||||
|
||||
if self.doctype == "Stock Entry":
|
||||
qty = row.transfer_qty
|
||||
warehouse = row.s_warehouse or row.t_warehouse
|
||||
|
||||
serial_nos = row.serial_no
|
||||
if is_rejected:
|
||||
serial_nos = row.get("rejected_serial_no")
|
||||
type_of_transaction = "Inward" if not self.is_return else "Outward"
|
||||
qty = flt(
|
||||
row.get("rejected_qty") * row.get("conversion_factor", 1.0),
|
||||
frappe.get_precision("Serial and Batch Entry", "qty"),
|
||||
)
|
||||
warehouse = row.get("rejected_warehouse")
|
||||
|
||||
if (
|
||||
self.is_internal_transfer()
|
||||
and self.doctype in ["Sales Invoice", "Delivery Note"]
|
||||
and self.is_return
|
||||
):
|
||||
warehouse = row.get("target_warehouse") or row.get("warehouse")
|
||||
type_of_transaction = "Outward"
|
||||
|
||||
if table_name == "packed_items":
|
||||
if not warehouse:
|
||||
warehouse = parent_details[row.parent_detail_docname].warehouse
|
||||
bundle_details["voucher_detail_no"] = parent_details[row.parent_detail_docname].name
|
||||
|
||||
bundle_details.update(
|
||||
{
|
||||
"qty": qty,
|
||||
"is_rejected": is_rejected,
|
||||
"type_of_transaction": type_of_transaction,
|
||||
"warehouse": warehouse,
|
||||
"batches": frappe._dict({row.batch_no: qty}) if row.batch_no else None,
|
||||
"serial_nos": get_serial_nos(serial_nos) if serial_nos else None,
|
||||
"batch_no": row.batch_no,
|
||||
}
|
||||
)
|
||||
|
||||
def create_serial_batch_bundle(self, bundle_details, row):
|
||||
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
|
||||
|
||||
sn_doc = SerialBatchCreation(bundle_details).make_serial_and_batch_bundle()
|
||||
|
||||
field = "serial_and_batch_bundle"
|
||||
if bundle_details.get("is_rejected"):
|
||||
field = "rejected_serial_and_batch_bundle"
|
||||
|
||||
row.set(field, sn_doc.name)
|
||||
row.db_set({field: sn_doc.name})
|
||||
|
||||
def validate_serial_nos_and_batches_with_bundle(self, row):
|
||||
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
|
||||
|
||||
throw_error = False
|
||||
if row.serial_no:
|
||||
serial_nos = frappe.get_all(
|
||||
"Serial and Batch Entry",
|
||||
fields=["serial_no"],
|
||||
filters={"parent": row.serial_and_batch_bundle},
|
||||
)
|
||||
serial_nos = sorted([cstr(d.serial_no) for d in serial_nos])
|
||||
parsed_serial_nos = get_serial_nos(row.serial_no)
|
||||
|
||||
if len(serial_nos) != len(parsed_serial_nos):
|
||||
throw_error = True
|
||||
elif serial_nos != parsed_serial_nos:
|
||||
for serial_no in serial_nos:
|
||||
if serial_no not in parsed_serial_nos:
|
||||
throw_error = True
|
||||
break
|
||||
|
||||
elif row.batch_no:
|
||||
batches = sorted(
|
||||
frappe.get_all(
|
||||
"Serial and Batch Entry",
|
||||
filters={"parent": row.serial_and_batch_bundle},
|
||||
pluck="batch_no",
|
||||
distinct=True,
|
||||
)
|
||||
)
|
||||
|
||||
if batches != [row.batch_no]:
|
||||
throw_error = True
|
||||
|
||||
if throw_error:
|
||||
frappe.throw(
|
||||
_(
|
||||
"At row {0}: Serial and Batch Bundle {1} has already created. Please remove the values from the serial no or batch no fields."
|
||||
).format(row.idx, row.serial_and_batch_bundle)
|
||||
)
|
||||
return SerialBatchBundleService(self).make_bundle_for_sales_purchase_return(table_name)
|
||||
|
||||
def set_use_serial_batch_fields(self):
|
||||
if frappe.get_single_value("Stock Settings", "use_serial_batch_fields"):
|
||||
for row in self.items:
|
||||
row.use_serial_batch_fields = 1
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
return SerialBatchBundleService(self).set_use_serial_batch_fields()
|
||||
|
||||
def get_gl_entries(
|
||||
self, inventory_account_map=None, default_expense_account=None, default_cost_center=None
|
||||
@@ -826,74 +390,22 @@ class StockController(AccountsController):
|
||||
)
|
||||
|
||||
def delete_auto_created_batches(self):
|
||||
for table_name in ["items", "packed_items", "supplied_items"]:
|
||||
if not self.get(table_name):
|
||||
continue
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
for row in self.get(table_name):
|
||||
update_values = {}
|
||||
if row.get("batch_no"):
|
||||
update_values["batch_no"] = None
|
||||
|
||||
if row.get("serial_and_batch_bundle"):
|
||||
update_values["serial_and_batch_bundle"] = None
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Bundle", row.serial_and_batch_bundle, {"is_cancelled": 1}
|
||||
)
|
||||
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Entry", {"parent": row.serial_and_batch_bundle}, {"is_cancelled": 1}
|
||||
)
|
||||
|
||||
if update_values:
|
||||
row.db_set(update_values)
|
||||
|
||||
if table_name == "items" and row.get("rejected_serial_and_batch_bundle"):
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Bundle", row.rejected_serial_and_batch_bundle, {"is_cancelled": 1}
|
||||
)
|
||||
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Entry",
|
||||
{"parent": row.rejected_serial_and_batch_bundle},
|
||||
{"is_cancelled": 1},
|
||||
)
|
||||
|
||||
row.db_set("rejected_serial_and_batch_bundle", None)
|
||||
|
||||
if row.get("current_serial_and_batch_bundle"):
|
||||
row.db_set("current_serial_and_batch_bundle", None)
|
||||
return SerialBatchBundleService(self).delete_auto_created_batches()
|
||||
|
||||
def set_serial_and_batch_bundle(self, table_name=None, ignore_validate=False):
|
||||
if not table_name:
|
||||
table_name = "items"
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
QTY_FIELD = {
|
||||
"serial_and_batch_bundle": "qty",
|
||||
"current_serial_and_batch_bundle": "current_qty",
|
||||
"rejected_serial_and_batch_bundle": "rejected_qty",
|
||||
}
|
||||
|
||||
for row in self.get(table_name):
|
||||
for field in QTY_FIELD.keys():
|
||||
if row.get(field):
|
||||
frappe.get_doc("Serial and Batch Bundle", row.get(field)).set_serial_and_batch_values(
|
||||
self, row, qty_field=QTY_FIELD[field]
|
||||
)
|
||||
return SerialBatchBundleService(self).set_serial_and_batch_bundle(table_name, ignore_validate)
|
||||
|
||||
def make_package_for_transfer(
|
||||
self, serial_and_batch_bundle, warehouse, type_of_transaction=None, do_not_submit=None, qty=0
|
||||
):
|
||||
return make_bundle_for_material_transfer(
|
||||
is_new=self.is_new(),
|
||||
docstatus=self.docstatus,
|
||||
voucher_type=self.doctype,
|
||||
voucher_no=self.name,
|
||||
serial_and_batch_bundle=serial_and_batch_bundle,
|
||||
warehouse=warehouse,
|
||||
type_of_transaction=type_of_transaction,
|
||||
do_not_submit=do_not_submit,
|
||||
qty=qty,
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
|
||||
return SerialBatchBundleService(self).make_package_for_transfer(
|
||||
serial_and_batch_bundle, warehouse, type_of_transaction, do_not_submit, qty
|
||||
)
|
||||
|
||||
def get_sl_entries(self, d, args):
|
||||
@@ -1107,6 +619,7 @@ class StockController(AccountsController):
|
||||
|
||||
def make_sl_entries(self, sl_entries, allow_negative_stock=False, via_landed_cost_voucher=False):
|
||||
from erpnext.stock.serial_batch_bundle import update_batch_qty
|
||||
from erpnext.stock.services.serial_batch_bundle import SerialBatchBundleService
|
||||
from erpnext.stock.stock_ledger import make_sl_entries
|
||||
|
||||
make_sl_entries(sl_entries, allow_negative_stock, via_landed_cost_voucher)
|
||||
@@ -1114,108 +627,7 @@ class StockController(AccountsController):
|
||||
self.doctype, self.name, self.docstatus, via_landed_cost_voucher=via_landed_cost_voucher
|
||||
)
|
||||
|
||||
self.validate_reserved_batches()
|
||||
|
||||
def validate_reserved_batches(self):
|
||||
if not frappe.db.get_single_value("Stock Settings", "enable_stock_reservation"):
|
||||
return
|
||||
|
||||
if self.doctype not in ["Delivery Note", "Sales Invoice", "Stock Entry"]:
|
||||
return
|
||||
|
||||
batches = frappe.get_all(
|
||||
"Serial and Batch Entry",
|
||||
filters={
|
||||
"voucher_type": self.doctype,
|
||||
"voucher_no": self.name,
|
||||
"docstatus": 1,
|
||||
"batch_no": ("is", "set"),
|
||||
"qty": ("<", 0),
|
||||
},
|
||||
pluck="batch_no",
|
||||
)
|
||||
|
||||
if not batches:
|
||||
return
|
||||
|
||||
field_mapper = {
|
||||
"Sales Invoice": [["Sales Order", "sales_order"]],
|
||||
"Delivery Note": [["Sales Order", "against_sales_order"]],
|
||||
"Stock Entry": [
|
||||
["Work Order", "work_order"],
|
||||
["Subcontracting Inward Order", "subcontracting_inward_order"],
|
||||
],
|
||||
}.get(self.doctype)
|
||||
|
||||
qty_field = {
|
||||
"Sales Invoice": "qty",
|
||||
"Delivery Note": "qty",
|
||||
"Stock Entry": "fg_completed_qty",
|
||||
}.get(self.doctype)
|
||||
|
||||
reserved_batches_data = self.get_reserved_batches(batches)
|
||||
items = self.items
|
||||
if self.doctype == "Stock Entry":
|
||||
items = [self]
|
||||
|
||||
for item in items:
|
||||
for field in field_mapper:
|
||||
if not item.get(field[1]):
|
||||
continue
|
||||
|
||||
value = item.get(field[1])
|
||||
for row in reserved_batches_data:
|
||||
if self.doctype in ["Sales Invoice", "Delivery Note"] and row.item_code != item.get(
|
||||
"item_code"
|
||||
):
|
||||
continue
|
||||
|
||||
if row.voucher_no == value:
|
||||
continue
|
||||
|
||||
batch_qty = get_batch_qty(
|
||||
row.batch_no,
|
||||
row.warehouse,
|
||||
posting_date=self.posting_date,
|
||||
posting_time=self.posting_time,
|
||||
consider_negative_batches=True,
|
||||
)
|
||||
|
||||
if item.get(qty_field) < batch_qty:
|
||||
continue
|
||||
|
||||
frappe.throw(
|
||||
_(
|
||||
"The batch {0} is already reserved in {1} {2}. So, cannot proceed with the {3} {4}, which is created against the {5} {6}."
|
||||
).format(
|
||||
frappe.bold(row.batch_no),
|
||||
frappe.bold(row.voucher_type),
|
||||
frappe.bold(row.voucher_no),
|
||||
frappe.bold(self.doctype),
|
||||
frappe.bold(self.name),
|
||||
frappe.bold(field[0]),
|
||||
frappe.bold(value),
|
||||
),
|
||||
title=_("Reserved Batch Conflict"),
|
||||
)
|
||||
|
||||
def get_reserved_batches(self, batches):
|
||||
doctype = frappe.qb.DocType("Stock Reservation Entry")
|
||||
child_doc = frappe.qb.DocType("Serial and Batch Entry")
|
||||
|
||||
return (
|
||||
frappe.qb.from_(doctype)
|
||||
.join(child_doc)
|
||||
.on(doctype.name == child_doc.parent)
|
||||
.select(
|
||||
child_doc.batch_no,
|
||||
doctype.voucher_type,
|
||||
doctype.voucher_no,
|
||||
doctype.item_code,
|
||||
doctype.warehouse,
|
||||
)
|
||||
.where((doctype.docstatus == 1) & (child_doc.batch_no.isin(batches)))
|
||||
).run(as_dict=True)
|
||||
SerialBatchBundleService(self).validate_reserved_batches()
|
||||
|
||||
def make_gl_entries_on_cancel(self, from_repost=False):
|
||||
if not from_repost:
|
||||
@@ -1227,18 +639,6 @@ class StockController(AccountsController):
|
||||
):
|
||||
self.make_gl_entries()
|
||||
|
||||
def get_serialized_items(self):
|
||||
serialized_items = []
|
||||
item_codes = list(set(d.item_code for d in self.get("items")))
|
||||
if item_codes:
|
||||
serialized_items = frappe.db.sql_list(
|
||||
"""select name from `tabItem`
|
||||
where has_serial_no=1 and name in ({})""".format(", ".join(["%s"] * len(item_codes))),
|
||||
tuple(item_codes),
|
||||
)
|
||||
|
||||
return serialized_items
|
||||
|
||||
def validate_warehouse(self):
|
||||
from erpnext.stock.utils import validate_disabled_warehouse, validate_warehouse_company
|
||||
|
||||
|
||||
695
erpnext/stock/services/serial_batch_bundle.py
Normal file
695
erpnext/stock/services/serial_batch_bundle.py
Normal file
@@ -0,0 +1,695 @@
|
||||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: GNU General Public License v3. See license.txt
|
||||
|
||||
"""Serial & Batch Bundle handling for stock transactions.
|
||||
|
||||
Extracted from ``StockController``. Owns creation, validation and teardown of
|
||||
Serial and Batch Bundles for a stock voucher. The controller keeps thin
|
||||
delegators for methods reached from other doctypes / ``run_method``; internal
|
||||
helpers live here only.
|
||||
"""
|
||||
|
||||
import frappe
|
||||
from frappe import _, bold
|
||||
from frappe.utils import cstr, flt, get_link_to_form, getdate
|
||||
|
||||
from erpnext.controllers.sales_and_purchase_return import (
|
||||
available_serial_batch_for_return,
|
||||
filter_serial_batches,
|
||||
make_serial_batch_bundle_for_return,
|
||||
)
|
||||
from erpnext.stock.doctype.batch.batch import get_batch_qty
|
||||
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
|
||||
combine_datetime,
|
||||
get_type_of_transaction,
|
||||
)
|
||||
|
||||
|
||||
class SerialBatchBundleService:
|
||||
def __init__(self, doc) -> None:
|
||||
self.doc = doc
|
||||
|
||||
def validate_warehouse_of_sabb(self):
|
||||
if self.doc.is_internal_transfer():
|
||||
return
|
||||
|
||||
doc_before_save = self.doc.get_doc_before_save()
|
||||
|
||||
for row in self.doc.items:
|
||||
if not row.get("serial_and_batch_bundle"):
|
||||
continue
|
||||
|
||||
sabb_details = frappe.db.get_value(
|
||||
"Serial and Batch Bundle",
|
||||
row.serial_and_batch_bundle,
|
||||
["type_of_transaction", "warehouse", "has_serial_no"],
|
||||
as_dict=True,
|
||||
)
|
||||
if not sabb_details:
|
||||
continue
|
||||
|
||||
if sabb_details.type_of_transaction != "Outward":
|
||||
continue
|
||||
|
||||
warehouse = row.get("warehouse") or row.get("s_warehouse")
|
||||
if sabb_details.warehouse != warehouse:
|
||||
frappe.throw(
|
||||
_(
|
||||
"Row #{0}: Warehouse {1} does not match with the warehouse {2} in Serial and Batch Bundle {3}."
|
||||
).format(row.idx, warehouse, sabb_details.warehouse, row.serial_and_batch_bundle)
|
||||
)
|
||||
|
||||
if self.doc.doctype == "Stock Reconciliation":
|
||||
continue
|
||||
|
||||
if sabb_details.has_serial_no and doc_before_save and doc_before_save.get("items"):
|
||||
prev_row = doc_before_save.get("items", {"idx": row.idx})
|
||||
if prev_row and prev_row[0].serial_and_batch_bundle != row.serial_and_batch_bundle:
|
||||
sabb_doc = frappe.get_doc("Serial and Batch Bundle", row.serial_and_batch_bundle)
|
||||
sabb_doc.validate_serial_no_status()
|
||||
|
||||
def validate_duplicate_serial_and_batch_bundle(self, table_name):
|
||||
if not self.doc.get(table_name):
|
||||
return
|
||||
|
||||
sbb_list = []
|
||||
for item in self.doc.get(table_name):
|
||||
if item.get("serial_and_batch_bundle"):
|
||||
sbb_list.append(item.get("serial_and_batch_bundle"))
|
||||
|
||||
if item.get("rejected_serial_and_batch_bundle"):
|
||||
sbb_list.append(item.get("rejected_serial_and_batch_bundle"))
|
||||
|
||||
if sbb_list:
|
||||
SLE = frappe.qb.DocType("Stock Ledger Entry")
|
||||
data = (
|
||||
frappe.qb.from_(SLE)
|
||||
.select(SLE.voucher_type, SLE.voucher_no, SLE.serial_and_batch_bundle)
|
||||
.where(
|
||||
(SLE.docstatus == 1)
|
||||
& (SLE.serial_and_batch_bundle.notnull())
|
||||
& (SLE.serial_and_batch_bundle.isin(sbb_list))
|
||||
)
|
||||
.limit(1)
|
||||
).run(as_dict=True)
|
||||
|
||||
if data:
|
||||
data = data[0]
|
||||
frappe.throw(
|
||||
_("Serial and Batch Bundle {0} is already used in {1} {2}.").format(
|
||||
frappe.bold(data.serial_and_batch_bundle), data.voucher_type, data.voucher_no
|
||||
)
|
||||
)
|
||||
|
||||
def validate_serialized_batch(self):
|
||||
from erpnext.controllers.stock_controller import BatchExpiredError
|
||||
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
|
||||
|
||||
is_material_issue = False
|
||||
if self.doc.doctype == "Stock Entry" and self.doc.purpose in ["Material Issue", "Material Transfer"]:
|
||||
is_material_issue = True
|
||||
|
||||
for d in self.doc.get("items"):
|
||||
if hasattr(d, "serial_no") and hasattr(d, "batch_no") and d.serial_no and d.batch_no:
|
||||
serial_nos = frappe.get_all(
|
||||
"Serial No",
|
||||
fields=["batch_no", "name", "warehouse"],
|
||||
filters={"name": ("in", get_serial_nos(d.serial_no))},
|
||||
)
|
||||
|
||||
for row in serial_nos:
|
||||
if row.warehouse and row.batch_no != d.batch_no:
|
||||
frappe.throw(
|
||||
_("Row #{0}: Serial No {1} does not belong to Batch {2}").format(
|
||||
d.idx, row.name, d.batch_no
|
||||
)
|
||||
)
|
||||
|
||||
if is_material_issue:
|
||||
continue
|
||||
|
||||
if (
|
||||
flt(d.qty) > 0.0
|
||||
and d.get("batch_no")
|
||||
and self.doc.get("posting_date")
|
||||
and self.doc.docstatus < 2
|
||||
):
|
||||
expiry_date = frappe.get_cached_value("Batch", d.get("batch_no"), "expiry_date")
|
||||
|
||||
if expiry_date and getdate(expiry_date) < getdate(self.doc.posting_date):
|
||||
frappe.throw(
|
||||
_("Row #{0}: The batch {1} has already expired.").format(
|
||||
d.idx, get_link_to_form("Batch", d.get("batch_no"))
|
||||
),
|
||||
BatchExpiredError,
|
||||
)
|
||||
|
||||
def clean_serial_nos(self):
|
||||
from erpnext.stock.doctype.serial_no.serial_no import clean_serial_no_string
|
||||
|
||||
for row in self.doc.get("items"):
|
||||
if hasattr(row, "serial_no") and row.serial_no:
|
||||
# remove extra whitespace and store one serial no on each line
|
||||
row.serial_no = clean_serial_no_string(row.serial_no)
|
||||
|
||||
for row in self.doc.get("packed_items") or []:
|
||||
if hasattr(row, "serial_no") and row.serial_no:
|
||||
# remove extra whitespace and store one serial no on each line
|
||||
row.serial_no = clean_serial_no_string(row.serial_no)
|
||||
|
||||
def make_bundle_using_old_serial_batch_fields(self, table_name=None, via_landed_cost_voucher=False):
|
||||
if self.doc.get("_action") == "update_after_submit":
|
||||
return
|
||||
|
||||
# To handle test cases
|
||||
if frappe.in_test and frappe.flags.use_serial_and_batch_fields:
|
||||
return
|
||||
|
||||
if not table_name:
|
||||
table_name = "items"
|
||||
|
||||
if self.doc.doctype == "Asset Capitalization":
|
||||
table_name = "stock_items"
|
||||
|
||||
parent_details = frappe._dict()
|
||||
if table_name == "packed_items":
|
||||
parent_details = self.get_parent_details_for_packed_items()
|
||||
|
||||
for row in self.doc.get(table_name):
|
||||
if (
|
||||
not via_landed_cost_voucher
|
||||
and row.serial_and_batch_bundle
|
||||
and (row.serial_no or row.batch_no)
|
||||
):
|
||||
self.validate_serial_nos_and_batches_with_bundle(row)
|
||||
|
||||
if not row.serial_no and not row.batch_no and not row.get("rejected_serial_no"):
|
||||
continue
|
||||
|
||||
if not row.use_serial_batch_fields and (
|
||||
row.serial_no or row.batch_no or row.get("rejected_serial_no")
|
||||
):
|
||||
row.use_serial_batch_fields = 1
|
||||
|
||||
if row.use_serial_batch_fields and (
|
||||
not row.serial_and_batch_bundle and not row.get("rejected_serial_and_batch_bundle")
|
||||
):
|
||||
bundle_details = {
|
||||
"item_code": row.get("rm_item_code") or row.item_code,
|
||||
"posting_datetime": combine_datetime(self.doc.posting_date, self.doc.posting_time),
|
||||
"voucher_type": self.doc.doctype,
|
||||
"voucher_no": self.doc.name,
|
||||
"voucher_detail_no": row.name,
|
||||
"company": self.doc.company,
|
||||
"is_rejected": 1 if row.get("rejected_warehouse") else 0,
|
||||
"use_serial_batch_fields": row.use_serial_batch_fields,
|
||||
"via_landed_cost_voucher": via_landed_cost_voucher,
|
||||
"do_not_submit": True if not via_landed_cost_voucher else False,
|
||||
}
|
||||
|
||||
if self.doc.is_internal_transfer() and row.get("from_warehouse") and not self.doc.is_return:
|
||||
self.update_bundle_details(bundle_details, table_name, row)
|
||||
bundle_details["type_of_transaction"] = "Outward"
|
||||
bundle_details["warehouse"] = row.get("from_warehouse")
|
||||
bundle_details["qty"] = row.get("stock_qty") or row.get("qty")
|
||||
self.create_serial_batch_bundle(bundle_details, row)
|
||||
continue
|
||||
|
||||
if row.get("qty") or row.get("consumed_qty") or row.get("stock_qty"):
|
||||
self.update_bundle_details(bundle_details, table_name, row, parent_details=parent_details)
|
||||
self.create_serial_batch_bundle(bundle_details, row)
|
||||
|
||||
if row.get("rejected_qty"):
|
||||
self.update_bundle_details(bundle_details, table_name, row, is_rejected=True)
|
||||
self.create_serial_batch_bundle(bundle_details, row)
|
||||
|
||||
def get_parent_details_for_packed_items(self):
|
||||
parent_details = frappe._dict()
|
||||
for row in self.doc.get("items"):
|
||||
parent_details[row.name] = row
|
||||
|
||||
return parent_details
|
||||
|
||||
def make_bundle_for_sales_purchase_return(self, table_name=None):
|
||||
if not self.doc.get("is_return"):
|
||||
return
|
||||
|
||||
if not table_name:
|
||||
table_name = "items"
|
||||
|
||||
self.make_bundle_for_non_rejected_qty(table_name)
|
||||
|
||||
if self.doc.doctype in ["Purchase Invoice", "Purchase Receipt"]:
|
||||
self.make_bundle_for_rejected_qty(table_name)
|
||||
|
||||
def make_bundle_for_rejected_qty(self, table_name=None):
|
||||
field, reference_ids = self.get_reference_ids(
|
||||
table_name, "rejected_qty", "rejected_serial_and_batch_bundle"
|
||||
)
|
||||
|
||||
if not reference_ids:
|
||||
return
|
||||
|
||||
child_doctype = self.doc.doctype + " Item"
|
||||
available_dict = available_serial_batch_for_return(
|
||||
field, child_doctype, reference_ids, is_rejected=True
|
||||
)
|
||||
|
||||
for row in self.doc.get(table_name):
|
||||
if data := available_dict.get(row.get(field)):
|
||||
qty_field = "rejected_qty"
|
||||
warehouse_field = "rejected_warehouse"
|
||||
if row.get("return_qty_from_rejected_warehouse"):
|
||||
qty_field = "qty"
|
||||
warehouse_field = "warehouse"
|
||||
|
||||
if not data.get("qty"):
|
||||
frappe.throw(
|
||||
_("For the {0}, no stock is available for the return in the warehouse {1}.").format(
|
||||
frappe.bold(row.item_code), row.get(warehouse_field)
|
||||
)
|
||||
)
|
||||
|
||||
data = filter_serial_batches(
|
||||
self.doc, data, row, warehouse_field=warehouse_field, qty_field=qty_field
|
||||
)
|
||||
bundle = make_serial_batch_bundle_for_return(data, row, self.doc, warehouse_field, qty_field)
|
||||
if row.get("return_qty_from_rejected_warehouse"):
|
||||
row.db_set(
|
||||
{
|
||||
"serial_and_batch_bundle": bundle,
|
||||
"batch_no": "",
|
||||
"serial_no": "",
|
||||
}
|
||||
)
|
||||
else:
|
||||
row.db_set(
|
||||
{
|
||||
"rejected_serial_and_batch_bundle": bundle,
|
||||
"batch_no": "",
|
||||
"rejected_serial_no": "",
|
||||
}
|
||||
)
|
||||
|
||||
def make_bundle_for_non_rejected_qty(self, table_name):
|
||||
field, reference_ids = self.get_reference_ids(table_name)
|
||||
if not reference_ids:
|
||||
return
|
||||
|
||||
child_doctype = self.doc.doctype + " Item"
|
||||
if table_name == "packed_items":
|
||||
field = "parent_detail_docname"
|
||||
child_doctype = "Packed Item"
|
||||
|
||||
available_dict = available_serial_batch_for_return(field, child_doctype, reference_ids)
|
||||
|
||||
for row in self.doc.get(table_name):
|
||||
value = row.get(field)
|
||||
if table_name == "packed_items" and row.get("parent_detail_docname"):
|
||||
value = self.get_value_for_packed_item(row)
|
||||
if not value:
|
||||
continue
|
||||
|
||||
if data := available_dict.get(value):
|
||||
data = filter_serial_batches(self.doc, data, row)
|
||||
bundle = make_serial_batch_bundle_for_return(data, row, self.doc)
|
||||
row.db_set(
|
||||
{
|
||||
"serial_and_batch_bundle": bundle,
|
||||
"batch_no": "",
|
||||
"serial_no": "",
|
||||
}
|
||||
)
|
||||
|
||||
if self.doc.doctype in ["Sales Invoice", "Delivery Note"]:
|
||||
row.db_set(
|
||||
"incoming_rate", frappe.db.get_value("Serial and Batch Bundle", bundle, "avg_rate")
|
||||
)
|
||||
|
||||
def get_value_for_packed_item(self, row):
|
||||
parent_items = self.doc.get("items", {"name": row.parent_detail_docname})
|
||||
if parent_items:
|
||||
ref = parent_items[0].get("dn_detail")
|
||||
return (row.item_code, ref)
|
||||
|
||||
return None
|
||||
|
||||
def get_reference_ids(self, table_name, qty_field=None, bundle_field=None) -> tuple[str, list[str]]:
|
||||
field = {
|
||||
"Sales Invoice": "sales_invoice_item",
|
||||
"Delivery Note": "dn_detail",
|
||||
"Purchase Receipt": "purchase_receipt_item",
|
||||
"Purchase Invoice": "purchase_invoice_item",
|
||||
"POS Invoice": "pos_invoice_item",
|
||||
}.get(self.doc.doctype)
|
||||
|
||||
if not bundle_field:
|
||||
bundle_field = "serial_and_batch_bundle"
|
||||
|
||||
if not qty_field:
|
||||
qty_field = "qty"
|
||||
|
||||
reference_ids = []
|
||||
|
||||
for row in self.doc.get(table_name):
|
||||
if not self.is_serial_batch_item(row.item_code):
|
||||
continue
|
||||
|
||||
if (
|
||||
row.get(field)
|
||||
and (
|
||||
qty_field == "qty"
|
||||
and not row.get("return_qty_from_rejected_warehouse")
|
||||
or qty_field == "rejected_qty"
|
||||
and (row.get("return_qty_from_rejected_warehouse") or row.get("rejected_warehouse"))
|
||||
)
|
||||
and not row.get("use_serial_batch_fields")
|
||||
and not row.get(bundle_field)
|
||||
):
|
||||
reference_ids.append(row.get(field))
|
||||
|
||||
if table_name == "packed_items" and row.get("parent_detail_docname"):
|
||||
parent_rows = self.doc.get("items", {"name": row.parent_detail_docname}) or []
|
||||
for d in parent_rows:
|
||||
if d.get(field) and not d.get(bundle_field):
|
||||
reference_ids.append(d.get(field))
|
||||
|
||||
return field, reference_ids
|
||||
|
||||
@frappe.request_cache
|
||||
def is_serial_batch_item(self, item_code) -> bool:
|
||||
if not frappe.db.exists("Item", item_code):
|
||||
frappe.throw(_("Item {0} does not exist.").format(bold(item_code)))
|
||||
|
||||
item_details = frappe.db.get_value("Item", item_code, ["has_serial_no", "has_batch_no"], as_dict=1)
|
||||
|
||||
if item_details.has_serial_no or item_details.has_batch_no:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def update_bundle_details(self, bundle_details, table_name, row, is_rejected=False, parent_details=None):
|
||||
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
|
||||
|
||||
# Since qty field is different for different doctypes
|
||||
qty = row.get("qty")
|
||||
warehouse = row.get("warehouse")
|
||||
|
||||
if table_name == "packed_items":
|
||||
type_of_transaction = "Inward"
|
||||
if not self.doc.is_return:
|
||||
type_of_transaction = "Outward"
|
||||
elif table_name == "supplied_items":
|
||||
qty = row.consumed_qty
|
||||
warehouse = self.doc.supplier_warehouse
|
||||
type_of_transaction = "Outward"
|
||||
if self.doc.is_return:
|
||||
type_of_transaction = "Inward"
|
||||
else:
|
||||
type_of_transaction = get_type_of_transaction(self.doc, row)
|
||||
|
||||
if hasattr(row, "stock_qty"):
|
||||
qty = row.stock_qty
|
||||
|
||||
if self.doc.doctype == "Stock Entry":
|
||||
qty = row.transfer_qty
|
||||
warehouse = row.s_warehouse or row.t_warehouse
|
||||
|
||||
serial_nos = row.serial_no
|
||||
if is_rejected:
|
||||
serial_nos = row.get("rejected_serial_no")
|
||||
type_of_transaction = "Inward" if not self.doc.is_return else "Outward"
|
||||
qty = flt(
|
||||
row.get("rejected_qty") * row.get("conversion_factor", 1.0),
|
||||
frappe.get_precision("Serial and Batch Entry", "qty"),
|
||||
)
|
||||
warehouse = row.get("rejected_warehouse")
|
||||
|
||||
if (
|
||||
self.doc.is_internal_transfer()
|
||||
and self.doc.doctype in ["Sales Invoice", "Delivery Note"]
|
||||
and self.doc.is_return
|
||||
):
|
||||
warehouse = row.get("target_warehouse") or row.get("warehouse")
|
||||
type_of_transaction = "Outward"
|
||||
|
||||
if table_name == "packed_items":
|
||||
if not warehouse:
|
||||
warehouse = parent_details[row.parent_detail_docname].warehouse
|
||||
bundle_details["voucher_detail_no"] = parent_details[row.parent_detail_docname].name
|
||||
|
||||
bundle_details.update(
|
||||
{
|
||||
"qty": qty,
|
||||
"is_rejected": is_rejected,
|
||||
"type_of_transaction": type_of_transaction,
|
||||
"warehouse": warehouse,
|
||||
"batches": frappe._dict({row.batch_no: qty}) if row.batch_no else None,
|
||||
"serial_nos": get_serial_nos(serial_nos) if serial_nos else None,
|
||||
"batch_no": row.batch_no,
|
||||
}
|
||||
)
|
||||
|
||||
def create_serial_batch_bundle(self, bundle_details, row):
|
||||
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
|
||||
|
||||
sn_doc = SerialBatchCreation(bundle_details).make_serial_and_batch_bundle()
|
||||
|
||||
field = "serial_and_batch_bundle"
|
||||
if bundle_details.get("is_rejected"):
|
||||
field = "rejected_serial_and_batch_bundle"
|
||||
|
||||
row.set(field, sn_doc.name)
|
||||
row.db_set({field: sn_doc.name})
|
||||
|
||||
def validate_serial_nos_and_batches_with_bundle(self, row):
|
||||
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
|
||||
|
||||
throw_error = False
|
||||
if row.serial_no:
|
||||
serial_nos = frappe.get_all(
|
||||
"Serial and Batch Entry",
|
||||
fields=["serial_no"],
|
||||
filters={"parent": row.serial_and_batch_bundle},
|
||||
)
|
||||
serial_nos = sorted([cstr(d.serial_no) for d in serial_nos])
|
||||
parsed_serial_nos = get_serial_nos(row.serial_no)
|
||||
|
||||
if len(serial_nos) != len(parsed_serial_nos):
|
||||
throw_error = True
|
||||
elif serial_nos != parsed_serial_nos:
|
||||
for serial_no in serial_nos:
|
||||
if serial_no not in parsed_serial_nos:
|
||||
throw_error = True
|
||||
break
|
||||
|
||||
elif row.batch_no:
|
||||
batches = sorted(
|
||||
frappe.get_all(
|
||||
"Serial and Batch Entry",
|
||||
filters={"parent": row.serial_and_batch_bundle},
|
||||
pluck="batch_no",
|
||||
distinct=True,
|
||||
)
|
||||
)
|
||||
|
||||
if batches != [row.batch_no]:
|
||||
throw_error = True
|
||||
|
||||
if throw_error:
|
||||
frappe.throw(
|
||||
_(
|
||||
"At row {0}: Serial and Batch Bundle {1} has already created. Please remove the values from the serial no or batch no fields."
|
||||
).format(row.idx, row.serial_and_batch_bundle)
|
||||
)
|
||||
|
||||
def set_use_serial_batch_fields(self):
|
||||
if frappe.get_single_value("Stock Settings", "use_serial_batch_fields"):
|
||||
for row in self.doc.items:
|
||||
row.use_serial_batch_fields = 1
|
||||
|
||||
def delete_auto_created_batches(self):
|
||||
for table_name in ["items", "packed_items", "supplied_items"]:
|
||||
if not self.doc.get(table_name):
|
||||
continue
|
||||
|
||||
for row in self.doc.get(table_name):
|
||||
update_values = {}
|
||||
if row.get("batch_no"):
|
||||
update_values["batch_no"] = None
|
||||
|
||||
if row.get("serial_and_batch_bundle"):
|
||||
update_values["serial_and_batch_bundle"] = None
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Bundle", row.serial_and_batch_bundle, {"is_cancelled": 1}
|
||||
)
|
||||
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Entry", {"parent": row.serial_and_batch_bundle}, {"is_cancelled": 1}
|
||||
)
|
||||
|
||||
if update_values:
|
||||
row.db_set(update_values)
|
||||
|
||||
if table_name == "items" and row.get("rejected_serial_and_batch_bundle"):
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Bundle", row.rejected_serial_and_batch_bundle, {"is_cancelled": 1}
|
||||
)
|
||||
|
||||
frappe.db.set_value(
|
||||
"Serial and Batch Entry",
|
||||
{"parent": row.rejected_serial_and_batch_bundle},
|
||||
{"is_cancelled": 1},
|
||||
)
|
||||
|
||||
row.db_set("rejected_serial_and_batch_bundle", None)
|
||||
|
||||
if row.get("current_serial_and_batch_bundle"):
|
||||
row.db_set("current_serial_and_batch_bundle", None)
|
||||
|
||||
def set_serial_and_batch_bundle(self, table_name=None, ignore_validate=False):
|
||||
if not table_name:
|
||||
table_name = "items"
|
||||
|
||||
QTY_FIELD = {
|
||||
"serial_and_batch_bundle": "qty",
|
||||
"current_serial_and_batch_bundle": "current_qty",
|
||||
"rejected_serial_and_batch_bundle": "rejected_qty",
|
||||
}
|
||||
|
||||
for row in self.doc.get(table_name):
|
||||
for field in QTY_FIELD.keys():
|
||||
if row.get(field):
|
||||
frappe.get_doc("Serial and Batch Bundle", row.get(field)).set_serial_and_batch_values(
|
||||
self.doc, row, qty_field=QTY_FIELD[field]
|
||||
)
|
||||
|
||||
def make_package_for_transfer(
|
||||
self, serial_and_batch_bundle, warehouse, type_of_transaction=None, do_not_submit=None, qty=0
|
||||
):
|
||||
from erpnext.controllers.stock_controller import make_bundle_for_material_transfer
|
||||
|
||||
return make_bundle_for_material_transfer(
|
||||
is_new=self.doc.is_new(),
|
||||
docstatus=self.doc.docstatus,
|
||||
voucher_type=self.doc.doctype,
|
||||
voucher_no=self.doc.name,
|
||||
serial_and_batch_bundle=serial_and_batch_bundle,
|
||||
warehouse=warehouse,
|
||||
type_of_transaction=type_of_transaction,
|
||||
do_not_submit=do_not_submit,
|
||||
qty=qty,
|
||||
)
|
||||
|
||||
def validate_reserved_batches(self):
|
||||
if not frappe.db.get_single_value("Stock Settings", "enable_stock_reservation"):
|
||||
return
|
||||
|
||||
if self.doc.doctype not in ["Delivery Note", "Sales Invoice", "Stock Entry"]:
|
||||
return
|
||||
|
||||
batches = frappe.get_all(
|
||||
"Serial and Batch Entry",
|
||||
filters={
|
||||
"voucher_type": self.doc.doctype,
|
||||
"voucher_no": self.doc.name,
|
||||
"docstatus": 1,
|
||||
"batch_no": ("is", "set"),
|
||||
"qty": ("<", 0),
|
||||
},
|
||||
pluck="batch_no",
|
||||
)
|
||||
|
||||
if not batches:
|
||||
return
|
||||
|
||||
field_mapper = {
|
||||
"Sales Invoice": [["Sales Order", "sales_order"]],
|
||||
"Delivery Note": [["Sales Order", "against_sales_order"]],
|
||||
"Stock Entry": [
|
||||
["Work Order", "work_order"],
|
||||
["Subcontracting Inward Order", "subcontracting_inward_order"],
|
||||
],
|
||||
}.get(self.doc.doctype)
|
||||
|
||||
qty_field = {
|
||||
"Sales Invoice": "qty",
|
||||
"Delivery Note": "qty",
|
||||
"Stock Entry": "fg_completed_qty",
|
||||
}.get(self.doc.doctype)
|
||||
|
||||
reserved_batches_data = self.get_reserved_batches(batches)
|
||||
items = self.doc.items
|
||||
if self.doc.doctype == "Stock Entry":
|
||||
items = [self.doc]
|
||||
|
||||
for item in items:
|
||||
for field in field_mapper:
|
||||
if not item.get(field[1]):
|
||||
continue
|
||||
|
||||
value = item.get(field[1])
|
||||
for row in reserved_batches_data:
|
||||
if self.doc.doctype in ["Sales Invoice", "Delivery Note"] and row.item_code != item.get(
|
||||
"item_code"
|
||||
):
|
||||
continue
|
||||
|
||||
if row.voucher_no == value:
|
||||
continue
|
||||
|
||||
batch_qty = get_batch_qty(
|
||||
row.batch_no,
|
||||
row.warehouse,
|
||||
posting_date=self.doc.posting_date,
|
||||
posting_time=self.doc.posting_time,
|
||||
consider_negative_batches=True,
|
||||
)
|
||||
|
||||
if item.get(qty_field) < batch_qty:
|
||||
continue
|
||||
|
||||
frappe.throw(
|
||||
_(
|
||||
"The batch {0} is already reserved in {1} {2}. So, cannot proceed with the {3} {4}, which is created against the {5} {6}."
|
||||
).format(
|
||||
frappe.bold(row.batch_no),
|
||||
frappe.bold(row.voucher_type),
|
||||
frappe.bold(row.voucher_no),
|
||||
frappe.bold(self.doc.doctype),
|
||||
frappe.bold(self.doc.name),
|
||||
frappe.bold(field[0]),
|
||||
frappe.bold(value),
|
||||
),
|
||||
title=_("Reserved Batch Conflict"),
|
||||
)
|
||||
|
||||
def get_reserved_batches(self, batches):
|
||||
doctype = frappe.qb.DocType("Stock Reservation Entry")
|
||||
child_doc = frappe.qb.DocType("Serial and Batch Entry")
|
||||
|
||||
return (
|
||||
frappe.qb.from_(doctype)
|
||||
.join(child_doc)
|
||||
.on(doctype.name == child_doc.parent)
|
||||
.select(
|
||||
child_doc.batch_no,
|
||||
doctype.voucher_type,
|
||||
doctype.voucher_no,
|
||||
doctype.item_code,
|
||||
doctype.warehouse,
|
||||
)
|
||||
.where((doctype.docstatus == 1) & (child_doc.batch_no.isin(batches)))
|
||||
).run(as_dict=True)
|
||||
|
||||
def get_serialized_items(self):
|
||||
serialized_items = []
|
||||
item_codes = list(set(d.item_code for d in self.doc.get("items")))
|
||||
if item_codes:
|
||||
serialized_items = frappe.db.sql_list(
|
||||
"""select name from `tabItem`
|
||||
where has_serial_no=1 and name in ({})""".format(", ".join(["%s"] * len(item_codes))),
|
||||
tuple(item_codes),
|
||||
)
|
||||
|
||||
return serialized_items
|
||||
Reference in New Issue
Block a user