schuetz3/schuetz3-erpnext
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: available serial no report

Browse Source 
(cherry picked from commit 5592d8e87f)
(cherry picked from commit c472af87b2)
This commit is contained in:
Mihir Kandoi
2025-03-07 11:54:15 +05:30
committed by Mergify
parent 52257c946e
commit 0f6a7edb53
6 changed files with 778 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -149,6 +149,12 @@ def get_serial_nos(serial_no):
return [s.strip() for s in cstr(serial_no).strip().replace(",", "\n").split("\n") if s.strip()]
def get_serial_nos_from_serial_and_batch_bundle(serial_and_batch_bundle):
table = frappe.qb.DocType("Serial and Batch Entry")
query = frappe.qb.from_(table).select(table.serial_no).where(table.parent == serial_and_batch_bundle)
return [item[0] for item in query.run(as_list=True)]
def clean_serial_no_string(serial_no: str) -> str:
if not serial_no:
return ""

0
erpnext/stock/report/available_serial_no/__init__.py Normal file
View File

118
erpnext/stock/report/available_serial_no/available_serial_no.js Normal file
View File

@@ -0,0 +1,118 @@
// Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
// License: GNU General Public License v3. See license.txt
frappe.query_reports["Available Serial No"] = {
filters: [
{
fieldname: "company",
label: __("Company"),
fieldtype: "Link",
options: "Company",
default: frappe.defaults.get_user_default("Company"),
reqd: 1,
},
{
fieldname: "from_date",
label: __("From Date"),
fieldtype: "Date",
default: frappe.datetime.add_months(frappe.datetime.get_today(), -1),
reqd: 1,
},
{
fieldname: "to_date",
label: __("To Date"),
fieldtype: "Date",
default: frappe.datetime.get_today(),
reqd: 1,
},
{
fieldname: "warehouse",
label: __("Warehouse"),
fieldtype: "Link",
options: "Warehouse",
get_query: function () {
const company = frappe.query_report.get_filter_value("company");
return {
filters: { company: company },
};
},
},
{
fieldname: "item_code",
label: __("Item"),
fieldtype: "Link",
options: "Item",
get_query: function () {
return {
query: "erpnext.controllers.queries.item_query",
filters: {
has_serial_no: 1,
},
};
},
},
{
fieldname: "item_group",
label: __("Item Group"),
fieldtype: "Link",
options: "Item Group",
},
{
fieldname: "batch_no",
label: __("Batch No"),
fieldtype: "Link",
options: "Batch",
on_change() {
const batch_no = frappe.query_report.get_filter_value("batch_no");
if (batch_no) {
frappe.query_report.set_filter_value("segregate_serial_batch_bundle", 1);
} else {
frappe.query_report.set_filter_value("segregate_serial_batch_bundle", 0);
}
},
},
{
fieldname: "brand",
label: __("Brand"),
fieldtype: "Link",
options: "Brand",
},
{
fieldname: "voucher_no",
label: __("Voucher #"),
fieldtype: "Data",
},
{
fieldname: "project",
label: __("Project"),
fieldtype: "Link",
options: "Project",
},
{
fieldname: "include_uom",
label: __("Include UOM"),
fieldtype: "Link",
options: "UOM",
},
{
fieldname: "valuation_field_type",
label: __("Valuation Field Type"),
fieldtype: "Select",
width: "80",
options: "Currency\nFloat",
default: "Currency",
},
],
formatter: function (value, row, column, data, default_formatter) {
value = default_formatter(value, row, column, data);
if (column.fieldname == "out_qty" && data && data.out_qty < 0) {
value = "<span style='color:red'>" + value + "</span>";
} else if (column.fieldname == "in_qty" && data && data.in_qty > 0) {
value = "<span style='color:green'>" + value + "</span>";
}
return value;
},
};
erpnext.utils.add_inventory_dimensions("Balance Serial No", 10);

