erpnext/stock/stock_balance.py

# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt


import frappe
from frappe.query_builder import DocType
from frappe.query_builder.functions import Sum
from frappe.query_builder.utils import Table
from frappe.utils import cstr, flt, now, nowdate, nowtime
from pypika.queries import QueryBuilder

from erpnext.controllers.stock_controller import create_repost_item_valuation_entry


def repost(only_actual=False, allow_negative_stock=False, allow_zero_rate=False, only_bin=False):
	"""
	Repost everything!
	"""
	frappe.db.auto_commit_on_many_writes = 1

	if allow_negative_stock:
		existing_allow_negative_stock = frappe.db.get_value(
			"Stock Settings", None, "allow_negative_stock"
		)
		frappe.db.set_value("Stock Settings", None, "allow_negative_stock", 1)

	item_warehouses = frappe.db.sql(
		"""
		select distinct item_code, warehouse
		from
			(select item_code, warehouse from tabBin
			union
			select item_code, warehouse from `tabStock Ledger Entry`) a
	"""
	)
	for d in item_warehouses:
		try:
			repost_stock(d[0], d[1], allow_zero_rate, only_actual, only_bin, allow_negative_stock)
			frappe.db.commit()
		except Exception:
			frappe.db.rollback()

	if allow_negative_stock:
		frappe.db.set_value(
			"Stock Settings", None, "allow_negative_stock", existing_allow_negative_stock
		)
	frappe.db.auto_commit_on_many_writes = 0


def repost_stock(
	item_code,
	warehouse,
	allow_zero_rate=False,
	only_actual=False,
	only_bin=False,
	allow_negative_stock=False,
):

	if not only_bin:
		repost_actual_qty(item_code, warehouse, allow_zero_rate, allow_negative_stock)

	if item_code and warehouse and not only_actual:
		qty_dict = {
			"reserved_qty": get_reserved_qty(item_code, warehouse),
			"indented_qty": get_indented_qty(item_code, warehouse),
			"ordered_qty": get_ordered_qty(item_code, warehouse),
			"planned_qty": get_planned_qty(item_code, warehouse),
		}
		if only_bin:
			qty_dict.update({"actual_qty": get_balance_qty_from_sle(item_code, warehouse)})

		update_bin_qty(item_code, warehouse, qty_dict)


def repost_actual_qty(item_code, warehouse, allow_zero_rate=False, allow_negative_stock=False):
	create_repost_item_valuation_entry(
		{
			"item_code": item_code,
			"warehouse": warehouse,
			"posting_date": "1900-01-01",
			"posting_time": "00:01",
			"allow_negative_stock": allow_negative_stock,
			"allow_zero_rate": allow_zero_rate,
		}
	)


def get_balance_qty_from_sle(item_code, warehouse):
	balance_qty = frappe.db.sql(
		"""select qty_after_transaction from `tabStock Ledger Entry`
		where item_code=%s and warehouse=%s and is_cancelled=0
		order by posting_date desc, posting_time desc, creation desc
		limit 1""",
		(item_code, warehouse),
	)

	return flt(balance_qty[0][0]) if balance_qty else 0.0


