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

refactor: Batch & Bundle get Stock ledger for snos

Browse Source
This commit is contained in:
krupalvora
2026-01-09 03:24:30 +00:00
committed by Rohit Waghchaure
parent d8d74236dd
commit c0149925ad
1 changed files with 30 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,7 +12,7 @@ import frappe.query_builder.functions
from frappe import _, _dict, bold
from frappe.model.document import Document
from frappe.model.naming import make_autoname
from frappe.query_builder.functions import Sum
from frappe.query_builder.functions import Sum, Concat_ws, Locate
from frappe.utils import (
cint,
cstr,
@@ -2985,33 +2985,33 @@ def get_ledgers_from_serial_batch_bundle(**kwargs) -> list[frappe._dict]:
def get_stock_ledgers_for_serial_nos(kwargs):
stock_ledger_entry = frappe.qb.DocType("Stock Ledger Entry")
serial_batch_entry = frappe.qb.DocType("Serial and Batch Entry")
serial_nos = kwargs.get("serial_nos") or kwargs.get("serial_no")
if serial_nos and not isinstance(serial_nos, list):
serial_nos = [serial_nos]
query = (
frappe.qb.from_(stock_ledger_entry)
.select(
stock_ledger_entry.posting_datetime,
stock_ledger_entry.actual_qty,
stock_ledger_entry.serial_no,
stock_ledger_entry.serial_and_batch_bundle,
)
.where(stock_ledger_entry.is_cancelled == 0)
.orderby(stock_ledger_entry.posting_datetime)
.orderby(stock_ledger_entry.creation)
)
if kwargs.get("posting_datetime"):
timestamp_condition = stock_ledger_entry.posting_datetime <= kwargs.posting_datetime
if kwargs.get("posting_date"):
if kwargs.get("posting_time") is None:
kwargs.posting_time = nowtime()
if kwargs.get("creation"):
timestamp_condition = stock_ledger_entry.posting_datetime < kwargs.posting_datetime
timestamp_condition |= (stock_ledger_entry.posting_datetime == kwargs.posting_datetime) & (
stock_ledger_entry.creation < kwargs.creation
)
timestamp_condition = CombineDatetime(
stock_ledger_entry.posting_date, stock_ledger_entry.posting_time
) <= CombineDatetime(kwargs.posting_date, kwargs.posting_time)
query = query.where(timestamp_condition)
for field in ["warehouse", "item_code", "serial_no"]:
for field in ["warehouse", "item_code"]:
if not kwargs.get(field):
continue
@@ -3020,10 +3020,24 @@ def get_stock_ledgers_for_serial_nos(kwargs):
else:
query = query.where(stock_ledger_entry[field] == kwargs.get(field))
if kwargs.ignore_voucher_detail_no:
query = query.where(stock_ledger_entry.voucher_detail_no != kwargs.ignore_voucher_detail_no)
if serial_nos:
query = (
query.left_join(serial_batch_entry)
.on(stock_ledger_entry.serial_and_batch_bundle == serial_batch_entry.parent)
.distinct()
)
elif kwargs.voucher_no:
bundle_match = serial_batch_entry.serial_no.isin(serial_nos)
padded_serial_no = Concat_ws("", "\n", stock_ledger_entry.serial_no, "\n")
direct_match = None
for sn in serial_nos:
cond = Locate(f"\n{sn}\n", padded_serial_no) > 0
direct_match = cond if direct_match is None else (direct_match | cond)
query = query.where(bundle_match | direct_match)
if kwargs.voucher_no:
query = query.where(stock_ledger_entry.voucher_no != kwargs.voucher_no)
return query.run(as_dict=True)