28
erpnext/stock/report/available_serial_no/available_serial_no.json Normal file
View File

@@ -0,0 +1,28 @@
{
"add_total_row": 0,
"apply_user_permissions": 1,
"creation": "2025-03-07 10:54:09.429215",
"disabled": 0,
"docstatus": 0,
"doctype": "Report",
"idx": 0,
"is_standard": "Yes",
"modified": "2025-03-07 10:54:09.429215",
"modified_by": "Administrator",
"module": "Stock",
"name": "Available Serial No",
"owner": "Administrator",
"prepared_report": 0,
"ref_doctype": "Stock Ledger Entry",
"report_name": "Available Serial No",
"report_type": "Script Report",
"roles": [
{
"role": "Stock User"
},
{
"role": "Accounts Manager"
}
],
"timeout": 0
}

581
erpnext/stock/report/available_serial_no/available_serial_no.py Normal file
View File

@@ -0,0 +1,581 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
from collections import defaultdict
import frappe
from frappe import _
from frappe.query_builder.functions import CombineDatetime, Sum
from frappe.utils import cint, flt, get_datetime
from erpnext.stock.doctype.inventory_dimension.inventory_dimension import get_inventory_dimensions
from erpnext.stock.doctype.serial_no.serial_no import (
get_serial_nos,
get_serial_nos_from_serial_and_batch_bundle,
)
from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import get_stock_balance_for
from erpnext.stock.doctype.warehouse.warehouse import apply_warehouse_filter
from erpnext.stock.utils import (
is_reposting_item_valuation_in_progress,
update_included_uom_in_report,
)
def execute(filters=None):
is_reposting_item_valuation_in_progress()
include_uom = filters.get("include_uom")
columns = get_columns(filters)
items = get_items(filters)
sl_entries = get_stock_ledger_entries(filters, items)
item_details = get_item_details(items, sl_entries, include_uom)
if filters.get("batch_no"):
opening_row = get_opening_balance_from_batch(filters, columns, sl_entries)
else:
opening_row = get_opening_balance(filters, columns, sl_entries)
precision = cint(frappe.db.get_single_value("System Settings", "float_precision"))
data = []
conversion_factors = []
if opening_row:
data.append(opening_row)
conversion_factors.append(0)
actual_qty = stock_value = 0
if opening_row:
actual_qty = opening_row.get("qty_after_transaction")
stock_value = opening_row.get("stock_value")
available_serial_nos = {}
inventory_dimension_filters_applied = check_inventory_dimension_filters_applied(filters)
batch_balance_dict = frappe._dict({})
if actual_qty and filters.get("batch_no"):
batch_balance_dict[filters.batch_no] = [actual_qty, stock_value]
for sle in sl_entries:
item_detail = item_details[sle.item_code]
sle.update(item_detail)
if filters.get("batch_no") or inventory_dimension_filters_applied:
actual_qty += flt(sle.actual_qty, precision)
stock_value += sle.stock_value_difference
if sle.batch_no:
if not batch_balance_dict.get(sle.batch_no):
batch_balance_dict[sle.batch_no] = [0, 0]
batch_balance_dict[sle.batch_no][0] += sle.actual_qty
if sle.voucher_type == "Stock Reconciliation" and not sle.actual_qty:
actual_qty = sle.qty_after_transaction
stock_value = sle.stock_value
sle.update({"qty_after_transaction": actual_qty, "stock_value": stock_value})
sle.update({"in_qty": max(sle.actual_qty, 0), "out_qty": min(sle.actual_qty, 0)})
if frappe.get_value("Item", sle.item_code, "has_serial_no"):
update_available_serial_nos(available_serial_nos, sle)
if sle.actual_qty:
sle["in_out_rate"] = flt(sle.stock_value_difference / sle.actual_qty, precision)
elif sle.voucher_type == "Stock Reconciliation":
sle["in_out_rate"] = sle.valuation_rate
data.append(sle)
if include_uom:
conversion_factors.append(item_detail.conversion_factor)
update_included_uom_in_report(columns, data, include_uom, conversion_factors)
return columns, data
def update_available_serial_nos(available_serial_nos, sle):
serial_nos = (
get_serial_nos(sle.serial_no)
if sle.serial_no
else get_serial_nos_from_serial_and_batch_bundle(sle.serial_and_batch_bundle)
)
key = (sle.item_code, sle.warehouse)
if key not in available_serial_nos:
stock_balance = get_stock_balance_for(
sle.item_code, sle.warehouse, sle.posting_date, sle.posting_time
)
serials = get_serial_nos(stock_balance["serial_nos"]) if stock_balance["serial_nos"] else []
available_serial_nos.setdefault(key, serials)
sle.balance_serial_no = "\n".join(serials)
return
existing_serial_no = available_serial_nos[key]
for sn in serial_nos:
if sn in existing_serial_no:
existing_serial_no.remove(sn)
else:
existing_serial_no.append(sn)
sle.balance_serial_no = "\n".join(existing_serial_no)
def get_columns(filters):
columns = [
{"label": _("Date"), "fieldname": "date", "fieldtype": "Datetime", "width": 150},
{
"label": _("Item"),
"fieldname": "item_code",
"fieldtype": "Link",
"options": "Item",
"width": 100,
},
{"label": _("Item Name"), "fieldname": "item_name", "width": 100},
{
"label": _("Stock UOM"),
"fieldname": "stock_uom",
"fieldtype": "Link",
"options": "UOM",
"width": 90,
},
]
for dimension in get_inventory_dimensions():
columns.append(
{
"label": _(dimension.doctype),
"fieldname": dimension.fieldname,
"fieldtype": "Link",
"options": dimension.doctype,
"width": 110,
}
)
columns.extend(
[
{
"label": _("In Qty"),
"fieldname": "in_qty",
"fieldtype": "Float",
"width": 80,
"convertible": "qty",
},
{
"label": _("Out Qty"),
"fieldname": "out_qty",
"fieldtype": "Float",
"width": 80,
"convertible": "qty",
},
{
"label": _("Balance Qty"),
"fieldname": "qty_after_transaction",
"fieldtype": "Float",
"width": 100,
"convertible": "qty",
},
{
"label": _("Warehouse"),
"fieldname": "warehouse",
"fieldtype": "Link",
"options": "Warehouse",
"width": 150,
},
{
"label": _("Item Group"),
"fieldname": "item_group",
"fieldtype": "Link",
"options": "Item Group",
"width": 100,
},
{
"label": _("Brand"),
"fieldname": "brand",
"fieldtype": "Link",
"options": "Brand",
"width": 100,
},
{"label": _("Description"), "fieldname": "description", "width": 200},
{
"label": _("Incoming Rate"),
"fieldname": "incoming_rate",
"fieldtype": "Currency",
"width": 110,
"options": "Company:company:default_currency",
"convertible": "rate",
},
{
"label": _("Avg Rate (Balance Stock)"),
"fieldname": "valuation_rate",
"fieldtype": filters.valuation_field_type,
"width": 180,
"options": "Company:company:default_currency"
if filters.valuation_field_type == "Currency"
else None,
"convertible": "rate",
},
{
"label": _("Valuation Rate"),
"fieldname": "in_out_rate",
"fieldtype": filters.valuation_field_type,
"width": 140,
"options": "Company:company:default_currency"
if filters.valuation_field_type == "Currency"
else None,
"convertible": "rate",
},
{
"label": _("Balance Value"),
"fieldname": "stock_value",
"fieldtype": "Currency",
"width": 110,
"options": "Company:company:default_currency",
},
{
"label": _("Value Change"),
"fieldname": "stock_value_difference",
"fieldtype": "Currency",
"width": 110,
"options": "Company:company:default_currency",
},
{"label": _("Voucher Type"), "fieldname": "voucher_type", "width": 110},
{
"label": _("Voucher #"),
"fieldname": "voucher_no",
"fieldtype": "Dynamic Link",
"options": "voucher_type",
"width": 100,
},
{
"label": _("Batch"),
"fieldname": "batch_no",
"fieldtype": "Link",
"options": "Batch",
"width": 100,
},
{
"label": _("Serial No"),
"fieldname": "serial_no",
"fieldtype": "Link",
"options": "Serial No",
"width": 100,
},
{
"label": _("Serial and Batch Bundle"),
"fieldname": "serial_and_batch_bundle",
"fieldtype": "Link",
"options": "Serial and Batch Bundle",
"width": 100,
},
{"label": _("Balance Serial No"), "fieldname": "balance_serial_no", "width": 100},
{
"label": _("Project"),
"fieldname": "project",
"fieldtype": "Link",
"options": "Project",
"width": 100,
},
{
"label": _("Company"),
"fieldname": "company",
"fieldtype": "Link",
"options": "Company",
"width": 110,
},
]
)
return columns
def get_stock_ledger_entries(filters, items):
from_date = get_datetime(filters.from_date + " 00:00:00")
to_date = get_datetime(filters.to_date + " 23:59:59")
sle = frappe.qb.DocType("Stock Ledger Entry")
query = (
frappe.qb.from_(sle)
.select(
sle.item_code,
sle.posting_datetime.as_("date"),
sle.warehouse,
sle.posting_date,
sle.posting_time,
sle.actual_qty,
sle.incoming_rate,
sle.valuation_rate,
sle.company,
sle.voucher_type,
sle.qty_after_transaction,
sle.stock_value_difference,
sle.serial_and_batch_bundle,
sle.voucher_no,
sle.stock_value,
sle.batch_no,
sle.serial_no,
sle.project,
)
.where((sle.docstatus < 2) & (sle.is_cancelled == 0) & (sle.posting_datetime[from_date:to_date]))
.orderby(sle.posting_datetime)
.orderby(sle.creation)
)
inventory_dimension_fields = get_inventory_dimension_fields()
if inventory_dimension_fields:
for fieldname in inventory_dimension_fields:
query = query.select(fieldname)
if fieldname in filters and filters.get(fieldname):
query = query.where(sle[fieldname].isin(filters.get(fieldname)))
if items:
query = query.where(sle.item_code.isin(items))
for field in ["voucher_no", "project", "company"]:
if filters.get(field) and field not in inventory_dimension_fields:
query = query.where(sle[field] == filters.get(field))
if filters.get("batch_no"):
bundles = get_serial_and_batch_bundles(filters)
if bundles:
query = query.where(
(sle.serial_and_batch_bundle.isin(bundles)) | (sle.batch_no == filters.batch_no)
)
else:
query = query.where(sle.batch_no == filters.batch_no)
query = apply_warehouse_filter(query, sle, filters)
return query.run(as_dict=True)
def get_serial_and_batch_bundles(filters):
SBB = frappe.qb.DocType("Serial and Batch Bundle")
SBE = frappe.qb.DocType("Serial and Batch Entry")
query = (
frappe.qb.from_(SBE)
.inner_join(SBB)
.on(SBE.parent == SBB.name)
.select(SBE.parent)
.where(
(SBB.docstatus == 1)
& (SBB.has_batch_no == 1)
& (SBB.voucher_no.notnull())
& (SBE.batch_no == filters.batch_no)
)
)
return query.run(pluck=SBE.parent)
def get_inventory_dimension_fields():
return [dimension.fieldname for dimension in get_inventory_dimensions()]
def get_items(filters):
item = frappe.qb.DocType("Item")
query = frappe.qb.from_(item).select(item.name).where(item.has_serial_no == 1)
conditions = []
if item_code := filters.get("item_code"):
conditions.append(item.name == item_code)
else:
if brand := filters.get("brand"):
conditions.append(item.brand == brand)
if item_group := filters.get("item_group"):
if condition := get_item_group_condition(item_group, item):
conditions.append(condition)
items = []
if conditions:
for condition in conditions:
query = query.where(condition)
items = [r[0] for r in query.run()]
return items
def get_item_details(items, sl_entries, include_uom):
item_details = {}
if not items:
items = list(set(d.item_code for d in sl_entries))
if not items:
return item_details
item = frappe.qb.DocType("Item")
query = (
frappe.qb.from_(item)
.select(item.name, item.item_name, item.description, item.item_group, item.brand, item.stock_uom)
.where(item.name.isin(items))
)
if include_uom:
ucd = frappe.qb.DocType("UOM Conversion Detail")
query = (
query.left_join(ucd)
.on((ucd.parent == item.name) & (ucd.uom == include_uom))
.select(ucd.conversion_factor)
)
res = query.run(as_dict=True)
for item in res:
item_details.setdefault(item.name, item)
return item_details
def get_sle_conditions(filters):
conditions = []
if filters.get("warehouse"):
warehouse_condition = get_warehouse_condition(filters.get("warehouse"))
if warehouse_condition:
conditions.append(warehouse_condition)
if filters.get("voucher_no"):
conditions.append("voucher_no=%(voucher_no)s")
if filters.get("batch_no"):
conditions.append("batch_no=%(batch_no)s")
if filters.get("project"):
conditions.append("project=%(project)s")
for dimension in get_inventory_dimensions():
if filters.get(dimension.fieldname):
conditions.append(f"{dimension.fieldname} in %({dimension.fieldname})s")
return "and {}".format(" and ".join(conditions)) if conditions else ""
def get_opening_balance_from_batch(filters, columns, sl_entries):
query_filters = {
"batch_no": filters.batch_no,
"docstatus": 1,
"is_cancelled": 0,
"posting_date": ("<", filters.from_date),
"company": filters.company,
}
for fields in ["item_code", "warehouse"]:
if filters.get(fields):
query_filters[fields] = filters.get(fields)
opening_data = frappe.get_all(
"Stock Ledger Entry",
fields=["sum(actual_qty) as qty_after_transaction", "sum(stock_value_difference) as stock_value"],
filters=query_filters,
)[0]
for field in ["qty_after_transaction", "stock_value", "valuation_rate"]:
if opening_data.get(field) is None:
opening_data[field] = 0.0
table = frappe.qb.DocType("Stock Ledger Entry")
sabb_table = frappe.qb.DocType("Serial and Batch Entry")
query = (
frappe.qb.from_(table)
.inner_join(sabb_table)
.on(table.serial_and_batch_bundle == sabb_table.parent)
.select(
Sum(sabb_table.qty).as_("qty"),
Sum(sabb_table.stock_value_difference).as_("stock_value"),
)
.where(
(sabb_table.batch_no == filters.batch_no)
& (sabb_table.docstatus == 1)
& (table.posting_date < filters.from_date)
& (table.is_cancelled == 0)
)
)
for field in ["item_code", "warehouse", "company"]:
if filters.get(field):
query = query.where(table[field] == filters.get(field))
bundle_data = query.run(as_dict=True)
if bundle_data:
opening_data.qty_after_transaction += flt(bundle_data[0].qty)
opening_data.stock_value += flt(bundle_data[0].stock_value)
if opening_data.qty_after_transaction:
opening_data.valuation_rate = flt(opening_data.stock_value) / flt(
opening_data.qty_after_transaction
)
return {
"item_code": _("'Opening'"),
"qty_after_transaction": opening_data.qty_after_transaction,
"valuation_rate": opening_data.valuation_rate,
"stock_value": opening_data.stock_value,
}
def get_opening_balance(filters, columns, sl_entries):
if not (filters.item_code and filters.warehouse and filters.from_date):
return
from erpnext.stock.stock_ledger import get_previous_sle
last_entry = get_previous_sle(
{
"item_code": filters.item_code,
"warehouse_condition": get_warehouse_condition(filters.warehouse),
"posting_date": filters.from_date,
"posting_time": "00:00:00",
}
)
# check if any SLEs are actually Opening Stock Reconciliation
for sle in list(sl_entries):
if (
sle.get("voucher_type") == "Stock Reconciliation"
and sle.posting_date == filters.from_date
and frappe.db.get_value("Stock Reconciliation", sle.voucher_no, "purpose") == "Opening Stock"
):
last_entry = sle
sl_entries.remove(sle)
row = {
"item_code": _("'Opening'"),
"qty_after_transaction": last_entry.get("qty_after_transaction", 0),
"valuation_rate": last_entry.get("valuation_rate", 0),
"stock_value": last_entry.get("stock_value", 0),
}
return row
def get_warehouse_condition(warehouse):
warehouse_details = frappe.db.get_value("Warehouse", warehouse, ["lft", "rgt"], as_dict=1)
if warehouse_details:
return f" exists (select name from `tabWarehouse` wh \
where wh.lft >= {warehouse_details.lft} and wh.rgt <= {warehouse_details.rgt} and warehouse = wh.name)"
return ""
def get_item_group_condition(item_group, item_table=None):
item_group_details = frappe.db.get_value("Item Group", item_group, ["lft", "rgt"], as_dict=1)
if item_group_details:
if item_table:
ig = frappe.qb.DocType("Item Group")
return item_table.item_group.isin(
frappe.qb.from_(ig)
.select(ig.name)
.where(
(ig.lft >= item_group_details.lft)
& (ig.rgt <= item_group_details.rgt)
& (item_table.item_group == ig.name)
)
)
else:
return f"item.item_group in (select ig.name from `tabItem Group` ig \
where ig.lft >= {item_group_details.lft} and ig.rgt <= {item_group_details.rgt} and item.item_group = ig.name)"
def check_inventory_dimension_filters_applied(filters) -> bool:
for dimension in get_inventory_dimensions():
if dimension.fieldname in filters and filters.get(dimension.fieldname):
return True
return False