def get_reserved_qty(item_code, warehouse):
	SalesOrder = DocType("Sales Order")
	SalesOrderItem = DocType("Sales Order Item")
	PackedItem = DocType("Packed Item")

	def append_open_so_query(q: QueryBuilder, child_table: Table) -> QueryBuilder:
		return (
			q.inner_join(SalesOrder)
			.on(SalesOrder.name == child_table.parent)
			.where(SalesOrder.docstatus == 1)
			.where(SalesOrder.status != "Closed")
		)

	tab = (
		frappe.qb.from_(SalesOrderItem)
		.select(
			SalesOrderItem.stock_qty.as_("dnpi_qty"),
			SalesOrderItem.qty.as_("so_item_qty"),
			SalesOrderItem.delivered_qty.as_("so_item_delivered_qty"),
			SalesOrderItem.returned_qty.as_("so_item_returned_qty"),
			SalesOrderItem.parent,
			SalesOrderItem.name,
		)
		.where(SalesOrderItem.item_code == item_code)
		.where(SalesOrderItem.warehouse == warehouse)
	)
	tab = append_open_so_query(tab, SalesOrderItem)

	dnpi = (
		frappe.qb.from_(PackedItem)
		.select(PackedItem.qty, PackedItem.parent_detail_docname, PackedItem.parent, PackedItem.name)
		.where(PackedItem.item_code == item_code)
		.where(PackedItem.warehouse == warehouse)
	)
	dnpi = append_open_so_query(dnpi, PackedItem)

	qty_queries = {}
	for key, so_item_field in [
		("so_item_qty", "qty"),
		("so_item_delivered_qty", "delivered_qty"),
		("so_item_returned_qty", "returned_qty"),
	]:
		qty_queries.update(
			{
				key: (
					frappe.qb.from_(SalesOrderItem)
					.select(SalesOrderItem[so_item_field])
					.where(SalesOrderItem.name == dnpi.parent_detail_docname)
					.where(SalesOrderItem.delivered_by_supplier == 0)
				)
			}
		)

	dnpi_parent = frappe.qb.from_(dnpi).select(dnpi.qty.as_("dnpi_qty"))
	for key, query in qty_queries.items():
		dnpi_parent = dnpi_parent.select(query.as_(key))
	dnpi_parent = dnpi_parent.select(dnpi.parent, dnpi.name)

	dnpi_parent = dnpi_parent + tab

	q = (
		frappe.qb.from_(dnpi_parent)
		.select(
			Sum(
				dnpi_parent.dnpi_qty
				* (
					(
						dnpi_parent.so_item_qty
						- dnpi_parent.so_item_delivered_qty
						- dnpi_parent.so_item_returned_qty
					)
					/ dnpi_parent.so_item_qty
				)
			)
		)
		.where(dnpi_parent.so_item_qty >= dnpi_parent.so_item_delivered_qty)
	)

	reserved_qty = q.run()
	return flt(reserved_qty[0][0]) if reserved_qty else 0


def get_indented_qty(item_code, warehouse):
	# Ordered Qty is always maintained in stock UOM
	inward_qty = frappe.db.sql(
		"""
		select sum(mr_item.stock_qty - mr_item.ordered_qty)
		from `tabMaterial Request Item` mr_item, `tabMaterial Request` mr
		where mr_item.item_code=%s and mr_item.warehouse=%s
			and mr.material_request_type in ('Purchase', 'Manufacture', 'Customer Provided', 'Material Transfer')
			and mr_item.stock_qty > mr_item.ordered_qty and mr_item.parent=mr.name
			and mr.status!='Stopped' and mr.docstatus=1
	""",
		(item_code, warehouse),
	)
	inward_qty = flt(inward_qty[0][0]) if inward_qty else 0

	outward_qty = frappe.db.sql(
		"""
		select sum(mr_item.stock_qty - mr_item.ordered_qty)
		from `tabMaterial Request Item` mr_item, `tabMaterial Request` mr
		where mr_item.item_code=%s and mr_item.warehouse=%s
			and mr.material_request_type = 'Material Issue'
			and mr_item.stock_qty > mr_item.ordered_qty and mr_item.parent=mr.name
			and mr.status!='Stopped' and mr.docstatus=1
	""",
		(item_code, warehouse),
	)
	outward_qty = flt(outward_qty[0][0]) if outward_qty else 0

	requested_qty = inward_qty - outward_qty

	return requested_qty


def get_ordered_qty(item_code, warehouse):
	ordered_qty = frappe.db.sql(
		"""
		select sum((po_item.qty - po_item.received_qty)*po_item.conversion_factor)
		from `tabPurchase Order Item` po_item, `tabPurchase Order` po
		where po_item.item_code=%s and po_item.warehouse=%s
		and po_item.qty > po_item.received_qty and po_item.parent=po.name
		and po.status not in ('Closed', 'Delivered') and po.docstatus=1
		and po_item.delivered_by_supplier = 0""",
		(item_code, warehouse),
	)

	return flt(ordered_qty[0][0]) if ordered_qty else 0


