Shiloh/brotherton-erpnext
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: auto create serial and batch bundle

Browse Source
This commit is contained in:
Rohit Waghchaure 2023-03-28 12:16:27 +05:30
parent c1132d1e6d
commit 648efca940
18 changed files with 556 additions and 642 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
erpnext/accounts/doctype/pricing_rule/pricing_rule.py
View File

@ -237,10 +237,6 @@ def apply_pricing_rule(args, doc=None):
item_list = args.get("items")
args.pop("items")
set_serial_nos_based_on_fifo = frappe.db.get_single_value(
"Stock Settings", "automatically_set_serial_nos_based_on_fifo"
)
item_code_list = tuple(item.get("item_code") for item in item_list)
query_items = frappe.get_all(
"Item",
@ -258,28 +254,9 @@ def apply_pricing_rule(args, doc=None):
data = get_pricing_rule_for_item(args_copy, doc=doc)
out.append(data)
if (
serialized_items.get(item.get("item_code"))
and not item.get("serial_no")
and set_serial_nos_based_on_fifo
and not args.get("is_return")
):
out[0].update(get_serial_no_for_item(args_copy))
return out
def get_serial_no_for_item(args):
from erpnext.stock.get_item_details import get_serial_no
item_details = frappe._dict(
{"doctype": args.doctype, "name": args.name, "serial_no": args.serial_no}
)
if args.get("parenttype") in ("Sales Invoice", "Delivery Note") and flt(args.stock_qty) > 0:
item_details.serial_no = get_serial_no(args)
return item_details
def update_pricing_rule_uom(pricing_rule, args):
child_doc = {"Item Code": "items", "Item Group": "item_groups", "Brand": "brands"}.get(
pricing_rule.apply_on

4
erpnext/accounts/doctype/sales_invoice/sales_invoice.py
View File

@ -36,7 +36,6 @@ from erpnext.controllers.accounts_controller import validate_account_head
from erpnext.controllers.selling_controller import SellingController
from erpnext.projects.doctype.timesheet.timesheet import get_projectwise_timesheet_data
from erpnext.setup.doctype.company.company import update_company_current_month_sales
from erpnext.stock.doctype.batch.batch import set_batch_nos
from erpnext.stock.doctype.delivery_note.delivery_note import update_billed_amount_based_on_so
from erpnext.stock.doctype.serial_no.serial_no import get_delivery_note_serial_no, get_serial_nos
@ -125,9 +124,6 @@ class SalesInvoice(SellingController):
if not self.is_opening:
self.is_opening = "No"
if self._action != "submit" and self.update_stock and not self.is_return:
set_batch_nos(self, "warehouse", True)
if self.redeem_loyalty_points:
lp = frappe.get_doc("Loyalty Program", self.loyalty_program)
self.loyalty_redemption_account = (

20
erpnext/controllers/stock_controller.py
View File

@ -372,6 +372,16 @@ class StockController(AccountsController):
row.db_set("serial_and_batch_bundle", None)
def set_serial_and_batch_bundle(self, table_name=None):
if not table_name:
table_name = "items"
for row in self.get(table_name):
if row.get("serial_and_batch_bundle"):
frappe.get_doc(
"Serial and Batch Bundle", row.serial_and_batch_bundle
).set_serial_and_batch_values(self, row)
def make_package_for_transfer(
self, serial_and_batch_bundle, warehouse, type_of_transaction=None, do_not_submit=None
):
@ -749,16 +759,6 @@ class StockController(AccountsController):
message = self.prepare_over_receipt_message(rule, values)
frappe.throw(msg=message, title=_("Over Receipt"))
def set_serial_and_batch_bundle(self, table_name=None):
if not table_name:
table_name = "items"
for row in self.get(table_name):
if row.get("serial_and_batch_bundle"):
frappe.get_doc(
"Serial and Batch Bundle", row.serial_and_batch_bundle
).set_serial_and_batch_values(self, row)
def prepare_over_receipt_message(self, rule, values):
message = _(
"{0} qty of Item {1} is being received into Warehouse {2} with capacity {3}."

104
erpnext/selling/sales_common.js
View File

@ -196,48 +196,6 @@ erpnext.selling.SellingController = class SellingController extends erpnext.Tran
refresh_field("incentives",row.name,row.parentfield);
}
warehouse(doc, cdt, cdn) {
var me = this;
var item = frappe.get_doc(cdt, cdn);
// check if serial nos entered are as much as qty in row
if (item.serial_no) {
let serial_nos = item.serial_no.split(`\n`).filter(sn => sn.trim()); // filter out whitespaces
if (item.qty === serial_nos.length) return;
}
if (item.serial_no && !item.batch_no) {
item.serial_no = null;
}
var has_batch_no;
frappe.db.get_value('Item', {'item_code': item.item_code}, 'has_batch_no', (r) => {
has_batch_no = r && r.has_batch_no;
if(item.item_code && item.warehouse) {
return this.frm.call({
method: "erpnext.stock.get_item_details.get_bin_details_and_serial_nos",
child: item,
args: {
item_code: item.item_code,
warehouse: item.warehouse,
has_batch_no: has_batch_no || 0,
stock_qty: item.stock_qty,
serial_no: item.serial_no || "",
},
callback:function(r){
if (in_list(['Delivery Note', 'Sales Invoice'], doc.doctype)) {
if (doc.doctype === 'Sales Invoice' && (!doc.update_stock)) return;
if (has_batch_no) {
me.set_batch_number(cdt, cdn);
me.batch_no(doc, cdt, cdn);
}
}
}
});
}
})
}
toggle_editable_price_list_rate() {
var df = frappe.meta.get_docfield(this.frm.doc.doctype + " Item", "price_list_rate", this.frm.doc.name);
var editable_price_list_rate = cint(frappe.defaults.get_default("editable_price_list_rate"));
@ -298,36 +256,6 @@ erpnext.selling.SellingController = class SellingController extends erpnext.Tran
}
}
batch_no(doc, cdt, cdn) {
super.batch_no(doc, cdt, cdn);
var item = frappe.get_doc(cdt, cdn);
if (item.serial_no) {
return;
}
item.serial_no = null;
var has_serial_no;
frappe.db.get_value('Item', {'item_code': item.item_code}, 'has_serial_no', (r) => {
has_serial_no = r && r.has_serial_no;
if(item.warehouse && item.item_code && item.batch_no) {
return this.frm.call({
method: "erpnext.stock.get_item_details.get_batch_qty_and_serial_no",
child: item,
args: {
"batch_no": item.batch_no,
"stock_qty": item.stock_qty || item.qty, //if stock_qty field is not available fetch qty (in case of Packed Items table)
"warehouse": item.warehouse,
"item_code": item.item_code,
"has_serial_no": has_serial_no
},
"fieldname": "actual_batch_qty"
});
}
})
}
set_dynamic_labels() {
super.set_dynamic_labels();
this.set_product_bundle_help(this.frm.doc);
@ -388,38 +316,6 @@ erpnext.selling.SellingController = class SellingController extends erpnext.Tran
}
}
/* Determine appropriate batch number and set it in the form.
* @param {string} cdt - Document Doctype
* @param {string} cdn - Document name
*/
set_batch_number(cdt, cdn) {
const doc = frappe.get_doc(cdt, cdn);
if (doc && doc.has_batch_no && doc.warehouse) {
this._set_batch_number(doc);
}
}
_set_batch_number(doc) {
if (doc.batch_no) {
return
}
let args = {'item_code': doc.item_code, 'warehouse': doc.warehouse, 'qty': flt(doc.qty) * flt(doc.conversion_factor)};
if (doc.has_serial_no && doc.serial_no) {
args['serial_no'] = doc.serial_no
}
return frappe.call({
method: 'erpnext.stock.doctype.batch.batch.get_batch_no',
args: args,
callback: function(r) {
if(r.message) {
frappe.model.set_value(doc.doctype, doc.name, 'batch_no', r.message);
}
}
});
}
pick_serial_and_batch(doc, cdt, cdn) {
let item = locals[cdt][cdn];
let me = this;

1
erpnext/setup/setup_wizard/operations/defaults_setup.py
View File

@ -36,7 +36,6 @@ def set_default_settings(args):
stock_settings.stock_uom = _("Nos")
stock_settings.auto_indent = 1
stock_settings.auto_insert_price_list_rate_if_missing = 1
stock_settings.automatically_set_serial_nos_based_on_fifo = 1
stock_settings.set_qty_in_transactions_based_on_serial_no_input = 1
stock_settings.save()

1
erpnext/setup/setup_wizard/operations/install_fixtures.py
View File

@ -486,7 +486,6 @@ def update_stock_settings():
stock_settings.stock_uom = _("Nos")
stock_settings.auto_indent = 1
stock_settings.auto_insert_price_list_rate_if_missing = 1
stock_settings.automatically_set_serial_nos_based_on_fifo = 1
stock_settings.set_qty_in_transactions_based_on_serial_no_input = 1
stock_settings.save()

64
erpnext/stock/doctype/batch/batch.py
View File

@ -2,6 +2,8 @@
# License: GNU General Public License v3. See license.txt
from collections import defaultdict
import frappe
from frappe import _
from frappe.model.document import Document
@ -257,54 +259,6 @@ def split_batch(batch_no, item_code, warehouse, qty, new_batch_id=None):
return batch.name
def set_batch_nos(doc, warehouse_field, throw=False, child_table="items"):
"""Automatically select `batch_no` for outgoing items in item table"""
for d in doc.get(child_table):
qty = d.get("stock_qty") or d.get("transfer_qty") or d.get("qty") or 0
warehouse = d.get(warehouse_field, None)
if warehouse and qty > 0 and frappe.db.get_value("Item", d.item_code, "has_batch_no"):
if not d.batch_no:
pass
else:
batch_qty = get_batch_qty(batch_no=d.batch_no, warehouse=warehouse)
if flt(batch_qty, d.precision("qty")) < flt(qty, d.precision("qty")):
frappe.throw(
_(
"Row #{0}: The batch {1} has only {2} qty. Please select another batch which has {3} qty available or split the row into multiple rows, to deliver/issue from multiple batches"
).format(d.idx, d.batch_no, batch_qty, qty)
)
@frappe.whitelist()
def get_batch_no(item_code, warehouse, qty=1, throw=False, serial_no=None):
"""
Get batch number using First Expiring First Out method.
:param item_code: `item_code` of Item Document
:param warehouse: name of Warehouse to check
:param qty: quantity of Items
:return: String represent batch number of batch with sufficient quantity else an empty String
"""
batch_no = None
batches = get_batches(item_code, warehouse, qty, throw, serial_no)
for batch in batches:
if flt(qty) <= flt(batch.qty):
batch_no = batch.batch_id
break
if not batch_no:
frappe.msgprint(
_(
"Please select a Batch for Item {0}. Unable to find a single batch that fulfills this requirement"
).format(frappe.bold(item_code))
)
if throw:
raise UnableToSelectBatchError
return batch_no
def get_batches(item_code, warehouse, qty=1, throw=False, serial_no=None):
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
@ -398,3 +352,17 @@ def get_pos_reserved_batch_qty(filters):
flt_reserved_batch_qty = flt(reserved_batch_qty[0][0])
return flt_reserved_batch_qty
def get_available_batches(kwargs):
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
get_auto_batch_nos,
)
batchwise_qty = defaultdict(float)
batches = get_auto_batch_nos(kwargs)
for batch in batches:
batchwise_qty[batch.get("batch_no")] += batch.get("qty")
return batchwise_qty

19
erpnext/stock/doctype/delivery_note/delivery_note.py
View File

@ -137,6 +137,7 @@ class DeliveryNote(SellingController):
self.validate_uom_is_integer("stock_uom", "stock_qty")
self.validate_uom_is_integer("uom", "qty")
self.validate_with_previous_doc()
self.set_serial_and_batch_bundle_from_pick_list()
from erpnext.stock.doctype.packed_item.packed_item import make_packing_list
@ -187,6 +188,24 @@ class DeliveryNote(SellingController):
]
)
def set_serial_and_batch_bundle_from_pick_list(self):
if not self.pick_list:
return
for item in self.items:
if item.pick_list_item:
filters = {
"item_code": item.item_code,
"voucher_type": "Pick List",
"voucher_no": self.pick_list,
"voucher_detail_no": item.pick_list_item,
}
bundle_id = frappe.db.get_value("Serial and Batch Bundle", filters, "name")
if bundle_id:
item.serial_and_batch_bundle = bundle_id
def validate_proj_cust(self):
"""check for does customer belong to same project as entered.."""
if self.project and self.customer:

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