45
erpnext/stock/report/available_serial_no/test_available_serial_no.py Normal file
View File

@@ -0,0 +1,45 @@
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
import frappe
from frappe.tests import IntegrationTestCase
from frappe.utils import add_days, today
from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
from erpnext.stock.doctype.item.test_item import create_item
from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
class TestStockLedgerReeport(IntegrationTestCase):
def setUp(self) -> None:
item = create_item("_Test Item with Serial No", is_stock_item=1)
item.has_serial_no = 1
item.serial_no_series = "TEST.###"
item.save(ignore_permissions=True)
self.filters = frappe._dict(
company="_Test Company",
from_date=today(),
to_date=add_days(today(), 30),
item_code="_Test Item With Serial No",
)
def tearDown(self) -> None:
frappe.db.rollback()
def test_available_serial_no(self):
report = frappe.get_doc("Report", "Available Serial No")
make_purchase_receipt(qty=10, item_code="_Test Item with Serial No")
data = report.get_data(filters=self.filters)
serial_nos = [item for item in data[-1][-1]["balance_serial_no"].split("\n")]
# Test 1: Since we have created an inward entry with Purchase Receipt of 10 qty, we should have 10 serial nos
self.assertEqual(len(serial_nos), 10)
create_delivery_note(qty=5, item_code="_Test Item with Serial No")
data = report.get_data(filters=self.filters)
serial_nos = [item for item in data[-1][-1]["balance_serial_no"].split("\n")]
# Test 2: Since we have created a delivery note of 5 qty, we should have 5 serial nos
self.assertEqual(len(serial_nos), 5)