def get_planned_qty(item_code, warehouse):
	planned_qty = frappe.db.sql(
		"""
		select sum(qty - produced_qty) from `tabWork Order`
		where production_item = %s and fg_warehouse = %s and status not in ("Stopped", "Completed", "Closed")
		and docstatus=1 and qty > produced_qty""",
		(item_code, warehouse),
	)

	return flt(planned_qty[0][0]) if planned_qty else 0


def update_bin_qty(item_code, warehouse, qty_dict=None):
	from erpnext.stock.utils import get_bin

	bin = get_bin(item_code, warehouse)
	mismatch = False
	for field, value in qty_dict.items():
		if flt(bin.get(field)) != flt(value):
			bin.set(field, flt(value))
			mismatch = True

	bin.modified = now()
	if mismatch:
		bin.set_projected_qty()
		bin.db_update()
		bin.clear_cache()


def set_stock_balance_as_per_serial_no(
	item_code=None, posting_date=None, posting_time=None, fiscal_year=None
):
	if not posting_date:
		posting_date = nowdate()
	if not posting_time:
		posting_time = nowtime()

	condition = " and item.name='%s'" % item_code.replace("'", "'") if item_code else ""

	bin = frappe.db.sql(
		"""select bin.item_code, bin.warehouse, bin.actual_qty, item.stock_uom
		from `tabBin` bin, tabItem item
		where bin.item_code = item.name and item.has_serial_no = 1 %s"""
		% condition
	)

	for d in bin:
		serial_nos = frappe.db.sql(
			"""select count(name) from `tabSerial No`
			where item_code=%s and warehouse=%s and docstatus < 2""",
			(d[0], d[1]),
		)

		sle = frappe.db.sql(
			"""select valuation_rate, company from `tabStock Ledger Entry`
			where item_code = %s and warehouse = %s and is_cancelled = 0
			order by posting_date desc limit 1""",
			(d[0], d[1]),
		)

		sle_dict = {
			"doctype": "Stock Ledger Entry",
			"item_code": d[0],
			"warehouse": d[1],
			"transaction_date": nowdate(),
			"posting_date": posting_date,
			"posting_time": posting_time,
			"voucher_type": "Stock Reconciliation (Manual)",
			"voucher_no": "",
			"voucher_detail_no": "",
			"actual_qty": flt(serial_nos[0][0]) - flt(d[2]),
			"stock_uom": d[3],
			"incoming_rate": sle and flt(serial_nos[0][0]) > flt(d[2]) and flt(sle[0][0]) or 0,
			"company": sle and cstr(sle[0][1]) or 0,
			"batch_no": "",
			"serial_no": "",
		}

		sle_doc = frappe.get_doc(sle_dict)
		sle_doc.flags.ignore_validate = True
		sle_doc.flags.ignore_links = True
		sle_doc.insert()

		args = sle_dict.copy()
		args.update({"sle_id": sle_doc.name})

		create_repost_item_valuation_entry(
			{
				"item_code": d[0],
				"warehouse": d[1],
				"posting_date": posting_date,
				"posting_time": posting_time,
			}
		)


def reset_serial_no_status_and_warehouse(serial_nos=None):
	if not serial_nos:
		serial_nos = frappe.db.sql_list("""select name from `tabSerial No` where docstatus = 0""")
		for serial_no in serial_nos:
			try:
				sr = frappe.get_doc("Serial No", serial_no)
				last_sle = sr.get_last_sle()
				if flt(last_sle.actual_qty) > 0:
					sr.warehouse = last_sle.warehouse

				sr.via_stock_ledger = True
				sr.save()
			except Exception:
				pass