@ -12,14 +12,18 @@ from frappe.model.document import Document
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, IfNull, Locate, Replace, Sum
from frappe.utils import cint, floor, flt, today
from frappe.query_builder.functions import Coalesce, Locate, Replace, Sum
from frappe.utils import cint, floor, flt
from frappe.utils.nestedset import get_descendants_of
from erpnext.selling.doctype.sales_order.sales_order import (
make_delivery_note as create_delivery_note_from_sales_order,
)
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
get_auto_batch_nos,
)
from erpnext.stock.get_item_details import get_conversion_factor
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
# TODO: Prioritize SO or WO group warehouse
@ -79,6 +83,7 @@ class PickList(Document):
)
def on_submit(self):
self.validate_serial_and_batch_bundle()
self.update_status()
self.update_bundle_picked_qty()
self.update_reference_qty()
@ -90,7 +95,29 @@ class PickList(Document):
self.update_reference_qty()
self.update_sales_order_picking_status()
def update_status(self, status=None):
def on_update(self):
self.linked_serial_and_batch_bundle()
def linked_serial_and_batch_bundle(self):
for row in self.locations:
if row.serial_and_batch_bundle:
frappe.get_doc(
"Serial and Batch Bundle", row.serial_and_batch_bundle
).set_serial_and_batch_values(self, row)
def remove_serial_and_batch_bundle(self):
for row in self.locations:
if row.serial_and_batch_bundle:
frappe.delete_doc("Serial and Batch Bundle", row.serial_and_batch_bundle)
def validate_serial_and_batch_bundle(self):
for row in self.locations:
if row.serial_and_batch_bundle:
doc = frappe.get_doc("Serial and Batch Bundle", row.serial_and_batch_bundle)
if doc.docstatus == 0:
doc.submit()
def update_status(self, status=None, update_modified=True):
if not status:
if self.docstatus == 0:
status = "Draft"
@ -192,6 +219,7 @@ class PickList(Document):
locations_replica = self.get("locations")
# reset
self.remove_serial_and_batch_bundle()
self.delete_key("locations")
updated_locations = frappe._dict()
for item_doc in items:
@ -476,18 +504,13 @@ def get_items_with_location_and_quantity(item_doc, item_location_map, docstatus)
if not stock_qty:
break
serial_nos = None
if item_location.serial_no:
serial_nos = "\n".join(item_location.serial_no[0 : cint(stock_qty)])
locations.append(
frappe._dict(
{
"qty": qty,
"stock_qty": stock_qty,
"warehouse": item_location.warehouse,
"serial_no": serial_nos,
"batch_no": item_location.batch_no,
"serial_and_batch_bundle": item_location.serial_and_batch_bundle,
}
)
)
@ -553,23 +576,6 @@ def get_available_item_locations(
if picked_item_details:
for location in list(locations):
key = (
(location["warehouse"], location["batch_no"])
if location.get("batch_no")
else location["warehouse"]
)
if key in picked_item_details:
picked_detail = picked_item_details[key]
if picked_detail.get("serial_no") and location.get("serial_no"):
location["serial_no"] = list(
set(location["serial_no"]).difference(set(picked_detail["serial_no"]))
)
location["qty"] = len(location["serial_no"])
else:
location["qty"] -= picked_detail.get("picked_qty")
if location["qty"] < 1:
locations.remove(location)
@ -620,31 +626,50 @@ 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
):
sle = frappe.qb.DocType("Stock Ledger Entry")
batch = frappe.qb.DocType("Batch")
query = (
frappe.qb.from_(sle)
.from_(batch)
.select(sle.warehouse, sle.batch_no, Sum(sle.actual_qty).as_("qty"))
.where(
(sle.batch_no == batch.name)
& (sle.item_code == item_code)
& (sle.company == company)
& (batch.disabled == 0)
& (sle.is_cancelled == 0)
& (IfNull(batch.expiry_date, "2200-01-01") > today())
locations = []
data = get_auto_batch_nos(
frappe._dict(
{
"item_code": item_code,
"warehouse": from_warehouses,
"qty": required_qty + total_picked_qty,
}
)
.groupby(sle.warehouse, sle.batch_no, sle.item_code)
.having(Sum(sle.actual_qty) > 0)
.orderby(IfNull(batch.expiry_date, "2200-01-01"), batch.creation, sle.batch_no, sle.warehouse)
.limit(cint(required_qty + total_picked_qty))
)
if from_warehouses:
query = query.where(sle.warehouse.isin(from_warehouses))
warehouse_wise_batches = frappe._dict()
for d in data:
if d.warehouse not in warehouse_wise_batches:
warehouse_wise_batches.setdefault(d.warehouse, defaultdict(float))
return query.run(as_dict=True)
warehouse_wise_batches[d.warehouse][d.batch_no] += d.qty
for warehouse, batches in warehouse_wise_batches.items():
qty = sum(batches.values())
bundle_doc = SerialBatchCreation(
{
"item_code": item_code,
"warehouse": warehouse,
"voucher_type": "Pick List",
"total_qty": qty,
"batches": batches,
"type_of_transaction": "Outward",
"company": company,
"do_not_submit": True,
}
).make_serial_and_batch_bundle()
locations.append(
{
"qty": qty,
"warehouse": warehouse,
"item_code": item_code,
"serial_and_batch_bundle": bundle_doc.name,
}
)
return locations
def get_available_item_locations_for_serial_and_batched_item(

162
erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py
View File

@ -10,7 +10,6 @@ from frappe import _, bold
from frappe.model.document import Document
from frappe.query_builder.functions import CombineDatetime, Sum
from frappe.utils import add_days, cint, flt, get_link_to_form, today
from pypika import Case
from erpnext.stock.serial_batch_bundle import BatchNoValuation, SerialNoValuation
@ -24,8 +23,6 @@ class SerialandBatchBundle(Document):
self.validate_serial_and_batch_no()
self.validate_duplicate_serial_and_batch_no()
self.validate_voucher_no()
def before_save(self):
if self.type_of_transaction == "Maintenance":
return
@ -168,13 +165,16 @@ class SerialandBatchBundle(Document):
if not self.voucher_no or self.voucher_no != row.parent:
values_to_set["voucher_no"] = row.parent
if self.voucher_type != parent.doctype:
values_to_set["voucher_type"] = parent.doctype
if not self.voucher_detail_no or self.voucher_detail_no != row.name:
values_to_set["voucher_detail_no"] = row.name
if parent.get("posting_date") and (
not self.posting_date or self.posting_date != parent.posting_date
):
values_to_set["posting_date"] = parent.posting_date
values_to_set["posting_date"] = parent.posting_date or today()
if parent.get("posting_time") and (
not self.posting_time or self.posting_time != parent.posting_time
@ -222,6 +222,9 @@ class SerialandBatchBundle(Document):
if self.voucher_no and not frappe.db.exists(self.voucher_type, self.voucher_no):
frappe.throw(_(f"The {self.voucher_type} # {self.voucher_no} does not exist"))
if frappe.get_cached_value(self.voucher_type, self.voucher_no, "docstatus") != 1:
frappe.throw(_(f"The {self.voucher_type} # {self.voucher_no} should be submit first."))
def check_future_entries_exists(self):
if not self.has_serial_no:
return
@ -681,73 +684,43 @@ def get_auto_batch_nos(kwargs):
batches = []
reserved_batches = get_reserved_batches_for_pos(kwargs)
if reserved_batches:
remove_batches_reserved_for_pos(available_batches, reserved_batches)
stock_ledgers_batches = get_stock_ledgers_batches(kwargs)
if stock_ledgers_batches:
update_available_batches(available_batches, stock_ledgers_batches)
if not qty:
return batches
for batch in available_batches:
if qty > 0:
batch_qty = flt(batch.qty)
if qty > batch_qty:
batches.append(
{
"batch_no": batch.batch_no,
"qty": batch_qty,
}
frappe._dict(
{
"batch_no": batch.batch_no,
"qty": batch_qty,
"warehouse": batch.warehouse,
}
)
)
qty -= batch_qty
else:
batches.append(
{
"batch_no": batch.batch_no,
"qty": qty,
}
frappe._dict(
{
"batch_no": batch.batch_no,
"qty": qty,
"warehouse": batch.warehouse,
}
)
)
qty = 0
return batches
def get_reserved_batches_for_pos(kwargs):
reserved_batches = defaultdict(float)
pos_invoices = frappe.get_all(
"POS Invoice",
fields=[
"`tabPOS Invoice Item`.batch_no",
"`tabPOS Invoice Item`.qty",
"`tabPOS Invoice Item`.serial_and_batch_bundle",
],
filters=[
["POS Invoice", "consolidated_invoice", "is", "not set"],
["POS Invoice", "docstatus", "=", 1],
["POS Invoice Item", "item_code", "=", kwargs.item_code],
],
)
ids = [
pos_invoice.serial_and_batch_bundle
for pos_invoice in pos_invoices
if pos_invoice.serial_and_batch_bundle
]
for d in get_serial_batch_ledgers(ids, docstatus=1, name=ids):
if not d.batch_no:
continue
reserved_batches[d.batch_no] += flt(d.qty)
# Will be deprecated in v16
for pos_invoice in pos_invoices:
if not pos_invoice.batch_no:
continue
reserved_batches[pos_invoice.batch_no] += flt(pos_invoice.qty)
return reserved_batches
def remove_batches_reserved_for_pos(available_batches, reserved_batches):
def update_available_batches(available_batches, reserved_batches):
for batch in available_batches:
if batch.batch_no in reserved_batches:
available_batches[batch.batch_no] -= reserved_batches[batch.batch_no]
@ -766,16 +739,28 @@ def get_available_batches(kwargs):
.on(batch_ledger.batch_no == batch_table.name)
.select(
batch_ledger.batch_no,
batch_ledger.warehouse,
Sum(batch_ledger.qty).as_("qty"),
)
.where(
(stock_ledger_entry.item_code == kwargs.item_code)
& (stock_ledger_entry.warehouse == kwargs.warehouse)
& ((batch_table.expiry_date >= today()) | (batch_table.expiry_date.isnull()))
)
.where(((batch_table.expiry_date >= today()) | (batch_table.expiry_date.isnull())))
.groupby(batch_ledger.batch_no)
)
for field in ["warehouse", "item_code"]:
if not kwargs.get(field):
continue
if isinstance(kwargs.get(field), list):
query = query.where(stock_ledger_entry[field].isin(kwargs.get(field)))
else:
query = query.where(stock_ledger_entry[field] == kwargs.get(field))
if kwargs.get("batch_no"):
if isinstance(kwargs.batch_no, list):
query = query.where(batch_ledger.name.isin(kwargs.batch_no))
else:
query = query.where(batch_ledger.name == kwargs.batch_no)
if kwargs.based_on == "LIFO":
query = query.orderby(batch_table.creation, order=frappe.qb.desc)
elif kwargs.based_on == "Expiry":
@ -789,6 +774,7 @@ def get_available_batches(kwargs):
return data
# For work order and subcontracting
def get_voucher_wise_serial_batch_from_bundle(**kwargs) -> Dict[str, Dict]:
data = get_ledgers_from_serial_batch_bundle(**kwargs)
if not data:
@ -878,42 +864,34 @@ def get_available_serial_nos(item_code, warehouse):
return frappe.get_all("Serial No", filters=filters, fields=fields)
def get_available_batch_nos(item_code, warehouse):
sl_entries = get_stock_ledger_entries(item_code, warehouse)
batchwise_qty = defaultdict(float)
precision = frappe.get_precision("Stock Ledger Entry", "qty")
for entry in sl_entries:
batchwise_qty[entry.batch_no] += flt(entry.qty, precision)
return batchwise_qty
def get_stock_ledger_entries(item_code, warehouse):
def get_stock_ledgers_batches(kwargs):
stock_ledger_entry = frappe.qb.DocType("Stock Ledger Entry")
batch_ledger = frappe.qb.DocType("Serial and Batch Entry")
return (
query = (
frappe.qb.from_(stock_ledger_entry)
.left_join(batch_ledger)
.on(stock_ledger_entry.serial_and_batch_bundle == batch_ledger.parent)
.select(
stock_ledger_entry.warehouse,
stock_ledger_entry.item_code,
Sum(
Case()
.when(stock_ledger_entry.serial_and_batch_bundle, batch_ledger.qty)
.else_(stock_ledger_entry.actual_qty)
.as_("qty")
),
Case()
.when(stock_ledger_entry.serial_and_batch_bundle, batch_ledger.batch_no)
.else_(stock_ledger_entry.batch_no)
.as_("batch_no"),
Sum(stock_ledger_entry.actual_qty).as_("qty"),
stock_ledger_entry.batch_no,
)
.where(
(stock_ledger_entry.item_code == item_code)
& (stock_ledger_entry.warehouse == warehouse)
& (stock_ledger_entry.is_cancelled == 0)
)
).run(as_dict=True)
.where((stock_ledger_entry.is_cancelled == 0))
.groupby(stock_ledger_entry.batch_no, stock_ledger_entry.warehouse)
)
for field in ["warehouse", "item_code"]:
if not kwargs.get(field):
continue
if isinstance(kwargs.get(field), list):
query = query.where(stock_ledger_entry[field].isin(kwargs.get(field)))
else:
query = query.where(stock_ledger_entry[field] == kwargs.get(field))
data = query.run(as_dict=True)
batches = defaultdict(float)
for d in data:
batches[d.batch_no] += d.qty
return batches

13
erpnext/stock/doctype/serial_no/serial_no.py
View File

@ -322,3 +322,16 @@ def fetch_serial_numbers(filters, qty, do_not_include=None):
serial_numbers = query.run(as_dict=True)
return serial_numbers
def get_serial_nos_for_outward(kwargs):
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
get_auto_serial_nos,
)
serial_nos = get_auto_serial_nos(kwargs)
if not serial_nos:
return []
return [d.serial_no for d in serial_nos]

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

@ -28,7 +28,7 @@ from erpnext.controllers.taxes_and_totals import init_landed_taxes_and_totals
from erpnext.manufacturing.doctype.bom.bom import add_additional_cost, validate_bom_no
from erpnext.setup.doctype.brand.brand import get_brand_defaults
from erpnext.setup.doctype.item_group.item_group import get_item_group_defaults
from erpnext.stock.doctype.batch.batch import get_batch_no, get_batch_qty, set_batch_nos
from erpnext.stock.doctype.batch.batch import get_batch_qty
from erpnext.stock.doctype.item.item import get_item_defaults
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import (
@ -39,7 +39,11 @@ from erpnext.stock.get_item_details import (
get_conversion_factor,
get_default_cost_center,
)
from erpnext.stock.serial_batch_bundle import get_empty_batches_based_work_order
from erpnext.stock.serial_batch_bundle import (
SerialBatchCreation,
get_empty_batches_based_work_order,
get_serial_or_batch_items,
)
from erpnext.stock.stock_ledger import NegativeStockError, get_previous_sle, get_valuation_rate
from erpnext.stock.utils import get_bin, get_incoming_rate
@ -143,9 +147,6 @@ class StockEntry(StockController):
if not self.from_bom:
self.fg_completed_qty = 0.0
if self._action != "submit":
set_batch_nos(self, "s_warehouse")
self.validate_serialized_batch()
self.set_actual_qty()
self.calculate_rate_and_amount()
@ -242,6 +243,9 @@ class StockEntry(StockController):
if self.purpose == "Material Transfer" and self.outgoing_stock_entry:
self.set_material_request_transfer_status("In Transit")
def before_save(self):
self.make_serial_and_batch_bundle_for_outward()
def on_update(self):
self.set_serial_and_batch_bundle()
@ -894,6 +898,30 @@ class StockEntry(StockController):
serial_nos.append(sn)
def make_serial_and_batch_bundle_for_outward(self):
serial_or_batch_items = get_serial_or_batch_items(self.items)
for row in self.items:
if row.serial_and_batch_bundle or row.item_code not in serial_or_batch_items:
continue
bundle_doc = SerialBatchCreation(
{
"item_code": row.item_code,
"warehouse": row.s_warehouse,
"posting_date": self.posting_date,
"posting_time": self.posting_time,
"voucher_type": self.doctype,
"voucher_detail_no": row.name,
"total_qty": row.qty,
"type_of_transaction": "Outward",
"company": self.company,
"do_not_submit": True,
}
).make_serial_and_batch_bundle()
row.serial_and_batch_bundle = bundle_doc.name
def validate_subcontract_order(self):
"""Throw exception if more raw material is transferred against Subcontract Order than in
the raw materials supplied table"""
@ -1445,15 +1473,6 @@ class StockEntry(StockController):
stock_and_rate = get_warehouse_details(args) if args.get("warehouse") else {}
ret.update(stock_and_rate)
# automatically select batch for outgoing item
if (
args.get("s_warehouse", None)
and args.get("qty")
and ret.get("has_batch_no")
and not args.get("batch_no")
):
args.batch_no = get_batch_no(args["item_code"], args["s_warehouse"], args["qty"])
if (
self.purpose == "Send to Subcontractor"
and self.get(self.subcontract_data.order_field)

22
erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py
View File

@ -8,7 +8,7 @@ import frappe
from frappe import _
from frappe.core.doctype.role.role import get_users
from frappe.model.document import Document
from frappe.utils import add_days, cint, formatdate, get_datetime, get_link_to_form, getdate
from frappe.utils import add_days, cint, formatdate, get_datetime, getdate
from erpnext.accounts.utils import get_fiscal_year
from erpnext.controllers.item_variant import ItemTemplateCannotHaveStock
@ -51,7 +51,6 @@ class StockLedgerEntry(Document):
def on_submit(self):
self.check_stock_frozen_date()
self.calculate_batch_qty()
if not self.get("via_landed_cost_voucher"):
SerialBatchBundle(
@ -63,18 +62,6 @@ class StockLedgerEntry(Document):
self.validate_serial_batch_no_bundle()
def calculate_batch_qty(self):
if self.batch_no:
batch_qty = (
frappe.db.get_value(
"Stock Ledger Entry",
{"docstatus": 1, "batch_no": self.batch_no, "is_cancelled": 0},
"sum(actual_qty)",
)
or 0
)
frappe.db.set_value("Batch", self.batch_no, "batch_qty", batch_qty)
def validate_mandatory(self):
mandatory = ["warehouse", "posting_date", "voucher_type", "voucher_no", "company"]
for k in mandatory:
@ -123,12 +110,15 @@ class StockLedgerEntry(Document):
)
if bundle_data.docstatus != 1:
link = get_link_to_form("Serial and Batch Bundle", self.serial_and_batch_bundle)
frappe.throw(_(f"Serial and Batch Bundle {link} should be submitted first"))
self.submit_serial_and_batch_bundle()
if self.serial_and_batch_bundle and not (item_detail.has_serial_no or item_detail.has_batch_no):
frappe.throw(_(f"Serial No and Batch No are not allowed for Item {self.item_code}"))
def submit_serial_and_batch_bundle(self):
doc = frappe.get_doc("Serial and Batch Bundle", self.serial_and_batch_bundle)
doc.submit()
def check_stock_frozen_date(self):
stock_settings = frappe.get_cached_doc("Stock Settings")

11
erpnext/stock/doctype/stock_reconciliation/stock_reconciliation.py
View File

@ -13,7 +13,7 @@ from erpnext.accounts.utils import get_company_default
from erpnext.controllers.stock_controller import StockController
from erpnext.stock.doctype.batch.batch import get_batch_qty
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
get_available_batch_nos,
get_auto_batch_nos,
get_available_serial_nos,
)
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
@ -114,7 +114,14 @@ class StockReconciliation(StockController):
)
if item_details.has_batch_no:
batch_nos_details = get_available_batch_nos(item.item_code, item.warehouse)
batch_nos_details = get_auto_batch_nos(
frappe._dict(
{
"item_code": item.item_code,
"warehouse": item.warehouse,
}
)
)
for batch_no, qty in batch_nos_details.items():
serial_and_batch_bundle.append(

48
erpnext/stock/doctype/stock_settings/stock_settings.json
View File

@ -38,10 +38,11 @@
"allow_partial_reservation",
"serial_and_batch_item_settings_tab",
"section_break_7",
"automatically_set_serial_nos_based_on_fifo",
"set_qty_in_transactions_based_on_serial_no_input",
"column_break_10",
"auto_create_serial_and_batch_bundle_for_outward",
"pick_serial_and_batch_based_on",
"section_break_plhx",
"disable_serial_no_and_batch_selector",
"column_break_mhzc",
"use_naming_series",
"naming_series_prefix",
"stock_planning_tab",
@ -149,22 +150,6 @@
"fieldtype": "Check",
"label": "Allow Negative Stock"
},
{
"fieldname": "column_break_10",
"fieldtype": "Column Break"
},
{
"default": "1",
"fieldname": "automatically_set_serial_nos_based_on_fifo",
"fieldtype": "Check",
"label": "Automatically Set Serial Nos Based on FIFO"
},
{
"default": "1",
"fieldname": "set_qty_in_transactions_based_on_serial_no_input",
"fieldtype": "Check",
"label": "Set Qty in Transactions Based on Serial No Input"
},
{
"fieldname": "auto_material_request",
"fieldtype": "Section Break",
@ -376,6 +361,29 @@
"fieldname": "allow_partial_reservation",
"fieldtype": "Check",
"label": "Allow Partial Reservation"
},
{
"fieldname": "section_break_plhx",
"fieldtype": "Section Break"
},
{
"fieldname": "column_break_mhzc",
"fieldtype": "Column Break"
},
{
"default": "FIFO",
"depends_on": "auto_create_serial_and_batch_bundle_for_outward",
"fieldname": "pick_serial_and_batch_based_on",
"fieldtype": "Select",
"label": "Pick Serial / Batch Based On",
"mandatory_depends_on": "auto_create_serial_and_batch_bundle_for_outward",
"options": "FIFO\nLIFO\nExpiry"
},
{
"default": "1",
"fieldname": "auto_create_serial_and_batch_bundle_for_outward",
"fieldtype": "Check",
"label": "Auto Create Serial and Batch Bundle For Outward"
}
],
"icon": "icon-cog",
@ -383,7 +391,7 @@
"index_web_pages_for_search": 1,
"issingle": 1,
"links": [],
"modified": "2023-05-29 15:09:54.959411",
"modified": "2023-05-29 15:10:54.959411",
"modified_by": "Administrator",
"module": "Stock",
"name": "Stock Settings",

97
erpnext/stock/get_item_details.py
View File

@ -8,7 +8,7 @@ import frappe
from frappe import _, throw
from frappe.model import child_table_fields, default_fields
from frappe.model.meta import get_field_precision
from frappe.query_builder.functions import CombineDatetime, IfNull, Sum
from frappe.query_builder.functions import IfNull, Sum
from frappe.utils import add_days, add_months, cint, cstr, flt, getdate
from erpnext import get_company_currency
@ -1089,28 +1089,6 @@ def get_pos_profile(company, pos_profile=None, user=None):
return pos_profile and pos_profile[0] or None
def get_serial_nos_by_fifo(args, sales_order=None):
if frappe.db.get_single_value("Stock Settings", "automatically_set_serial_nos_based_on_fifo"):
sn = frappe.qb.DocType("Serial No")
query = (
frappe.qb.from_(sn)
.select(sn.name)
.where((sn.item_code == args.item_code) & (sn.warehouse == args.warehouse))
.orderby(CombineDatetime(sn.purchase_date, sn.purchase_time))
.limit(abs(cint(args.stock_qty)))
)
if sales_order:
query = query.where(sn.sales_order == sales_order)
if args.batch_no:
query = query.where(sn.batch_no == args.batch_no)
serial_nos = query.run(as_list=True)
serial_nos = [s[0] for s in serial_nos]
return "\n".join(serial_nos)
@frappe.whitelist()
def get_conversion_factor(item_code, uom):
variant_of = frappe.db.get_value("Item", item_code, "variant_of", cache=True)
@ -1176,51 +1154,6 @@ def get_company_total_stock(item_code, company):
).run()[0][0]
@frappe.whitelist()
def get_serial_no_details(item_code, warehouse, stock_qty, serial_no):
args = frappe._dict(
{"item_code": item_code, "warehouse": warehouse, "stock_qty": stock_qty, "serial_no": serial_no}
)
serial_no = get_serial_no(args)
return {"serial_no": serial_no}
@frappe.whitelist()
def get_bin_details_and_serial_nos(
item_code, warehouse, has_batch_no=None, stock_qty=None, serial_no=None
):
bin_details_and_serial_nos = {}
bin_details_and_serial_nos.update(get_bin_details(item_code, warehouse))
if flt(stock_qty) > 0:
if has_batch_no:
args = frappe._dict({"item_code": item_code, "warehouse": warehouse, "stock_qty": stock_qty})
serial_no = get_serial_no(args)
bin_details_and_serial_nos.update({"serial_no": serial_no})
return bin_details_and_serial_nos
bin_details_and_serial_nos.update(
get_serial_no_details(item_code, warehouse, stock_qty, serial_no)
)
return bin_details_and_serial_nos
@frappe.whitelist()
def get_batch_qty_and_serial_no(batch_no, stock_qty, warehouse, item_code, has_serial_no):
batch_qty_and_serial_no = {}
batch_qty_and_serial_no.update(get_batch_qty(batch_no, warehouse, item_code))
if (flt(batch_qty_and_serial_no.get("actual_batch_qty")) >= flt(stock_qty)) and has_serial_no:
args = frappe._dict(
{"item_code": item_code, "warehouse": warehouse, "stock_qty": stock_qty, "batch_no": batch_no}
)
serial_no = get_serial_no(args)
batch_qty_and_serial_no.update({"serial_no": serial_no})
return batch_qty_and_serial_no
@frappe.whitelist()
def get_batch_qty(batch_no, warehouse, item_code):
from erpnext.stock.doctype.batch import batch
@ -1395,32 +1328,8 @@ def get_gross_profit(out):
@frappe.whitelist()
def get_serial_no(args, serial_nos=None, sales_order=None):
serial_no = None
if isinstance(args, str):
args = json.loads(args)
args = frappe._dict(args)
if args.get("doctype") == "Sales Invoice" and not args.get("update_stock"):
return ""
if args.get("warehouse") and args.get("stock_qty") and args.get("item_code"):
has_serial_no = frappe.get_value("Item", {"item_code": args.item_code}, "has_serial_no")
if args.get("batch_no") and has_serial_no == 1:
return get_serial_nos_by_fifo(args, sales_order)
elif has_serial_no == 1:
args = json.dumps(
{
"item_code": args.get("item_code"),
"warehouse": args.get("warehouse"),
"stock_qty": args.get("stock_qty"),
}
)
args = process_args(args)
serial_no = get_serial_nos_by_fifo(args, sales_order)
if not serial_no and serial_nos:
# For POS
serial_no = serial_nos
return serial_no
serial_nos = serial_nos or []
return serial_nos
def update_party_blanket_order(args, out):

