refactor: stock balance report
diff --git a/erpnext/stock/doctype/closing_stock_balance/__init__.py b/erpnext/stock/doctype/closing_stock_balance/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/erpnext/stock/doctype/closing_stock_balance/__init__.py
diff --git a/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.js b/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.js
new file mode 100644
index 0000000..6e90884
--- /dev/null
+++ b/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.js
@@ -0,0 +1,23 @@
+// Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and contributors
+// For license information, please see license.txt
+
+frappe.ui.form.on("Closing Stock Balance", {
+ refresh(frm) {
+ frm.trigger("generate_closing_balance");
+ },
+
+ generate_closing_balance(frm) {
+ if (in_list(["Queued", "Failed"], frm.doc.status)) {
+ frm.add_custom_button(__("Generate Closing Stock Balance"), () => {
+ frm.call({
+ method: "enqueue_job",
+ doc: frm.doc,
+ freeze: true,
+ callback: () => {
+ frm.reload_doc();
+ }
+ })
+ })
+ }
+ }
+});
diff --git a/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.json b/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.json
new file mode 100644
index 0000000..225da6d
--- /dev/null
+++ b/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.json
@@ -0,0 +1,148 @@
+{
+ "actions": [],
+ "allow_rename": 1,
+ "autoname": "naming_series:",
+ "creation": "2023-05-17 09:58:42.086911",
+ "default_view": "List",
+ "doctype": "DocType",
+ "editable_grid": 1,
+ "engine": "InnoDB",
+ "field_order": [
+ "naming_series",
+ "company",
+ "status",
+ "column_break_p0s0",
+ "from_date",
+ "to_date",
+ "filters_section",
+ "item_code",
+ "item_group",
+ "include_uom",
+ "column_break_rm5w",
+ "warehouse",
+ "warehouse_type",
+ "amended_from"
+ ],
+ "fields": [
+ {
+ "fieldname": "naming_series",
+ "fieldtype": "Select",
+ "label": "Naming Series",
+ "options": "CBAL-.#####"
+ },
+ {
+ "fieldname": "company",
+ "fieldtype": "Link",
+ "label": "Company",
+ "options": "Company"
+ },
+ {
+ "default": "Draft",
+ "fieldname": "status",
+ "fieldtype": "Select",
+ "in_list_view": 1,
+ "in_preview": 1,
+ "label": "Status",
+ "options": "Draft\nQueued\nIn Progress\nCompleted\nFailed\nCanceled",
+ "read_only": 1
+ },
+ {
+ "fieldname": "column_break_p0s0",
+ "fieldtype": "Column Break"
+ },
+ {
+ "fieldname": "from_date",
+ "fieldtype": "Date",
+ "label": "From Date"
+ },
+ {
+ "fieldname": "to_date",
+ "fieldtype": "Date",
+ "label": "To Date"
+ },
+ {
+ "collapsible": 1,
+ "fieldname": "filters_section",
+ "fieldtype": "Section Break",
+ "label": "Filters"
+ },
+ {
+ "fieldname": "item_code",
+ "fieldtype": "Link",
+ "label": "Item Code",
+ "options": "Item"
+ },
+ {
+ "fieldname": "item_group",
+ "fieldtype": "Link",
+ "label": "Item Group",
+ "options": "Item Group"
+ },
+ {
+ "fieldname": "column_break_rm5w",
+ "fieldtype": "Column Break"
+ },
+ {
+ "fieldname": "warehouse",
+ "fieldtype": "Link",
+ "label": "Warehouse",
+ "options": "Warehouse"
+ },
+ {
+ "fieldname": "warehouse_type",
+ "fieldtype": "Link",
+ "label": "Warehouse Type",
+ "options": "Warehouse Type"
+ },
+ {
+ "fieldname": "amended_from",
+ "fieldtype": "Link",
+ "label": "Amended From",
+ "no_copy": 1,
+ "options": "Closing Stock Balance",
+ "print_hide": 1,
+ "read_only": 1
+ },
+ {
+ "fieldname": "amended_from",
+ "fieldtype": "Link",
+ "label": "Amended From",
+ "no_copy": 1,
+ "options": "Closing Stock Balance",
+ "print_hide": 1,
+ "read_only": 1
+ },
+ {
+ "fieldname": "include_uom",
+ "fieldtype": "Link",
+ "label": "Include UOM",
+ "options": "UOM"
+ }
+ ],
+ "index_web_pages_for_search": 1,
+ "is_submittable": 1,
+ "links": [],
+ "modified": "2023-05-17 11:46:04.448220",
+ "modified_by": "Administrator",
+ "module": "Stock",
+ "name": "Closing Stock Balance",
+ "naming_rule": "By \"Naming Series\" field",
+ "owner": "Administrator",
+ "permissions": [
+ {
+ "create": 1,
+ "delete": 1,
+ "email": 1,
+ "export": 1,
+ "print": 1,
+ "read": 1,
+ "report": 1,
+ "role": "System Manager",
+ "share": 1,
+ "write": 1
+ }
+ ],
+ "sort_field": "modified",
+ "sort_order": "DESC",
+ "states": []
+}
\ No newline at end of file
diff --git a/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.py b/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.py
new file mode 100644
index 0000000..544c9f3
--- /dev/null
+++ b/erpnext/stock/doctype/closing_stock_balance/closing_stock_balance.py
@@ -0,0 +1,118 @@
+# Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and contributors
+# For license information, please see license.txt
+
+import json
+
+import frappe
+from frappe.core.doctype.prepared_report.prepared_report import create_json_gz_file
+from frappe.desk.form.load import get_attachments
+from frappe.model.document import Document
+from frappe.utils import get_link_to_form, gzip_decompress, parse_json
+from frappe.utils.background_jobs import enqueue
+
+from erpnext.stock.report.stock_balance.stock_balance import execute
+
+
+class ClosingStockBalance(Document):
+ def before_save(self):
+ self.set_status()
+
+ def set_status(self, save=False):
+ self.status = "Queued"
+ if self.docstatus == 2:
+ self.status = "Canceled"
+
+ if self.docstatus == 0:
+ self.status = "Draft"
+
+ if save:
+ self.db_set("status", self.status)
+
+ def validate(self):
+ self.validate_duplicate()
+
+ def validate_duplicate(self):
+ table = frappe.qb.DocType("Closing Stock Balance")
+
+ query = (
+ frappe.qb.from_(table)
+ .select(table.name)
+ .where(
+ (table.docstatus == 1)
+ & (table.company == self.company)
+ & (
+ (table.from_date.between(self.from_date, self.to_date))
+ | (table.to_date.between(self.from_date, self.to_date))
+ | (table.from_date >= self.from_date and table.to_date <= self.to_date)
+ )
+ )
+ )
+
+ for fieldname in ["warehouse", "item_code", "item_group", "warehouse_type"]:
+ if self.get(fieldname):
+ query = query.where(table.get(fieldname) == self.get(fieldname))
+
+ query = query.run(as_dict=True)
+
+ if query and query[0].name:
+ name = get_link_to_form("Closing Stock Balance", query[0].name)
+ msg = f"Closing Stock Balance {name} already exists for the selected date range"
+ frappe.throw(msg, title="Duplicate Closing Stock Balance")
+
+ def on_submit(self):
+ self.set_status(save=True)
+ self.enqueue_job()
+
+ def on_cancel(self):
+ self.set_status(save=True)
+
+ @frappe.whitelist()
+ def enqueue_job(self):
+ enqueue(prepare_closing_stock_balance, name=self.name, queue="long", timeout=1500)
+
+ def create_closing_stock_balance_entries(self):
+ columns, data = execute(
+ filters=frappe._dict(
+ {
+ "company": self.company,
+ "from_date": self.from_date,
+ "to_date": self.to_date,
+ "warehouse": self.warehouse,
+ "item_code": self.item_code,
+ "item_group": self.item_group,
+ "warehouse_type": self.warehouse_type,
+ "include_uom": self.include_uom,
+ "ignore_closing_balance": 1,
+ }
+ )
+ )
+
+ create_json_gz_file({"columns": columns, "data": data}, self.doctype, self.name)
+
+ def get_prepared_data(self):
+ if attachments := get_attachments(self.doctype, self.name):
+ attachment = attachments[0]
+ attached_file = frappe.get_doc("File", attachment.name)
+
+ data = gzip_decompress(attached_file.get_content())
+ if data := json.loads(data.decode("utf-8")):
+ data = data
+
+ return parse_json(data)
+
+ return frappe._dict({})
+
+
+def prepare_closing_stock_balance(name):
+ doc = frappe.get_doc("Closing Stock Balance", name)
+
+ doc.db_set("status", "In Progress")
+
+ try:
+ doc.create_closing_stock_balance_entries()
+ doc.db_set("status", "Completed")
+ except Exception as e:
+ doc.db_set("status", "Failed")
+ traceback = frappe.get_traceback()
+
+ frappe.log_error("Closing Stock Balance Failed", traceback, doc.doctype, doc.name)
diff --git a/erpnext/stock/doctype/closing_stock_balance/test_closing_stock_balance.py b/erpnext/stock/doctype/closing_stock_balance/test_closing_stock_balance.py
new file mode 100644
index 0000000..7d61f5c
--- /dev/null
+++ b/erpnext/stock/doctype/closing_stock_balance/test_closing_stock_balance.py
@@ -0,0 +1,9 @@
+# Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and Contributors
+# See license.txt
+
+# import frappe
+from frappe.tests.utils import FrappeTestCase
+
+
+class TestClosingStockBalance(FrappeTestCase):
+ pass
diff --git a/erpnext/stock/report/stock_balance/stock_balance.js b/erpnext/stock/report/stock_balance/stock_balance.js
index 9b3965d..a2b3b91 100644
--- a/erpnext/stock/report/stock_balance/stock_balance.js
+++ b/erpnext/stock/report/stock_balance/stock_balance.js
@@ -87,6 +87,11 @@
"label": __('Show Stock Ageing Data'),
"fieldtype": 'Check'
},
+ {
+ "fieldname": 'ignore_closing_balance',
+ "label": __('Ignore Closing Balance'),
+ "fieldtype": 'Check'
+ },
],
"formatter": function (value, row, column, data, default_formatter) {
diff --git a/erpnext/stock/report/stock_balance/stock_balance.py b/erpnext/stock/report/stock_balance/stock_balance.py
index 66991a9..f24e232 100644
--- a/erpnext/stock/report/stock_balance/stock_balance.py
+++ b/erpnext/stock/report/stock_balance/stock_balance.py
@@ -7,15 +7,16 @@
import frappe
from frappe import _
+from frappe.query_builder import Order
from frappe.query_builder.functions import Coalesce, CombineDatetime
-from frappe.utils import cint, date_diff, flt, getdate
+from frappe.utils import add_days, cint, date_diff, flt, getdate
from frappe.utils.nestedset import get_descendants_of
import erpnext
from erpnext.stock.doctype.inventory_dimension.inventory_dimension import get_inventory_dimensions
from erpnext.stock.doctype.warehouse.warehouse import apply_warehouse_filter
from erpnext.stock.report.stock_ageing.stock_ageing import FIFOSlots, get_average_age
-from erpnext.stock.utils import add_additional_uom_columns, is_reposting_item_valuation_in_progress
+from erpnext.stock.utils import add_additional_uom_columns
class StockBalanceFilter(TypedDict):
@@ -35,68 +36,83 @@
def execute(filters: Optional[StockBalanceFilter] = None):
- is_reposting_item_valuation_in_progress()
- if not filters:
- filters = {}
+ return StockBalanceReport(filters).run()
- if filters.get("company"):
- company_currency = erpnext.get_company_currency(filters.get("company"))
- else:
- company_currency = frappe.db.get_single_value("Global Defaults", "default_currency")
- include_uom = filters.get("include_uom")
- columns = get_columns(filters)
- items = get_items(filters)
- sle = get_stock_ledger_entries(filters, items)
+class StockBalanceReport(object):
+ def __init__(self, filters: Optional[StockBalanceFilter]) -> None:
+ self.filters = filters
+ self.from_date = getdate(filters.get("from_date"))
+ self.to_date = getdate(filters.get("to_date"))
- if filters.get("show_stock_ageing_data"):
- filters["show_warehouse_wise_stock"] = True
- item_wise_fifo_queue = FIFOSlots(filters, sle).generate()
+ self.start_from = None
+ self.data = []
+ self.columns = []
+ self.sle_entries: List[SLEntry] = []
+ self.set_company_currency()
- # if no stock ledger entry found return
- if not sle:
- return columns, []
+ def set_company_currency(self) -> None:
+ if self.filters.get("company"):
+ self.company_currency = erpnext.get_company_currency(self.filters.get("company"))
+ else:
+ self.company_currency = frappe.db.get_single_value("Global Defaults", "default_currency")
- iwb_map = get_item_warehouse_map(filters, sle)
- item_map = get_item_details(items, sle, filters)
- item_reorder_detail_map = get_item_reorder_details(item_map.keys())
+ def run(self):
+ self.float_precision = cint(frappe.db.get_default("float_precision")) or 3
- data = []
- conversion_factors = {}
+ self.inventory_dimensions = get_inventory_dimension_fields()
+ self.prepare_opening_data_from_closing_balance()
+ self.prepare_stock_ledger_entries()
+ self.prepare_new_data()
- _func = itemgetter(1)
+ if not self.columns:
+ self.columns = self.get_columns()
- to_date = filters.get("to_date")
+ self.add_additional_uom_columns()
- for group_by_key in iwb_map:
- item = group_by_key[1]
- warehouse = group_by_key[2]
- company = group_by_key[0]
+ return self.columns, self.data
- if item_map.get(item):
- qty_dict = iwb_map[group_by_key]
- item_reorder_level = 0
- item_reorder_qty = 0
- if item + warehouse in item_reorder_detail_map:
- item_reorder_level = item_reorder_detail_map[item + warehouse]["warehouse_reorder_level"]
- item_reorder_qty = item_reorder_detail_map[item + warehouse]["warehouse_reorder_qty"]
+ def prepare_opening_data_from_closing_balance(self) -> None:
+ self.opening_data = frappe._dict({})
- report_data = {
- "currency": company_currency,
- "item_code": item,
- "warehouse": warehouse,
- "company": company,
- "reorder_level": item_reorder_level,
- "reorder_qty": item_reorder_qty,
- }
- report_data.update(item_map[item])
- report_data.update(qty_dict)
+ closing_balance = self.get_closing_balance()
+ if not closing_balance:
+ return
- if include_uom:
- conversion_factors.setdefault(item, item_map[item].conversion_factor)
+ self.start_from = add_days(closing_balance[0].to_date, 1)
+ res = frappe.get_doc("Closing Stock Balance", closing_balance[0].name).get_prepared_data()
- if filters.get("show_stock_ageing_data"):
- fifo_queue = item_wise_fifo_queue[(item, warehouse)].get("fifo_queue")
+ for entry in res.data:
+ entry = frappe._dict(entry)
+
+ group_by_key = self.get_group_by_key(entry)
+ if group_by_key not in self.opening_data:
+ self.opening_data.setdefault(group_by_key, entry)
+
+ def prepare_new_data(self):
+ if not self.sle_entries:
+ return
+
+ if self.filters.get("show_stock_ageing_data"):
+ self.filters["show_warehouse_wise_stock"] = True
+ item_wise_fifo_queue = FIFOSlots(self.filters, self.sle_entries).generate()
+
+ _func = itemgetter(1)
+
+ self.item_warehouse_map = self.get_item_warehouse_map()
+
+ variant_values = {}
+ if self.filters.get("show_variant_attributes"):
+ variant_values = self.get_variant_values_for()
+
+ for key, report_data in self.item_warehouse_map.items():
+ if variant_data := variant_values.get(report_data.item_code):
+ report_data.update(variant_data)
+
+ if self.filters.get("show_stock_ageing_data"):
+ fifo_queue = item_wise_fifo_queue[(report_data.item_code, report_data.warehouse)].get(
+ "fifo_queue"
+ )
stock_ageing_data = {"average_age": 0, "earliest_age": 0, "latest_age": 0}
if fifo_queue:
@@ -104,321 +120,445 @@
if not fifo_queue:
continue
+ to_date = self.to_date
stock_ageing_data["average_age"] = get_average_age(fifo_queue, to_date)
stock_ageing_data["earliest_age"] = date_diff(to_date, fifo_queue[0][1])
stock_ageing_data["latest_age"] = date_diff(to_date, fifo_queue[-1][1])
report_data.update(stock_ageing_data)
- data.append(report_data)
+ self.data.append(report_data)
- add_additional_uom_columns(columns, data, include_uom, conversion_factors)
- return columns, data
+ def get_item_warehouse_map(self):
+ item_warehouse_map = {}
+ for entry in self.sle_entries:
+ group_by_key = self.get_group_by_key(entry)
+ if group_by_key not in item_warehouse_map:
+ self.initialize_data(item_warehouse_map, group_by_key, entry)
-def get_columns(filters: StockBalanceFilter):
- """return columns"""
- columns = [
- {
- "label": _("Item"),
- "fieldname": "item_code",
- "fieldtype": "Link",
- "options": "Item",
- "width": 100,
- },
- {"label": _("Item Name"), "fieldname": "item_name", "width": 150},
- {
- "label": _("Item Group"),
- "fieldname": "item_group",
- "fieldtype": "Link",
- "options": "Item Group",
- "width": 100,
- },
- {
- "label": _("Warehouse"),
- "fieldname": "warehouse",
- "fieldtype": "Link",
- "options": "Warehouse",
- "width": 100,
- },
- ]
+ self.prepare_item_warehouse_map(item_warehouse_map, entry, group_by_key)
- for dimension in get_inventory_dimensions():
- columns.append(
- {
- "label": _(dimension.doctype),
- "fieldname": dimension.fieldname,
- "fieldtype": "Link",
- "options": dimension.doctype,
- "width": 110,
- }
+ if self.opening_data.get(group_by_key):
+ del self.opening_data[group_by_key]
+
+ for group_by_key, entry in self.opening_data.items():
+ if group_by_key not in item_warehouse_map:
+ self.initialize_data(item_warehouse_map, group_by_key, entry)
+
+ qty_dict = item_warehouse_map[group_by_key]
+ qty_dict.val_rate = entry.valuation_rate
+ qty_dict.bal_qty += flt(qty_dict.opening_qty)
+ qty_dict.bal_val += flt(qty_dict.opening_val)
+
+ item_warehouse_map = filter_items_with_no_transactions(
+ item_warehouse_map, self.float_precision, self.inventory_dimensions
)
- columns.extend(
- [
- {
- "label": _("Stock UOM"),
- "fieldname": "stock_uom",
- "fieldtype": "Link",
- "options": "UOM",
- "width": 90,
- },
- {
- "label": _("Balance Qty"),
- "fieldname": "bal_qty",
- "fieldtype": "Float",
- "width": 100,
- "convertible": "qty",
- },
- {
- "label": _("Balance Value"),
- "fieldname": "bal_val",
- "fieldtype": "Currency",
- "width": 100,
- "options": "currency",
- },
- {
- "label": _("Opening Qty"),
- "fieldname": "opening_qty",
- "fieldtype": "Float",
- "width": 100,
- "convertible": "qty",
- },
- {
- "label": _("Opening Value"),
- "fieldname": "opening_val",
- "fieldtype": "Currency",
- "width": 110,
- "options": "currency",
- },
- {
- "label": _("In Qty"),
- "fieldname": "in_qty",
- "fieldtype": "Float",
- "width": 80,
- "convertible": "qty",
- },
- {"label": _("In Value"), "fieldname": "in_val", "fieldtype": "Float", "width": 80},
- {
- "label": _("Out Qty"),
- "fieldname": "out_qty",
- "fieldtype": "Float",
- "width": 80,
- "convertible": "qty",
- },
- {"label": _("Out Value"), "fieldname": "out_val", "fieldtype": "Float", "width": 80},
- {
- "label": _("Valuation Rate"),
- "fieldname": "val_rate",
- "fieldtype": "Currency",
- "width": 90,
- "convertible": "rate",
- "options": "currency",
- },
- {
- "label": _("Reorder Level"),
- "fieldname": "reorder_level",
- "fieldtype": "Float",
- "width": 80,
- "convertible": "qty",
- },
- {
- "label": _("Reorder Qty"),
- "fieldname": "reorder_qty",
- "fieldtype": "Float",
- "width": 80,
- "convertible": "qty",
- },
- {
- "label": _("Company"),
- "fieldname": "company",
- "fieldtype": "Link",
- "options": "Company",
- "width": 100,
- },
- ]
- )
+ return item_warehouse_map
- if filters.get("show_stock_ageing_data"):
- columns += [
- {"label": _("Average Age"), "fieldname": "average_age", "width": 100},
- {"label": _("Earliest Age"), "fieldname": "earliest_age", "width": 100},
- {"label": _("Latest Age"), "fieldname": "latest_age", "width": 100},
- ]
+ def prepare_item_warehouse_map(self, item_warehouse_map, entry, group_by_key):
+ opening_vouchers = self.get_opening_vouchers()
- if filters.get("show_variant_attributes"):
- columns += [
- {"label": att_name, "fieldname": att_name, "width": 100}
- for att_name in get_variants_attributes()
- ]
+ qty_dict = item_warehouse_map[group_by_key]
+ for field in self.inventory_dimensions:
+ qty_dict[field] = entry.get(field)
- return columns
-
-
-def apply_conditions(query, filters):
- sle = frappe.qb.DocType("Stock Ledger Entry")
- warehouse_table = frappe.qb.DocType("Warehouse")
-
- if not filters.get("from_date"):
- frappe.throw(_("'From Date' is required"))
-
- if to_date := filters.get("to_date"):
- query = query.where(sle.posting_date <= to_date)
- else:
- frappe.throw(_("'To Date' is required"))
-
- if company := filters.get("company"):
- query = query.where(sle.company == company)
-
- if filters.get("warehouse"):
- query = apply_warehouse_filter(query, sle, filters)
- elif warehouse_type := filters.get("warehouse_type"):
- query = (
- query.join(warehouse_table)
- .on(warehouse_table.name == sle.warehouse)
- .where(warehouse_table.warehouse_type == warehouse_type)
- )
-
- return query
-
-
-def get_stock_ledger_entries(filters: StockBalanceFilter, items: List[str]) -> List[SLEntry]:
- sle = frappe.qb.DocType("Stock Ledger Entry")
-
- query = (
- frappe.qb.from_(sle)
- .select(
- sle.item_code,
- sle.warehouse,
- sle.posting_date,
- sle.actual_qty,
- sle.valuation_rate,
- sle.company,
- sle.voucher_type,
- sle.qty_after_transaction,
- sle.stock_value_difference,
- sle.item_code.as_("name"),
- sle.voucher_no,
- sle.stock_value,
- sle.batch_no,
- )
- .where((sle.docstatus < 2) & (sle.is_cancelled == 0))
- .orderby(CombineDatetime(sle.posting_date, sle.posting_time))
- .orderby(sle.creation)
- .orderby(sle.actual_qty)
- )
-
- 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))
-
- query = apply_conditions(query, filters)
- return query.run(as_dict=True)
-
-
-def get_opening_vouchers(to_date):
- opening_vouchers = {"Stock Entry": [], "Stock Reconciliation": []}
-
- se = frappe.qb.DocType("Stock Entry")
- sr = frappe.qb.DocType("Stock Reconciliation")
-
- vouchers_data = (
- frappe.qb.from_(
- (
- frappe.qb.from_(se)
- .select(se.name, Coalesce("Stock Entry").as_("voucher_type"))
- .where((se.docstatus == 1) & (se.posting_date <= to_date) & (se.is_opening == "Yes"))
- )
- + (
- frappe.qb.from_(sr)
- .select(sr.name, Coalesce("Stock Reconciliation").as_("voucher_type"))
- .where((sr.docstatus == 1) & (sr.posting_date <= to_date) & (sr.purpose == "Opening Stock"))
- )
- ).select("voucher_type", "name")
- ).run(as_dict=True)
-
- if vouchers_data:
- for d in vouchers_data:
- opening_vouchers[d.voucher_type].append(d.name)
-
- return opening_vouchers
-
-
-def get_inventory_dimension_fields():
- return [dimension.fieldname for dimension in get_inventory_dimensions()]
-
-
-def get_item_warehouse_map(filters: StockBalanceFilter, sle: List[SLEntry]):
- iwb_map = {}
- from_date = getdate(filters.get("from_date"))
- to_date = getdate(filters.get("to_date"))
- opening_vouchers = get_opening_vouchers(to_date)
- float_precision = cint(frappe.db.get_default("float_precision")) or 3
- inventory_dimensions = get_inventory_dimension_fields()
-
- for d in sle:
- group_by_key = get_group_by_key(d, filters, inventory_dimensions)
- if group_by_key not in iwb_map:
- iwb_map[group_by_key] = frappe._dict(
- {
- "opening_qty": 0.0,
- "opening_val": 0.0,
- "in_qty": 0.0,
- "in_val": 0.0,
- "out_qty": 0.0,
- "out_val": 0.0,
- "bal_qty": 0.0,
- "bal_val": 0.0,
- "val_rate": 0.0,
- }
- )
-
- qty_dict = iwb_map[group_by_key]
- for field in inventory_dimensions:
- qty_dict[field] = d.get(field)
-
- if d.voucher_type == "Stock Reconciliation" and not d.batch_no:
- qty_diff = flt(d.qty_after_transaction) - flt(qty_dict.bal_qty)
+ if entry.voucher_type == "Stock Reconciliation" and not entry.batch_no:
+ qty_diff = flt(entry.qty_after_transaction) - flt(qty_dict.bal_qty)
else:
- qty_diff = flt(d.actual_qty)
+ qty_diff = flt(entry.actual_qty)
- value_diff = flt(d.stock_value_difference)
+ value_diff = flt(entry.stock_value_difference)
- if d.posting_date < from_date or d.voucher_no in opening_vouchers.get(d.voucher_type, []):
+ if entry.posting_date < self.from_date or entry.voucher_no in opening_vouchers.get(
+ entry.voucher_type, []
+ ):
qty_dict.opening_qty += qty_diff
qty_dict.opening_val += value_diff
- elif d.posting_date >= from_date and d.posting_date <= to_date:
- if flt(qty_diff, float_precision) >= 0:
+ elif entry.posting_date >= self.from_date and entry.posting_date <= self.to_date:
+
+ if flt(qty_diff, self.float_precision) >= 0:
qty_dict.in_qty += qty_diff
qty_dict.in_val += value_diff
else:
qty_dict.out_qty += abs(qty_diff)
qty_dict.out_val += abs(value_diff)
- qty_dict.val_rate = d.valuation_rate
+ # if self.opening_data and group_by_key in self.opening_data:
+ # qty_diff += flt(qty_dict.opening_qty)
+ # value_diff += flt(qty_dict.opening_val)
+
+ qty_dict.val_rate = entry.valuation_rate
qty_dict.bal_qty += qty_diff
qty_dict.bal_val += value_diff
- iwb_map = filter_items_with_no_transactions(iwb_map, float_precision, inventory_dimensions)
+ def initialize_data(self, item_warehouse_map, group_by_key, entry):
+ opening_data = self.opening_data.get(group_by_key, {})
- return iwb_map
+ item_warehouse_map[group_by_key] = frappe._dict(
+ {
+ "item_code": entry.item_code,
+ "warehouse": entry.warehouse,
+ "item_group": entry.item_group,
+ "company": entry.company,
+ "currency": self.company_currency,
+ "stock_uom": entry.stock_uom,
+ "item_name": entry.item_name,
+ "opening_qty": opening_data.get("bal_qty") or 0.0,
+ "opening_val": opening_data.get("bal_val") or 0.0,
+ "in_qty": 0.0,
+ "in_val": 0.0,
+ "out_qty": 0.0,
+ "out_val": 0.0,
+ "bal_qty": 0.0,
+ "bal_val": 0.0,
+ "val_rate": 0.0,
+ }
+ )
+
+ def get_group_by_key(self, row) -> tuple:
+ group_by_key = [row.company, row.item_code, row.warehouse]
+
+ for fieldname in self.inventory_dimensions:
+ if self.filters.get(fieldname):
+ group_by_key.append(row.get(fieldname))
+
+ return tuple(group_by_key)
+
+ def get_closing_balance(self) -> List[Dict[str, Any]]:
+ if self.filters.get("ignore_closing_balance"):
+ return []
+
+ table = frappe.qb.DocType("Closing Stock Balance")
+
+ query = (
+ frappe.qb.from_(table)
+ .select(table.name, table.to_date)
+ .where(
+ (table.docstatus == 1)
+ & (table.company == self.filters.company)
+ & ((table.to_date <= self.from_date))
+ )
+ .orderby(table.to_date, order=Order.desc)
+ .limit(1)
+ )
+
+ for fieldname in ["warehouse", "item_code", "item_group", "warehouse_type"]:
+ if self.filters.get(fieldname):
+ query = query.where(table[fieldname] == self.filters.get(fieldname))
+
+ return query.run(as_dict=True)
+
+ def prepare_stock_ledger_entries(self):
+ sle = frappe.qb.DocType("Stock Ledger Entry")
+ item_table = frappe.qb.DocType("Item")
+
+ query = (
+ frappe.qb.from_(sle)
+ .inner_join(item_table)
+ .on(sle.item_code == item_table.name)
+ .select(
+ sle.item_code,
+ sle.warehouse,
+ sle.posting_date,
+ sle.actual_qty,
+ sle.valuation_rate,
+ sle.company,
+ sle.voucher_type,
+ sle.qty_after_transaction,
+ sle.stock_value_difference,
+ sle.item_code.as_("name"),
+ sle.voucher_no,
+ sle.stock_value,
+ sle.batch_no,
+ item_table.item_group,
+ item_table.stock_uom,
+ item_table.item_name,
+ )
+ .where((sle.docstatus < 2) & (sle.is_cancelled == 0))
+ .orderby(CombineDatetime(sle.posting_date, sle.posting_time))
+ .orderby(sle.creation)
+ .orderby(sle.actual_qty)
+ )
+
+ query = self.apply_inventory_dimensions_filters(query, sle)
+ query = self.apply_warehouse_filters(query, sle)
+ query = self.apply_items_filters(query, item_table)
+ query = self.apply_date_filters(query, sle)
+
+ if self.filters.get("company"):
+ query = query.where(sle.company == self.filters.get("company"))
+
+ self.sle_entries = query.run(as_dict=True)
+
+ def apply_inventory_dimensions_filters(self, query, sle) -> str:
+ inventory_dimension_fields = get_inventory_dimension_fields()
+ if inventory_dimension_fields:
+ for fieldname in inventory_dimension_fields:
+ query = query.select(fieldname)
+ if self.filters.get(fieldname):
+ query = query.where(sle[fieldname].isin(self.filters.get(fieldname)))
+
+ return query
+
+ def apply_warehouse_filters(self, query, sle) -> str:
+ warehouse_table = frappe.qb.DocType("Warehouse")
+
+ if self.filters.get("warehouse"):
+ query = apply_warehouse_filter(query, sle, self.filters)
+ elif warehouse_type := self.filters.get("warehouse_type"):
+ query = (
+ query.join(warehouse_table)
+ .on(warehouse_table.name == sle.warehouse)
+ .where(warehouse_table.warehouse_type == warehouse_type)
+ )
+
+ return query
+
+ def apply_items_filters(self, query, item_table) -> str:
+ if item_group := self.filters.get("item_group"):
+ children = get_descendants_of("Item Group", item_group, ignore_permissions=True)
+ query = query.where(item_table.item_group.isin(children + [item_group]))
+
+ for field in ["item_code", "brand"]:
+ if not self.filters.get(field):
+ continue
+
+ query = query.where(item_table[field] == self.filters.get(field))
+
+ return query
+
+ def apply_date_filters(self, query, sle) -> str:
+ if not self.filters.ignore_closing_balance and self.start_from:
+ query = query.where(sle.posting_date >= self.start_from)
+
+ if self.to_date:
+ query = query.where(sle.posting_date <= self.to_date)
+
+ return query
+
+ def get_columns(self):
+ columns = [
+ {
+ "label": _("Item"),
+ "fieldname": "item_code",
+ "fieldtype": "Link",
+ "options": "Item",
+ "width": 100,
+ },
+ {"label": _("Item Name"), "fieldname": "item_name", "width": 150},
+ {
+ "label": _("Item Group"),
+ "fieldname": "item_group",
+ "fieldtype": "Link",
+ "options": "Item Group",
+ "width": 100,
+ },
+ {
+ "label": _("Warehouse"),
+ "fieldname": "warehouse",
+ "fieldtype": "Link",
+ "options": "Warehouse",
+ "width": 100,
+ },
+ ]
+
+ for dimension in get_inventory_dimensions():
+ columns.append(
+ {
+ "label": _(dimension.doctype),
+ "fieldname": dimension.fieldname,
+ "fieldtype": "Link",
+ "options": dimension.doctype,
+ "width": 110,
+ }
+ )
+
+ columns.extend(
+ [
+ {
+ "label": _("Stock UOM"),
+ "fieldname": "stock_uom",
+ "fieldtype": "Link",
+ "options": "UOM",
+ "width": 90,
+ },
+ {
+ "label": _("Balance Qty"),
+ "fieldname": "bal_qty",
+ "fieldtype": "Float",
+ "width": 100,
+ "convertible": "qty",
+ },
+ {
+ "label": _("Balance Value"),
+ "fieldname": "bal_val",
+ "fieldtype": "Currency",
+ "width": 100,
+ "options": "currency",
+ },
+ {
+ "label": _("Opening Qty"),
+ "fieldname": "opening_qty",
+ "fieldtype": "Float",
+ "width": 100,
+ "convertible": "qty",
+ },
+ {
+ "label": _("Opening Value"),
+ "fieldname": "opening_val",
+ "fieldtype": "Currency",
+ "width": 110,
+ "options": "currency",
+ },
+ {
+ "label": _("In Qty"),
+ "fieldname": "in_qty",
+ "fieldtype": "Float",
+ "width": 80,
+ "convertible": "qty",
+ },
+ {"label": _("In Value"), "fieldname": "in_val", "fieldtype": "Float", "width": 80},
+ {
+ "label": _("Out Qty"),
+ "fieldname": "out_qty",
+ "fieldtype": "Float",
+ "width": 80,
+ "convertible": "qty",
+ },
+ {"label": _("Out Value"), "fieldname": "out_val", "fieldtype": "Float", "width": 80},
+ {
+ "label": _("Valuation Rate"),
+ "fieldname": "val_rate",
+ "fieldtype": "Currency",
+ "width": 90,
+ "convertible": "rate",
+ "options": "currency",
+ },
+ {
+ "label": _("Reorder Level"),
+ "fieldname": "reorder_level",
+ "fieldtype": "Float",
+ "width": 80,
+ "convertible": "qty",
+ },
+ {
+ "label": _("Reorder Qty"),
+ "fieldname": "reorder_qty",
+ "fieldtype": "Float",
+ "width": 80,
+ "convertible": "qty",
+ },
+ {
+ "label": _("Company"),
+ "fieldname": "company",
+ "fieldtype": "Link",
+ "options": "Company",
+ "width": 100,
+ },
+ ]
+ )
+
+ if self.filters.get("show_stock_ageing_data"):
+ columns += [
+ {"label": _("Average Age"), "fieldname": "average_age", "width": 100},
+ {"label": _("Earliest Age"), "fieldname": "earliest_age", "width": 100},
+ {"label": _("Latest Age"), "fieldname": "latest_age", "width": 100},
+ ]
+
+ if self.filters.get("show_variant_attributes"):
+ columns += [
+ {"label": att_name, "fieldname": att_name, "width": 100}
+ for att_name in get_variants_attributes()
+ ]
+
+ return columns
+
+ def add_additional_uom_columns(self):
+ if not self.filters.get("include_uom"):
+ return
+
+ conversion_factors = self.get_itemwise_conversion_factor()
+ add_additional_uom_columns(self.columns, self.data, self.filters.include_uom, conversion_factors)
+
+ def get_itemwise_conversion_factor(self):
+ items = []
+ if self.filters.item_code or self.filters.item_group:
+ items = [d.item_code for d in self.data]
+
+ table = frappe.qb.DocType("UOM Conversion Detail")
+ query = (
+ frappe.qb.from_(table)
+ .select(table.conversion_factor)
+ .where((table.parenttype == "Item") & (table.uom == self.filters.include_uom))
+ )
+
+ if items:
+ query = query.where(table.parent.isin(items))
+
+ result = query.run(as_dict=1)
+ if not result:
+ return {}
+
+ return {d.parent: d.conversion_factor for d in result}
+
+ def get_variant_values_for(self):
+ """Returns variant values for items."""
+ attribute_map = {}
+ items = []
+ if self.filters.item_code or self.filters.item_group:
+ items = [d.item_code for d in self.data]
+
+ filters = {}
+ if items:
+ filters = {"parent": ("in", items)}
+
+ attribute_info = frappe.get_all(
+ "Item Variant Attribute",
+ fields=["parent", "attribute", "attribute_value"],
+ filters=filters,
+ )
+
+ for attr in attribute_info:
+ attribute_map.setdefault(attr["parent"], {})
+ attribute_map[attr["parent"]].update({attr["attribute"]: attr["attribute_value"]})
+
+ return attribute_map
+
+ def get_opening_vouchers(self):
+ opening_vouchers = {"Stock Entry": [], "Stock Reconciliation": []}
+
+ se = frappe.qb.DocType("Stock Entry")
+ sr = frappe.qb.DocType("Stock Reconciliation")
+
+ vouchers_data = (
+ frappe.qb.from_(
+ (
+ frappe.qb.from_(se)
+ .select(se.name, Coalesce("Stock Entry").as_("voucher_type"))
+ .where((se.docstatus == 1) & (se.posting_date <= self.to_date) & (se.is_opening == "Yes"))
+ )
+ + (
+ frappe.qb.from_(sr)
+ .select(sr.name, Coalesce("Stock Reconciliation").as_("voucher_type"))
+ .where(
+ (sr.docstatus == 1) & (sr.posting_date <= self.to_date) & (sr.purpose == "Opening Stock")
+ )
+ )
+ ).select("voucher_type", "name")
+ ).run(as_dict=True)
+
+ if vouchers_data:
+ for d in vouchers_data:
+ opening_vouchers[d.voucher_type].append(d.name)
+
+ return opening_vouchers
-def get_group_by_key(row, filters, inventory_dimension_fields) -> tuple:
- group_by_key = [row.company, row.item_code, row.warehouse]
-
- for fieldname in inventory_dimension_fields:
- if filters.get(fieldname):
- group_by_key.append(row.get(fieldname))
-
- return tuple(group_by_key)
+def get_inventory_dimension_fields():
+ return [dimension.fieldname for dimension in get_inventory_dimensions()]
def filter_items_with_no_transactions(iwb_map, float_precision: float, inventory_dimensions: list):
@@ -431,6 +571,9 @@
if key in inventory_dimensions:
continue
+ if key in ["item_code", "warehouse", "item_name", "item_group", "projecy", "stock_uom"]:
+ continue
+
val = flt(val, float_precision)
qty_dict[key] = val
if key != "val_rate" and val:
@@ -445,96 +588,6 @@
return iwb_map
-def get_items(filters: StockBalanceFilter) -> List[str]:
- "Get items based on item code, item group or brand."
- if item_code := filters.get("item_code"):
- return [item_code]
- else:
- item_filters = {}
- if item_group := filters.get("item_group"):
- children = get_descendants_of("Item Group", item_group, ignore_permissions=True)
- item_filters["item_group"] = ("in", children + [item_group])
- if brand := filters.get("brand"):
- item_filters["brand"] = brand
-
- return frappe.get_all("Item", filters=item_filters, pluck="name", order_by=None)
-
-
-def get_item_details(items: List[str], sle: List[SLEntry], filters: StockBalanceFilter):
- item_details = {}
- if not items:
- items = list(set(d.item_code for d in sle))
-
- if not items:
- return item_details
-
- item_table = frappe.qb.DocType("Item")
-
- query = (
- frappe.qb.from_(item_table)
- .select(
- item_table.name,
- item_table.item_name,
- item_table.description,
- item_table.item_group,
- item_table.brand,
- item_table.stock_uom,
- )
- .where(item_table.name.isin(items))
- )
-
- if uom := filters.get("include_uom"):
- uom_conv_detail = frappe.qb.DocType("UOM Conversion Detail")
- query = (
- query.left_join(uom_conv_detail)
- .on((uom_conv_detail.parent == item_table.name) & (uom_conv_detail.uom == uom))
- .select(uom_conv_detail.conversion_factor)
- )
-
- result = query.run(as_dict=1)
-
- for item_table in result:
- item_details.setdefault(item_table.name, item_table)
-
- if filters.get("show_variant_attributes"):
- variant_values = get_variant_values_for(list(item_details))
- item_details = {k: v.update(variant_values.get(k, {})) for k, v in item_details.items()}
-
- return item_details
-
-
-def get_item_reorder_details(items):
- item_reorder_details = frappe._dict()
-
- if items:
- item_reorder_details = frappe.get_all(
- "Item Reorder",
- ["parent", "warehouse", "warehouse_reorder_qty", "warehouse_reorder_level"],
- filters={"parent": ("in", items)},
- )
-
- return dict((d.parent + d.warehouse, d) for d in item_reorder_details)
-
-
def get_variants_attributes() -> List[str]:
"""Return all item variant attributes."""
return frappe.get_all("Item Attribute", pluck="name")
-
-
-def get_variant_values_for(items):
- """Returns variant values for items."""
- attribute_map = {}
-
- attribute_info = frappe.get_all(
- "Item Variant Attribute",
- ["parent", "attribute", "attribute_value"],
- {
- "parent": ("in", items),
- },
- )
-
- for attr in attribute_info:
- attribute_map.setdefault(attr["parent"], {})
- attribute_map[attr["parent"]].update({attr["attribute"]: attr["attribute_value"]})
-
- return attribute_map