2
erpnext/stock/report/batch_wise_balance_history/batch_wise_balance_history.py
View File

@ -131,7 +131,7 @@ def get_stock_ledger_entries_for_batch_bundle(filters):
& (sle.has_batch_no == 1)
& (sle.posting_date <= filters["to_date"])
)
.groupby(batch_package.batch_no)
.groupby(batch_package.batch_no, batch_package.warehouse)
.orderby(sle.item_code, sle.warehouse)
)

441
erpnext/stock/serial_batch_bundle.py
View File

@ -49,103 +49,64 @@ class SerialBatchBundle:
if (
not self.sle.is_cancelled
and not self.sle.serial_and_batch_bundle
and self.sle.actual_qty > 0
and self.item_details.has_serial_no == 1
and self.item_details.serial_no_series
and self.allow_to_make_auto_bundle()
):
self.make_serial_batch_no_bundle()
elif not self.sle.is_cancelled:
self.validate_item_and_warehouse()
def auto_create_serial_nos(self, batch_no=None):
sr_nos = []
serial_nos_details = []
for i in range(cint(self.sle.actual_qty)):
serial_no = make_autoname(self.item_details.serial_no_series, "Serial No")
sr_nos.append(serial_no)
serial_nos_details.append(
(
serial_no,
serial_no,
now(),
now(),
frappe.session.user,
frappe.session.user,
self.warehouse,
self.company,
self.item_code,
self.item_details.item_name,
self.item_details.description,
"Active",
batch_no,
)
)
if serial_nos_details:
fields = [
"name",
"serial_no",
"creation",
"modified",
"owner",
"modified_by",
"warehouse",
"company",
"item_code",
"item_name",
"description",
"status",
"batch_no",
]
frappe.db.bulk_insert("Serial No", fields=fields, values=set(serial_nos_details))
return sr_nos
def make_serial_batch_no_bundle(self):
sn_doc = frappe.new_doc("Serial and Batch Bundle")
sn_doc.item_code = self.item_code
sn_doc.warehouse = self.warehouse
sn_doc.item_name = self.item_details.item_name
sn_doc.item_group = self.item_details.item_group
sn_doc.has_serial_no = self.item_details.has_serial_no
sn_doc.has_batch_no = self.item_details.has_batch_no
sn_doc.voucher_type = self.sle.voucher_type
sn_doc.voucher_no = self.sle.voucher_no
sn_doc.voucher_detail_no = self.sle.voucher_detail_no
sn_doc.total_qty = self.sle.actual_qty
sn_doc.avg_rate = self.sle.incoming_rate
sn_doc.total_amount = flt(self.sle.actual_qty) * flt(self.sle.incoming_rate)
sn_doc.type_of_transaction = "Inward"
sn_doc.posting_date = self.sle.posting_date
sn_doc.posting_time = self.sle.posting_time
sn_doc.is_rejected = self.is_rejected_entry()
self.validate_item()
sn_doc.flags.ignore_mandatory = True
sn_doc.insert()
sn_doc = SerialBatchCreation(
{
"item_code": self.item_code,
"warehouse": self.warehouse,
"posting_date": self.sle.posting_date,
"posting_time": self.sle.posting_time,
"voucher_type": self.sle.voucher_type,
"voucher_no": self.sle.voucher_no,
"voucher_detail_no": self.sle.voucher_detail_no,
"total_qty": self.sle.actual_qty,
"avg_rate": self.sle.incoming_rate,
"total_amount": flt(self.sle.actual_qty) * flt(self.sle.incoming_rate),
"type_of_transaction": "Inward" if self.sle.actual_qty > 0 else "Outward",
"company": self.company,
"is_rejected": self.is_rejected_entry(),
}
).make_serial_and_batch_bundle()
batch_no = ""
if self.item_details.has_batch_no:
batch_no = self.create_batch()
incoming_rate = self.sle.incoming_rate
if not incoming_rate:
incoming_rate = frappe.get_cached_value(
self.child_doctype, self.sle.voucher_detail_no, "valuation_rate"
)
if self.item_details.has_serial_no:
sr_nos = self.auto_create_serial_nos(batch_no)
self.add_serial_no_to_bundle(sn_doc, sr_nos, incoming_rate, batch_no)
elif self.item_details.has_batch_no:
self.add_batch_no_to_bundle(sn_doc, batch_no, incoming_rate)
sn_doc.save()
sn_doc.submit()
self.set_serial_and_batch_bundle(sn_doc)
def validate_item(self):
msg = ""
if self.sle.actual_qty > 0:
if not self.item_details.has_batch_no and not self.item_details.has_serial_no:
msg = f"Item {self.item_code} is not a batch or serial no item"
if self.item_details.has_serial_no and not self.item_details.serial_no_series:
msg += f". If you want auto pick serial bundle, then kindly set Serial No Series in Item {self.item_code}"
if (
self.item_details.has_batch_no
and not self.item_details.batch_number_series
and not frappe.db.get_single_value("Stock Settings", "naming_series_prefix")
):
msg += f". If you want auto pick batch bundle, then kindly set Batch Number Series in Item {self.item_code}"
elif self.sle.actual_qty < 0:
if not frappe.db.get_single_value(
"Stock Settings", "auto_create_serial_and_batch_bundle_for_outward"
):
msg += ". If you want auto pick serial/batch bundle, then kindly enable 'Auto Create Serial and Batch Bundle' in Stock Settings."
if msg:
error_msg = (
f"Serial and Batch Bundle not set for item {self.item_code} in warehouse {self.warehouse}."
+ msg
)
frappe.throw(_(error_msg))
def set_serial_and_batch_bundle(self, sn_doc):
self.sle.db_set("serial_and_batch_bundle", sn_doc.name)
@ -169,72 +130,19 @@ class SerialBatchBundle:
def is_rejected_entry(self):
return is_rejected(self.sle.voucher_type, self.sle.voucher_detail_no, self.sle.warehouse)
def add_serial_no_to_bundle(self, sn_doc, serial_nos, incoming_rate, batch_no=None):
for serial_no in serial_nos:
sn_doc.append(
"entries",
{
"serial_no": serial_no,
"qty": 1,
"incoming_rate": incoming_rate,
"batch_no": batch_no,
"warehouse": self.warehouse,
"is_outward": 0,
},
)
def add_batch_no_to_bundle(self, sn_doc, batch_no, incoming_rate):
stock_value_difference = flt(self.sle.actual_qty) * flt(incoming_rate)
if self.sle.actual_qty < 0:
stock_value_difference *= -1
sn_doc.append(
"entries",
{
"batch_no": batch_no,
"qty": self.sle.actual_qty,
"incoming_rate": incoming_rate,
"stock_value_difference": stock_value_difference,
},
)
def create_batch(self):
from erpnext.stock.doctype.batch.batch import make_batch
return make_batch(
frappe._dict(
{
"item": self.item_code,
"reference_doctype": self.sle.voucher_type,
"reference_name": self.sle.voucher_no,
}
)
)
def process_batch_no(self):
if (
not self.sle.is_cancelled
and not self.sle.serial_and_batch_bundle
and self.sle.actual_qty > 0
and self.item_details.has_batch_no == 1
and self.item_details.create_new_batch
and self.item_details.batch_number_series
and self.allow_to_make_auto_bundle()
):
self.make_serial_batch_no_bundle()
elif not self.sle.is_cancelled:
self.validate_item_and_warehouse()
def validate_item_and_warehouse(self):
data = frappe.db.get_value(
"Serial and Batch Bundle",
self.sle.serial_and_batch_bundle,
["item_code", "warehouse", "voucher_no", "name"],
as_dict=1,
)
if self.sle.serial_and_batch_bundle and not frappe.db.exists(
"Serial and Batch Bundle",
{
@ -270,18 +178,6 @@ class SerialBatchBundle:
{"is_cancelled": 1, "voucher_no": ""},
)
def allow_to_make_auto_bundle(self):
if self.sle.voucher_type in ["Stock Entry", "Purchase Receipt", "Purchase Invoice"]:
if self.sle.voucher_type == "Stock Entry":
stock_entry_type = frappe.get_cached_value("Stock Entry", self.sle.voucher_no, "purpose")
if stock_entry_type in ["Material Receipt", "Manufacture", "Repack"]:
return True
return True
return False
def post_process(self):
if not self.sle.serial_and_batch_bundle:
return
@ -296,6 +192,9 @@ class SerialBatchBundle:
):
self.set_batch_no_in_serial_nos()
if self.item_details.has_batch_no == 1:
self.update_batch_qty()
def set_warehouse_and_status_in_serial_nos(self):
serial_nos = get_serial_nos(self.sle.serial_and_batch_bundle, check_outward=False)
warehouse = self.warehouse if self.sle.actual_qty > 0 else None
@ -330,6 +229,20 @@ class SerialBatchBundle:
.where(sn_table.name.isin(serial_nos))
).run()
def update_batch_qty(self):
from erpnext.stock.doctype.batch.batch import get_available_batches
batches = get_batch_nos(self.sle.serial_and_batch_bundle)
batches_qty = get_available_batches(
frappe._dict(
{"item_code": self.item_code, "warehouse": self.warehouse, "batch_no": list(batches.keys())}
)
)
for batch_no, qty in batches_qty.items():
frappe.db.set_value("Batch", batch_no, "batch_qty", qty)
def get_serial_nos(serial_and_batch_bundle, check_outward=True):
filters = {"parent": serial_and_batch_bundle}
@ -489,6 +402,7 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
self.batch_avg_rate = defaultdict(float)
self.available_qty = defaultdict(float)
for ledger in entries:
self.batch_avg_rate[ledger.batch_no] += flt(ledger.incoming_rate) / flt(ledger.qty)
self.available_qty[ledger.batch_no] += flt(ledger.qty)
@ -502,11 +416,13 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
batch_nos = list(self.batch_nos.keys())
timestamp_condition = CombineDatetime(
parent.posting_date, parent.posting_time
) < CombineDatetime(self.sle.posting_date, self.sle.posting_time)
timestamp_condition = ""
if self.sle.posting_date and self.sle.posting_time:
timestamp_condition = CombineDatetime(
parent.posting_date, parent.posting_time
) < CombineDatetime(self.sle.posting_date, self.sle.posting_time)
return (
query = (
frappe.qb.from_(parent)
.inner_join(child)
.on(parent.name == child.parent)
@ -524,21 +440,19 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
& (parent.is_cancelled == 0)
& (parent.type_of_transaction != "Maintenance")
)
.where(timestamp_condition)
.groupby(child.batch_no)
).run(as_dict=True)
)
if timestamp_condition:
query.where(timestamp_condition)
return query.run(as_dict=True)
def get_batch_nos(self) -> list:
if self.sle.get("batch_nos"):
return self.sle.batch_nos
entries = frappe.get_all(
"Serial and Batch Entry",
fields=["batch_no", "qty", "name"],
filters={"parent": self.sle.serial_and_batch_bundle, "is_outward": 1},
)
return {d.batch_no: d for d in entries}
return get_batch_nos(self.sle.serial_and_batch_bundle)
def set_stock_value_difference(self):
self.stock_value_change = 0
@ -566,6 +480,16 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
return abs(flt(self.stock_value_change) / flt(self.sle.actual_qty))
def get_batch_nos(serial_and_batch_bundle):
entries = frappe.get_all(
"Serial and Batch Entry",
fields=["batch_no", "qty", "name"],
filters={"parent": serial_and_batch_bundle, "is_outward": 1},
)
return {d.batch_no: d for d in entries}
def get_empty_batches_based_work_order(work_order, item_code):
batches = get_batches_from_work_order(work_order, item_code)
if not batches:
@ -631,8 +555,35 @@ def set_batch_details_from_package(ids, batches):
class SerialBatchCreation:
def __init__(self, args):
self.set(args)
self.set_item_details()
def set(self, args):
self.__dict__ = {}
for key, value in args.items():
setattr(self, key, value)
self.__dict__[key] = value
def get(self, key):
return self.__dict__.get(key)
def set_item_details(self):
fields = [
"has_batch_no",
"has_serial_no",
"item_name",
"item_group",
"serial_no_series",
"create_new_batch",
"batch_number_series",
"description",
]
item_details = frappe.get_cached_value("Item", self.item_code, fields, as_dict=1)
for key, value in item_details.items():
setattr(self, key, value)
self.__dict__.update(item_details)
def duplicate_package(self):
if not self.serial_and_batch_bundle:
@ -643,7 +594,167 @@ class SerialBatchCreation:
new_package = frappe.copy_doc(package)
new_package.type_of_transaction = self.type_of_transaction
new_package.returned_against = self.returned_against
print(new_package.voucher_type, new_package.voucher_no)
new_package.save()
self.serial_and_batch_bundle = new_package.name
def make_serial_and_batch_bundle(self):
doc = frappe.new_doc("Serial and Batch Bundle")
valid_columns = doc.meta.get_valid_columns()
for key, value in self.__dict__.items():
if key in valid_columns:
doc.set(key, value)
if self.type_of_transaction == "Outward":
self.set_auto_serial_batch_entries_for_outward()
elif self.type_of_transaction == "Inward":
self.set_auto_serial_batch_entries_for_inward()
self.set_serial_batch_entries(doc)
doc.save()
if not hasattr(self, "do_not_submit") or not self.do_not_submit:
doc.submit()
return doc
def set_auto_serial_batch_entries_for_outward(self):
from erpnext.stock.doctype.batch.batch import get_available_batches
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos_for_outward
kwargs = frappe._dict(
{
"item_code": self.item_code,
"warehouse": self.warehouse,
"qty": abs(self.total_qty),
"based_on": frappe.db.get_single_value("Stock Settings", "pick_serial_and_batch_based_on"),
}
)
if self.has_serial_no and not self.get("serial_nos"):
self.serial_nos = get_serial_nos_for_outward(kwargs)
elif self.has_batch_no and not self.get("batches"):
self.batches = get_available_batches(kwargs)
def set_auto_serial_batch_entries_for_inward(self):
self.batch_no = None
if self.has_batch_no:
self.batch_no = self.create_batch()
if self.has_serial_no:
self.serial_nos = self.get_auto_created_serial_nos()
else:
self.batches = frappe._dict({self.batch_no: abs(self.total_qty)})
def set_serial_batch_entries(self, doc):
if self.get("serial_nos"):
serial_no_wise_batch = frappe._dict({})
if self.has_batch_no:
serial_no_wise_batch = self.get_serial_nos_batch(self.serial_nos)
qty = -1 if self.type_of_transaction == "Outward" else 1
for serial_no in self.serial_nos:
doc.append(
"entries",
{
"serial_no": serial_no,
"qty": qty,
"batch_no": serial_no_wise_batch.get(serial_no) or self.get("batch_no"),
"incoming_rate": self.get("incoming_rate"),
},
)
if self.get("batches"):
for batch_no, batch_qty in self.batches.items():
doc.append(
"entries",
{
"batch_no": batch_no,
"qty": batch_qty * (-1 if self.type_of_transaction == "Outward" else 1),
"incoming_rate": self.get("incoming_rate"),
},
)
def get_serial_nos_batch(self, serial_nos):
return frappe._dict(
frappe.get_all(
"Serial No",
fields=["name", "batch_no"],
filters={"name": ("in", serial_nos)},
as_list=1,
)
)
def create_batch(self):
from erpnext.stock.doctype.batch.batch import make_batch
return make_batch(
frappe._dict(
{
"item": self.item_code,
"reference_doctype": self.voucher_type,
"reference_name": self.voucher_no,
}
)
)
def get_auto_created_serial_nos(self):
sr_nos = []
serial_nos_details = []
for i in range(abs(cint(self.total_qty))):
serial_no = make_autoname(self.serial_no_series, "Serial No")
sr_nos.append(serial_no)
serial_nos_details.append(
(
serial_no,
serial_no,
now(),
now(),
frappe.session.user,
frappe.session.user,
self.warehouse,
self.company,
self.item_code,
self.item_name,
self.description,
"Active",
self.batch_no,
)
)
if serial_nos_details:
fields = [
"name",
"serial_no",
"creation",
"modified",
"owner",
"modified_by",
"warehouse",
"company",
"item_code",
"item_name",
"description",
"status",
"batch_no",
]
frappe.db.bulk_insert("Serial No", fields=fields, values=set(serial_nos_details))
return sr_nos
def get_serial_or_batch_items(items):
serial_or_batch_items = frappe.get_all(
"Item",
filters={"name": ("in", [d.item_code for d in items])},
or_filters={"has_serial_no": 1, "has_batch_no": 1},
)
if not serial_or_batch_items:
return
else:
serial_or_batch_items = [d.name for d in serial_or_batch_items]
return serial_or_batch_items