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

import copy
import json

import frappe
from frappe import _, bold
from frappe.model.document import Document
from frappe.model.naming import NamingSeries
from frappe.query_builder import Interval
from frappe.query_builder.functions import Count, CurDate, UnixTimestamp
from frappe.utils import (
	cint,
	cstr,
	flt,
	formatdate,
	get_link_to_form,
	getdate,
	now_datetime,
	nowtime,
	strip,
	strip_html,
)
from frappe.utils.html_utils import clean_html
from pypika import Order

import erpnext
from erpnext.controllers.item_variant import (
	ItemVariantExistsError,
	copy_attributes_to_variant,
	get_variant,
	make_variant_item_code,
	validate_item_variant_attributes,
)
from erpnext.setup.utils import get_exchange_rate
from erpnext.stock.doctype.item_default.item_default import ItemDefault
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
from erpnext.stock.utils import get_valuation_method


class DuplicateReorderRows(frappe.ValidationError):
	pass


class StockExistsForTemplate(frappe.ValidationError):
	pass


class InvalidBarcode(frappe.ValidationError):
	pass


class DataValidationError(frappe.ValidationError):
	pass


class Item(Document):
	# begin: auto-generated types
	# This code is auto-generated. Do not modify anything in this block.

	from typing import TYPE_CHECKING

	if TYPE_CHECKING:
		from frappe.types import DF

		from erpnext.stock.doctype.item_barcode.item_barcode import ItemBarcode
		from erpnext.stock.doctype.item_customer_detail.item_customer_detail import ItemCustomerDetail
		from erpnext.stock.doctype.item_default.item_default import ItemDefault
		from erpnext.stock.doctype.item_reorder.item_reorder import ItemReorder
		from erpnext.stock.doctype.item_supplier.item_supplier import ItemSupplier
		from erpnext.stock.doctype.item_tax.item_tax import ItemTax
		from erpnext.stock.doctype.item_variant_attribute.item_variant_attribute import ItemVariantAttribute
		from erpnext.stock.doctype.uom_conversion_detail.uom_conversion_detail import UOMConversionDetail

		allow_alternative_item: DF.Check
		allow_negative_stock: DF.Check
		asset_category: DF.Link | None
		asset_naming_series: DF.Literal[None]
		attributes: DF.Table[ItemVariantAttribute]
		auto_create_assets: DF.Check
		barcodes: DF.Table[ItemBarcode]
		batch_number_series: DF.Data | None
		brand: DF.Link | None
		country_of_origin: DF.Link | None
		create_new_batch: DF.Check
		customer_code: DF.SmallText | None
		customer_items: DF.Table[ItemCustomerDetail]
		customs_tariff_number: DF.Link | None
		default_bom: DF.Link | None
		default_item_manufacturer: DF.Link | None
		default_manufacturer_part_no: DF.Data | None
		default_material_request_type: DF.Literal[
			"Purchase", "Material Transfer", "Material Issue", "Manufacture", "Customer Provided"
		]
		delivered_by_supplier: DF.Check
		description: DF.TextEditor | None
		disabled: DF.Check
		enable_deferred_expense: DF.Check
		enable_deferred_revenue: DF.Check
		end_of_life: DF.Date | None
		grant_commission: DF.Check
		has_batch_no: DF.Check
		has_expiry_date: DF.Check
		has_serial_no: DF.Check
		has_variants: DF.Check
		image: DF.AttachImage | None
		include_item_in_manufacturing: DF.Check
		inspection_required_before_delivery: DF.Check
		inspection_required_before_purchase: DF.Check
		is_customer_provided_item: DF.Check
		is_fixed_asset: DF.Check
		is_grouped_asset: DF.Check
		is_purchase_item: DF.Check
		is_sales_item: DF.Check
		is_stock_item: DF.Check
		is_sub_contracted_item: DF.Check
		item_code: DF.Data
		item_defaults: DF.Table[ItemDefault]
		item_group: DF.Link
		item_name: DF.Data | None
		last_purchase_rate: DF.Float
		lead_time_days: DF.Int
		max_discount: DF.Float
		min_order_qty: DF.Float
		naming_series: DF.Literal["STO-ITEM-.YYYY.-"]
		no_of_months: DF.Int
		no_of_months_exp: DF.Int
		opening_stock: DF.Float
		over_billing_allowance: DF.Float
		over_delivery_receipt_allowance: DF.Float
		purchase_tax_withholding_category: DF.Link | None
		purchase_uom: DF.Link | None
		quality_inspection_template: DF.Link | None
		reorder_levels: DF.Table[ItemReorder]
		retain_sample: DF.Check
		safety_stock: DF.Float
		sales_tax_withholding_category: DF.Link | None
		sales_uom: DF.Link | None
		sample_quantity: DF.Int
		serial_no_series: DF.Data | None
		shelf_life_in_days: DF.Int
		standard_rate: DF.Currency
		stock_uom: DF.Link
		supplier_items: DF.Table[ItemSupplier]
		taxes: DF.Table[ItemTax]
		total_projected_qty: DF.Float
		uoms: DF.Table[UOMConversionDetail]
		use_serial_no_wise_valuation: DF.Check
		valuation_method: DF.Literal["", "FIFO", "Moving Average", "LIFO"]
		valuation_rate: DF.Currency
		variant_based_on: DF.Literal["Item Attribute", "Manufacturer"]
		variant_of: DF.Link | None
		warranty_period: DF.Data | None
		weight_per_unit: DF.Float
		weight_uom: DF.Link | None
	# end: auto-generated types

	def onload(self):
		self.set_onload("stock_exists", self.stock_ledger_created())
		self.set_onload("asset_naming_series", get_asset_naming_series())
		self.set_onload(
			"current_valuation_method", get_valuation_method(self.name, erpnext.get_default_company())
		)
		self.set_onload("asset_exists", self.has_submitted_assets())

	def autoname(self):
		if frappe.db.get_default("item_naming_by") == "Naming Series":
			if self.variant_of:
				if not self.item_code:
					template_item_name = frappe.db.get_value("Item", self.variant_of, "item_name")
					make_variant_item_code(self.variant_of, template_item_name, self)
			else:
				from frappe.model.naming import set_name_by_naming_series

				set_name_by_naming_series(self)
				self.item_code = self.name

		self.item_code = strip(self.item_code)
		self.name = self.item_code

	def after_insert(self):
		"""set opening stock and item price"""
		if self.standard_rate and not self.has_variants:
			price_lists = {
				self.get_selling_price_list(default.default_price_list)
				for default in self.item_defaults or [frappe._dict()]
			}
			if [price_list for price_list in price_lists if self.add_price(price_list)]:
				frappe.msgprint(
					_("Item Price created at rate {0}").format(
						frappe.format(self.standard_rate, {"fieldtype": "Currency"})
					),
					indicator="green",
					alert=True,
				)

		if self.opening_stock:
			if self.opening_stock > 10000 and self.has_serial_no:
				frappe.enqueue(
					self.set_opening_stock,
					queue="long",
					timeout=600,
					job_name=f"set_opening_stock_for_{self.name}",
					enqueue_after_commit=True,
				)
				frappe.msgprint(
					_(
						"Opening stock creation has been queued and will be created in the background. Please check the Stock Reconciliation after some time."
					),
					indicator="orange",
					alert=True,
				)

			else:
				self.set_opening_stock()

	def validate(self):
		if not self.item_name:
			self.item_name = self.item_code

		self.validate_uom()
		self.validate_description()
		self.add_default_uom_in_conversion_factor_table()
		self.validate_conversion_factor()
		self.validate_item_type()
		self.validate_naming_series()
		if self.is_new():
			self.validate_opening_stock()
		self.validate_shelf_life()
		self.check_for_active_boms()
		self.fill_customer_code()
		self.check_item_tax()
		self.validate_barcode()
		self.validate_warehouse_for_reorder()
		self.update_bom_item_desc()

		self.validate_variant()
		self.validate_has_variants()
		self.validate_attributes_in_variants()
		self.validate_stock_exists_for_template_item()
		self.validate_attributes()
		self.validate_variant_attributes()
		self.validate_variant_based_on_change()
		self.validate_fixed_asset()
		self.clear_retain_sample()
		self.validate_retain_sample()
		self.validate_uom_conversion_factor()
		self.validate_customer_provided_part()
		self.update_defaults_from_item_group()
		self.validate_item_defaults()
		self.validate_auto_reorder_enabled_in_stock_settings()
		self.validate_serial_and_batch_no_enabled_in_stock_settings()
		self.cant_change()
		self.validate_serialized_change_with_bundle()
		self.validate_serial_no_wise_valuation()
		self.set_valuation_method_for_serial_no_wise_valuation()
		self.validate_item_tax_net_rate_range()

		if not self.is_new():
			self.old_item_group = frappe.db.get_value(self.doctype, self.name, "item_group")

	def on_update(self):
		from erpnext.stock.utils import clear_valuation_method_cache

		self.update_variants()
		self.update_item_price()
		clear_valuation_method_cache()

	def validate_description(self):
		"""Clean HTML description if set"""
		if (
			cint(frappe.get_single_value("Stock Settings", "clean_description_html"))
			and self.description != self.item_name  # perf: Avoid cleaning up a fallback
		):
			old_desc = self.description
			self.description = clean_html(self.description)

			if (
				old_desc
				and self.description
				and "<img src" in old_desc
				and "<img src" not in self.description
			):
				frappe.msgprint(
					_(
						'Image in the description has been removed. To disable this behavior, uncheck "{0}" in {1}.'
					).format(
						frappe.get_meta("Stock Settings").get_label("clean_description_html"),
						get_link_to_form("Stock Settings"),
					),
					alert=True,
				)

	def validate_customer_provided_part(self):
		if self.is_customer_provided_item:
			if self.is_purchase_item:
				frappe.throw(_('"Customer Provided Item" cannot be Purchase Item also'))
			if self.valuation_rate:
				frappe.throw(_('"Customer Provided Item" cannot have Valuation Rate'))
			self.default_material_request_type = "Customer Provided"

	def get_selling_price_list(self, price_list=None):
		if price_list and frappe.get_cached_value("Price List", price_list, "selling"):
			return price_list

		return frappe.get_single_value("Selling Settings", "selling_price_list") or frappe.db.get_value(
			"Price List", "Standard Selling"
		)

	def add_price(self, price_list=None):
		"""Add a new price"""
		price_list = self.get_selling_price_list(price_list)
		if not price_list:
			return False

		default_currency = erpnext.get_default_currency()
		currency = frappe.get_cached_value("Price List", price_list, "currency") or default_currency
		price_list_rate = flt(self.standard_rate)
		if default_currency and currency != default_currency:
			exchange_rate = flt(get_exchange_rate(currency, default_currency, args="for_selling"))
			if not exchange_rate:
				return False

			price_list_rate = price_list_rate / exchange_rate

		frappe.get_doc(
			{
				"doctype": "Item Price",
				"price_list": price_list,
				"item_code": self.name,
				"uom": self.stock_uom,
				"brand": self.brand,
				"currency": currency,
				"price_list_rate": price_list_rate,
			}
		).insert()
		return True

	def validate_opening_stock(self):
		if not self.is_stock_item or not self.opening_stock:
			return

		if self.has_serial_no and not self.serial_no_series:
			frappe.throw(
				_("Serial Number Series is mandatory when entering Opening Stock for a serialized item.")
			)

		if self.has_batch_no and not self.create_new_batch:
			frappe.throw(
				_("Enable Automatically Create New Batch when entering Opening Stock for a batched item.")
			)

	def set_opening_stock(self):
		"""set opening stock"""
		if not self.is_stock_item:
			return

		if self.valuation_rate is None and not self.is_customer_provided_item:
			frappe.throw(_("Valuation Rate is mandatory if Opening Stock entered"))

		for default in self.item_defaults or [
			frappe._dict({"company": frappe.defaults.get_defaults().company})
		]:
			default_warehouse = default.default_warehouse
			if not default_warehouse and default.company:
				default_warehouse = frappe.get_cached_value("Company", default.company, "default_warehouse")

			if not default_warehouse:
				default_warehouse = frappe.db.get_value(
					"Warehouse", {"warehouse_name": _("Stores"), "company": default.company}
				)

			if default_warehouse:
				opening_account = frappe.db.get_value(
					"Account",
					{"company": default.company, "account_type": "Temporary", "is_group": 0},
					"name",
				)

				if not opening_account:
					frappe.throw(
						_(
							"Please set a Temporary Opening account for company {0} to create an Opening Stock reconciliation."
						).format(frappe.bold(default.company))
					)
				stock_reco = create_opening_stock_reconciliation(
					item_code=self.name,
					company=default.company,
					qty=self.opening_stock,
					valuation_rate=self.valuation_rate,
					warehouse=default_warehouse,
					expense_account=opening_account,
				)
				stock_reco.add_comment("Comment", _("Opening Stock"))

				stock_reco_link = frappe.utils.get_link_to_form("Stock Reconciliation", stock_reco.name)
				if self.valuation_rate == 0:
					frappe.msgprint(
						_("Opening Stock reconciliation created with zero valuation rate: {0}").format(
							stock_reco_link
						),
						indicator="orange",
						alert=True,
					)
				else:
					frappe.msgprint(
						_("Opening Stock reconciliation created: {0}").format(stock_reco_link),
						indicator="green",
						alert=True,
					)

	def validate_fixed_asset(self):
		if self.is_fixed_asset:
			if self.is_stock_item:
				frappe.throw(_("Fixed Asset Item must be a non-stock item."))

			if not self.asset_category:
				frappe.throw(_("Asset Category is mandatory for Fixed Asset item"))

			if self.stock_ledger_created():
				frappe.throw(_("Cannot be a fixed asset item as Stock Ledger is created."))

		if not self.is_fixed_asset and not self.is_new():
			if self.has_submitted_assets():
				frappe.throw(
					_('"Is Fixed Asset" cannot be unchecked, as Asset record exists against the item')
				)

	def validate_retain_sample(self):
		if self.retain_sample and not frappe.db.exists(
			"Company", {"sample_retention_warehouse": ("is", "set")}
		):
			frappe.throw(_("Please select Sample Retention Warehouse in Company first"))
		if self.retain_sample and not self.has_batch_no:
			frappe.throw(
				_(
					"{0} Retain Sample is based on batch, please check Has Batch No to retain sample of item"
				).format(self.item_code)
			)

	def validate_shelf_life(self):
		if (
			self.has_batch_no
			and self.has_expiry_date
			and self.create_new_batch
			and cint(self.shelf_life_in_days) <= 0
		):
			frappe.throw(
				_("{0} must be greater than zero.").format(
					self.get_label_from_fieldname("shelf_life_in_days")
				)
			)

	def clear_retain_sample(self):
		if not self.has_batch_no:
			self.retain_sample = False

		if not self.retain_sample:
			self.sample_quantity = 0

	def add_default_uom_in_conversion_factor_table(self):
		if not self.is_new() and self.has_value_changed("stock_uom"):
			self.uoms = []
			frappe.msgprint(
				_("Successfully changed Stock UOM, please redefine conversion factors for new UOM."),
				alert=True,
			)

		uoms_list = [d.uom for d in self.get("uoms")]

		if self.stock_uom not in uoms_list:
			self.append("uoms", {"uom": self.stock_uom, "conversion_factor": 1})

	def validate_item_tax_net_rate_range(self):
		for tax in self.get("taxes"):
			if flt(tax.maximum_net_rate) < flt(tax.minimum_net_rate):
				frappe.throw(
					_("Taxes row #{0}: {1} cannot be smaller than {2}").format(
						tax.idx,
						bold(_(tax.meta.get_label("maximum_net_rate"))),
						bold(_(tax.meta.get_label("minimum_net_rate"))),
					)
				)

	def update_template_tables(self):
		template = frappe.get_cached_doc("Item", self.variant_of)

		# add item taxes from template
		for d in template.get("taxes"):
			self.append("taxes", d)

		# copy re-order table if empty
		if not self.get("reorder_levels"):
			for d in template.get("reorder_levels"):
				n = {}
				for k in (
					"warehouse",
					"warehouse_reorder_level",
					"warehouse_reorder_qty",
					"material_request_type",
				):
					n[k] = d.get(k)
				self.append("reorder_levels", n)

	def validate_conversion_factor(self):
		check_list = []
		for d in self.get("uoms"):
			if cstr(d.uom) in check_list:
				frappe.throw(
					_(
						"Unit of Measure {0} has been entered more than once in Conversion Factor Table"
					).format(d.uom)
				)
			else:
				check_list.append(cstr(d.uom))

			if d.uom and cstr(d.uom) == cstr(self.stock_uom) and flt(d.conversion_factor) != 1:
				frappe.throw(
					_("Conversion factor for default Unit of Measure must be 1 in row {0}").format(d.idx)
				)

	def validate_item_type(self):
		if self.has_serial_no == 1 and self.is_stock_item == 0 and not self.is_fixed_asset:
			frappe.throw(_("'Has Serial No' can not be 'Yes' for non-stock item"))

		if self.has_serial_no == 0 and self.serial_no_series:
			self.serial_no_series = None

	def validate_naming_series(self):
		for field in ["serial_no_series", "batch_number_series"]:
			series = self.get(field)
			if series and "#" in series and "." not in series:
				frappe.throw(
					_("Invalid naming series (. missing) for {0}").format(
						frappe.bold(self.meta.get_field(field).label)
					)
				)

			if self.is_new() and series:
				obj = NamingSeries(series)
				prefix = obj.get_prefix()
				doctype = frappe.qb.DocType("Series")

				query = frappe.qb.from_(doctype).select(doctype.name).where(doctype.name.like(f"{prefix}%"))

				prefix_exists = query.run(as_dict=True)
				if prefix_exists:
					frappe.msgprint(
						_(
							"The {0} prefix '{1}' already exists. Please change the Serial No Series, otherwise you will get a Duplicate Entry error."
						).format(bold(frappe.unscrub(field)), bold(prefix)),
						title=_("Serial No Series Overlap"),
						indicator="yellow",
						alert=True,
					)

	def check_for_active_boms(self):
		if self.default_bom:
			bom_item = frappe.db.get_value("BOM", self.default_bom, "item")
			if bom_item not in (self.name, self.variant_of):
				frappe.throw(
					_("Default BOM ({0}) must be active for this item or its template").format(bom_item)
				)

	def fill_customer_code(self):
		"""
		Append all the customer codes and insert into "customer_code" field of item table.
		Used to search Item by customer code.
		"""
		customer_codes = set(d.ref_code for d in self.get("customer_items", []))
		self.customer_code = ",".join(customer_codes)

	def check_item_tax(self):
		"""Check whether Tax Rate is not entered twice for same Tax Type"""
		check_list = []
		for d in self.get("taxes"):
			if d.item_tax_template:
				if (d.item_tax_template, d.tax_category) in check_list:
					frappe.throw(
						_("{0} entered twice {1} in Item Taxes").format(
							frappe.bold(d.item_tax_template),
							f"for tax category {frappe.bold(d.tax_category)}" if d.tax_category else "",
						)
					)
				else:
					check_list.append((d.item_tax_template, d.tax_category))

	def validate_barcode(self):
		import barcodenumber

		if len(self.barcodes) > 0:
			for item_barcode in self.barcodes:
				options = frappe.get_meta("Item Barcode").get_options("barcode_type").split("\n")
				if item_barcode.barcode:
					duplicate = frappe.db.sql(
						"""select parent from `tabItem Barcode` where barcode = %s and parent != %s""",
						(item_barcode.barcode, self.name),
					)
					if duplicate:
						frappe.throw(
							_("Barcode {0} already used in Item {1}").format(
								item_barcode.barcode, duplicate[0][0]
							)
						)

					item_barcode.barcode_type = (
						"" if item_barcode.barcode_type not in options else item_barcode.barcode_type
					)
					if item_barcode.barcode_type:
						barcode_type = convert_erpnext_to_barcodenumber(
							item_barcode.barcode_type.replace("-", "").upper(), item_barcode.barcode
						)
						if barcode_type in barcodenumber.barcodes():
							if not barcodenumber.check_code(barcode_type, item_barcode.barcode):
								frappe.throw(
									_("Barcode {0} is not a valid {1} code").format(
										item_barcode.barcode, item_barcode.barcode_type
									),
									InvalidBarcode,
								)

	def validate_warehouse_for_reorder(self):
		"""Validate Reorder level table for duplicate and conditional mandatory"""
		warehouse_material_request_type: list[tuple[str, str]] = []

		_warehouse_before_save = frappe._dict()
		if not self.is_new() and self._doc_before_save:
			_warehouse_before_save = {
				d.name: (d.warehouse, d.warehouse_group)
				for d in self._doc_before_save.get("reorder_levels") or []
			}

		for d in self.get("reorder_levels"):
			if not d.warehouse_group:
				d.warehouse_group = d.warehouse
			if (d.get("warehouse"), d.get("material_request_type")) not in warehouse_material_request_type:
				warehouse_material_request_type += [(d.get("warehouse"), d.get("material_request_type"))]
			else:
				frappe.throw(
					_(
						"Row #{0}: A reorder entry already exists for warehouse {1} with reorder type {2}."
					).format(d.idx, d.warehouse, d.material_request_type),
					DuplicateReorderRows,
				)

			if d.warehouse_reorder_level and not d.warehouse_reorder_qty:
				frappe.throw(_("Row #{0}: Please set reorder quantity").format(d.idx))

			if d.warehouse_group and d.warehouse:
				if _warehouse_before_save.get(d.name) == (d.warehouse, d.warehouse_group):
					continue

				child_warehouses = get_child_warehouses(d.warehouse_group)
				if d.warehouse not in child_warehouses:
					frappe.throw(
						_(
							"Row #{0}: The warehouse {1} is not a child warehouse of a group warehouse {2}"
						).format(d.idx, bold(d.warehouse), bold(d.warehouse_group)),
						title=_("Incorrect Check in (group) Warehouse for Reorder"),
					)

	def stock_ledger_created(self):
		if not hasattr(self, "_stock_ledger_created"):
			self._stock_ledger_created = len(
				frappe.db.sql(
					"""select name from `tabStock Ledger Entry`
				where item_code = %s and is_cancelled = 0 limit 1""",
					self.name,
				)
			)
		return self._stock_ledger_created

	def has_submitted_assets(self):
		return bool(frappe.db.exists("Asset", {"item_code": self.name, "docstatus": 1}))

	def update_item_price(self):
		if self.is_new():
			return

		frappe.db.sql(
			"""
				UPDATE `tabItem Price`
				SET
					item_name=%(item_name)s,
					item_description=%(item_description)s,
					brand=%(brand)s
				WHERE item_code=%(item_code)s
			""",
			dict(
				item_name=self.item_name,
				item_description=self.description,
				brand=self.brand,
				item_code=self.name,
			),
		)

	def on_trash(self):
		frappe.db.sql("""delete from tabBin where item_code=%s""", self.name)
		frappe.db.sql("delete from `tabItem Price` where item_code=%s", self.name)
		for variant_of in frappe.get_all("Item", filters={"variant_of": self.name}):
			frappe.delete_doc("Item", variant_of.name)

	def before_rename(self, old_name, new_name, merge=False):
		if self.item_name == old_name:
			frappe.db.set_value("Item", old_name, "item_name", new_name)

		if merge:
			self.validate_properties_before_merge(new_name)
			self.validate_duplicate_product_bundles_before_merge(old_name, new_name)
			self.validate_duplicate_lead_times_before_merge(old_name, new_name)
			self.delete_old_bins(old_name, new_name)

	def after_rename(self, old_name, new_name, merge):
		if merge:
			self.validate_duplicate_item_in_stock_reconciliation(old_name, new_name)
			frappe.msgprint(
				_("It can take upto few hours for accurate stock values to be visible after merging items."),
				indicator="orange",
				title=_("Note"),
			)

		frappe.db.set_value("Item", new_name, "item_code", new_name)

		if merge:
			self.set_last_purchase_rate(new_name)
			self.recalculate_bin_qty(new_name)

	def delete_old_bins(self, old_name, new_name=None):
		if not new_name:
			frappe.db.delete("Bin", {"item_code": old_name})
			return

		target_warehouses = frappe.get_all("Bin", filters={"item_code": new_name}, pluck="warehouse")
		if target_warehouses:
			frappe.db.delete("Bin", {"item_code": old_name, "warehouse": ("in", target_warehouses)})

	def validate_duplicate_item_in_stock_reconciliation(self, old_name, new_name):
		records = frappe.db.sql(
			""" SELECT parent, COUNT(*) as records
			FROM `tabStock Reconciliation Item`
			WHERE item_code = %s and docstatus = 1
			GROUP By item_code, warehouse, parent
			HAVING records > 1
		""",
			new_name,
			as_dict=1,
		)

		if not records:
			return
		document = _("Stock Reconciliation") if len(records) == 1 else _("Stock Reconciliations")

		msg = _("The items {0} and {1} are present in the following {2} :").format(
			frappe.bold(old_name), frappe.bold(new_name), document
		)

		msg += " <br>"
		msg += ", ".join([get_link_to_form("Stock Reconciliation", d.parent) for d in records]) + "<br><br>"

		msg += _(
			"Note: To merge the items, create a separate Stock Reconciliation for the old item {0}"
		).format(frappe.bold(old_name))

		frappe.throw(_(msg), title=_("Cannot Merge"), exc=DataValidationError)

	def validate_properties_before_merge(self, new_name):
		# Validate properties before merging
		if not frappe.db.exists("Item", new_name):
			frappe.throw(_("Item {0} does not exist").format(new_name))

		field_list = [
			"stock_uom",
			"is_stock_item",
			"has_serial_no",
			"has_batch_no",
			"has_variants",
			"variant_of",
		]
		new_properties = [cstr(d) for d in frappe.db.get_value("Item", new_name, field_list)]

		if self.has_variants or cint(frappe.db.get_value("Item", new_name, "has_variants")):
			frappe.throw(_("Templates cannot be merged"), title=_("Cannot Merge"), exc=DataValidationError)

		if new_properties != [cstr(self.get(field)) for field in field_list]:
			msg = _("To merge, following properties must be same for both items")
			msg += ": \n" + ", ".join([_(self.meta.get_label(fld)) for fld in field_list])
			frappe.throw(msg, title=_("Cannot Merge"), exc=DataValidationError)

	def validate_duplicate_product_bundles_before_merge(self, old_name, new_name):
		"Block merge if both old and new items have product bundles."
		old_bundle = frappe.get_value("Product Bundle", filters={"new_item_code": old_name, "disabled": 0})
		new_bundle = frappe.get_value("Product Bundle", filters={"new_item_code": new_name, "disabled": 0})

		if old_bundle and new_bundle:
			bundle_link = get_link_to_form("Product Bundle", old_bundle)
			old_name, new_name = frappe.bold(old_name), frappe.bold(new_name)

			msg = _("Please delete Product Bundle {0}, before merging {1} into {2}").format(
				bundle_link, old_name, new_name
			)
			frappe.throw(msg, title=_("Cannot Merge"), exc=DataValidationError)

	def validate_duplicate_lead_times_before_merge(self, old_name, new_name):
		old_lead_time = frappe.db.get_value("Item Lead Time", {"item_code": old_name})
		if not (old_lead_time and frappe.db.exists("Item Lead Time", {"item_code": new_name})):
			return

		frappe.throw(
			_("Please delete Item Lead Time {0}, before merging {1} into {2}").format(
				get_link_to_form("Item Lead Time", old_lead_time),
				frappe.bold(old_name),
				frappe.bold(new_name),
			),
			title=_("Cannot Merge"),
			exc=DataValidationError,
		)

	def set_last_purchase_rate(self, new_name):
		last_purchase_rate = get_last_purchase_details(new_name).get("base_net_rate", 0)
		frappe.db.set_value("Item", new_name, "last_purchase_rate", last_purchase_rate, update_modified=False)

	def recalculate_bin_qty(self, new_name):
		from erpnext.stock.stock_balance import repost_stock

		existing_allow_negative_stock = frappe.get_single_value("Stock Settings", "allow_negative_stock")
		frappe.db.set_single_value("Stock Settings", "allow_negative_stock", 1)

		repost_stock_for_warehouses = frappe.get_all(
			"Stock Ledger Entry",
			"warehouse",
			filters={"item_code": new_name},
			pluck="warehouse",
			distinct=True,
		)

		bin_warehouses = set(frappe.get_all("Bin", filters={"item_code": new_name}, pluck="warehouse"))

		# Delete all existing bins to avoid duplicate bins for the same item and warehouse
		frappe.db.delete("Bin", {"item_code": new_name})

		for warehouse in repost_stock_for_warehouses:
			repost_stock(new_name, warehouse)

		for warehouse in bin_warehouses - set(repost_stock_for_warehouses):
			repost_stock(new_name, warehouse, only_bin=True)

		for warehouse in bin_warehouses | set(repost_stock_for_warehouses):
			rebuild_bin_reservations(new_name, warehouse)

		frappe.db.set_single_value("Stock Settings", "allow_negative_stock", existing_allow_negative_stock)

	def update_bom_item_desc(self):
		if self.is_new():
			return

		if self.db_get("description") != self.description:
			frappe.db.sql(
				"""
				update `tabBOM`
				set description = %s
				where item = %s and docstatus < 2
			""",
				(self.description, self.name),
			)

			frappe.db.sql(
				"""
				update `tabBOM Item`
				set description = %s
				where item_code = %s and docstatus < 2
			""",
				(self.description, self.name),
			)

			frappe.db.sql(
				"""
				update `tabBOM Explosion Item`
				set description = %s
				where item_code = %s and docstatus < 2
			""",
				(self.description, self.name),
			)

	def validate_item_defaults(self):
		companies = {row.company for row in self.item_defaults}

		if len(companies) != len(self.item_defaults):
			frappe.throw(_("Cannot set multiple Item Defaults for a company."))

		validate_item_default_company_links(self.item_defaults)

	def update_defaults_from_item_group(self):
		"""Get defaults from Item Group"""
		if self.item_defaults or not self.item_group:
			return

		item_group = frappe.get_cached_doc("Item Group", self.item_group)
		if item_group.item_group_defaults:
			for item in item_group.item_group_defaults:
				self.append(
					"item_defaults",
					{
						"company": item.company,
						"default_warehouse": item.default_warehouse,
						"default_price_list": item.default_price_list,
						"buying_cost_center": item.buying_cost_center,
						"default_supplier": item.default_supplier,
						"expense_account": item.expense_account,
						"selling_cost_center": item.selling_cost_center,
						"income_account": item.income_account,
					},
				)
		else:
			defaults = frappe.defaults.get_defaults() or {}

			# To check default warehouse is belong to the default company
			if (
				defaults.get("default_warehouse")
				and defaults.company
				and defaults.company
				== frappe.get_cached_value("Warehouse", defaults.default_warehouse, "company")
			):
				self.append(
					"item_defaults",
					{"company": defaults.get("company"), "default_warehouse": defaults.default_warehouse},
				)

		if not self.taxes and item_group.taxes:
			for tax in item_group.taxes:
				self.append(
					"taxes",
					{
						"item_tax_template": tax.item_tax_template,
						"tax_category": tax.tax_category,
						"valid_from": tax.valid_from,
						"minimum_net_rate": tax.minimum_net_rate,
						"maximum_net_rate": tax.maximum_net_rate,
					},
				)

	def update_variants(self):
		if self.flags.dont_update_variants or frappe.db.get_single_value(
			"Item Variant Settings", "do_not_update_variants"
		):
			return
		if self.has_variants:
			variants = frappe.db.get_all("Item", fields=["item_code"], filters={"variant_of": self.name})
			if variants:
				if len(variants) <= 30:
					update_variants(variants, self, publish_progress=False)
					frappe.msgprint(_("Item Variants updated"))
				else:
					frappe.enqueue(
						"erpnext.stock.doctype.item.item.update_variants",
						variants=variants,
						template=self,
						now=frappe.in_test,
						timeout=600,
						enqueue_after_commit=True,
					)

	def validate_variant(self):
		if self.variant_of:
			has_variants, based_on = frappe.get_value(
				"Item", self.variant_of, ["has_variants", "variant_based_on"]
			)
			if not has_variants:
				frappe.throw(_("Item {0} is not a template item.").format(frappe.bold(self.variant_of)))

			if based_on == "Item Attribute":
				previous_doc = self.get_doc_before_save()
				saved_attributes = (
					{(row.attribute, row.attribute_value) for row in previous_doc.attributes}
					if previous_doc
					else set()
				)

				for d in self.attributes:
					if (d.attribute, d.attribute_value) in saved_attributes:
						continue

					if not frappe.db.exists(
						"Item Variant Attribute", {"attribute": d.attribute, "parent": self.variant_of}
					):
						frappe.throw(
							_("Attribute {0} is not valid for the selected template.").format(
								frappe.bold(d.attribute)
							)
						)

					numeric_values, disabled = frappe.get_value(
						"Item Variant Attribute",
						{"attribute": d.attribute, "parent": self.variant_of},
						["numeric_values", "disabled"],
					)

					if disabled:
						frappe.throw(_("Attribute {0} is disabled.").format(frappe.bold(d.attribute)))

					if (
						not numeric_values
						and d.attribute_value
						and not frappe.db.exists(
							"Item Attribute Value",
							{"parent": d.attribute, "attribute_value": d.attribute_value},
						)
					):
						frappe.throw(
							_("Attribute Value {0} is not valid for the selected attribute {1}.").format(
								frappe.bold(d.attribute_value), frappe.bold(d.attribute)
							)
						)

	def validate_has_variants(self):
		if self.is_new():
			return

		if not self.has_variants and self.has_value_changed("has_variants"):
			if frappe.db.exists("Item", {"variant_of": self.name}):
				frappe.throw(_("Item has variants."))

	def validate_attributes_in_variants(self):
		if not self.has_variants or self.is_new():
			return

		old_doc = self.get_doc_before_save()
		old_doc_attributes = set([attr.attribute for attr in old_doc.attributes])
		own_attributes = [attr.attribute for attr in self.attributes]

		# Check if old attributes were removed from the list
		# Is old_attrs is a subset of new ones
		# that means we need not check any changes
		if old_doc_attributes.issubset(set(own_attributes)):
			return

		from collections import defaultdict

		# get all item variants
		items = [item["name"] for item in frappe.get_all("Item", {"variant_of": self.name})]

		# get all deleted attributes
		deleted_attribute = list(old_doc_attributes.difference(set(own_attributes)))

		# fetch all attributes of these items
		item_attributes = frappe.get_all(
			"Item Variant Attribute",
			filters={"parent": ["in", items], "attribute": ["in", deleted_attribute]},
			fields=["attribute", "parent"],
		)
		not_included = defaultdict(list)

		for attr in item_attributes:
			if attr["attribute"] not in own_attributes:
				not_included[attr["parent"]].append(attr["attribute"])

		if not len(not_included):
			return

		def body(docnames):
			docnames.sort()
			return "<br>".join(docnames)

		def table_row(title, body):
			return f"""<tr>
				<td>{title}</td>
				<td>{body}</td>
			</tr>"""

		rows = ""
		for docname, attr_list in not_included.items():
			link = f"<a href='/app/item/{docname}'>{frappe.bold(docname)}</a>"
			rows += table_row(link, body(attr_list))

		error_description = _(
			"The following deleted attributes exist in Variants but not in the Template. You can either delete the Variants or keep the attribute(s) in template."
		)

		message = """
			<div>{}</div><br>
			<table class="table">
				<thead>
					<td>{}</td>
					<td>{}</td>
				</thead>
				{}
			</table>
		""".format(error_description, _("Variant Items"), _("Attributes"), rows)

		frappe.throw(message, title=_("Variant Attribute Error"), is_minimizable=True, wide=True)

	def validate_stock_exists_for_template_item(self):
		if self.stock_ledger_created() and self._doc_before_save:
			if (
				cint(self._doc_before_save.has_variants) != cint(self.has_variants)
				or self._doc_before_save.variant_of != self.variant_of
			):
				frappe.throw(
					_(
						"Cannot change Variant properties after stock transaction. You will have to make a new Item to do this."
					).format(self.name),
					StockExistsForTemplate,
				)

			if self.has_variants or self.variant_of:
				if not self.is_child_table_same("attributes"):
					frappe.throw(
						_(
							"Cannot change Attributes after stock transaction. Make a new Item and transfer stock to the new Item"
						)
					)

	def validate_variant_based_on_change(self):
		if not self.is_new() and (
			self.variant_of or (self.has_variants and frappe.get_all("Item", {"variant_of": self.name}))
		):
			if self.variant_based_on != frappe.db.get_value("Item", self.name, "variant_based_on"):
				frappe.throw(_("Variant Based On cannot be changed"))

	def validate_uom(self):
		allow_different_uom = frappe.get_cached_value(
			"Item Variant Settings", "Item Variant Settings", "allow_different_uom"
		)

		if not self.is_new():
			check_stock_uom_with_bin(self.name, self.stock_uom)
		if self.has_variants and not allow_different_uom:
			for d in frappe.db.get_all("Item", filters={"variant_of": self.name}):
				check_stock_uom_with_bin(d.name, self.stock_uom)
		if self.variant_of:
			if not allow_different_uom:
				template_uom = frappe.db.get_value("Item", self.variant_of, "stock_uom")
				if template_uom != self.stock_uom:
					frappe.throw(
						_(
							"Default Unit of Measure for Variant '{0}' must be same as in Template '{1}'"
						).format(self.stock_uom, template_uom)
					)

	def validate_uom_conversion_factor(self):
		for d in self.get("uoms"):
			if not d.conversion_factor:
				d.conversion_factor = get_uom_conv_factor(d.uom, self.stock_uom)

			if flt(d.conversion_factor) <= 0:
				frappe.throw(_("Row {0}: Conversion Factor must be greater than zero.").format(d.idx))

	def validate_attributes(self):
		if not (self.has_variants or self.variant_of):
			return

		if not self.variant_based_on:
			self.variant_based_on = "Item Attribute"

		if self.variant_based_on == "Item Attribute":
			attributes = []
			if not self.attributes:
				frappe.throw(_("Attribute table is mandatory"))
			for d in self.attributes:
				if d.attribute in attributes:
					frappe.throw(
						_("Attribute {0} selected multiple times in Attributes Table").format(d.attribute)
					)
				else:
					attributes.append(d.attribute)

	def validate_variant_attributes(self):
		if (
			self.variant_of
			and self.variant_based_on == "Item Attribute"
			and (self.is_new() or self.has_attribute_changes())
		):
			# remove attributes with no attribute_value set
			self.attributes = [d for d in self.attributes if cstr(d.attribute_value).strip()]

			args = {}
			for i, d in enumerate(self.attributes):
				d.idx = i + 1
				args[d.attribute] = d.attribute_value

			variant = get_variant(self.variant_of, args, self.name)
			if variant:
				frappe.throw(
					_("Item variant {0} exists with same attributes").format(variant), ItemVariantExistsError
				)

			validate_item_variant_attributes(self, args)

			# copy variant_of value for each attribute row
			for d in self.attributes:
				d.variant_of = self.variant_of

	def has_attribute_changes(self) -> bool:
		doc_before_save = self.get_doc_before_save()
		if not doc_before_save:
			return True

		def get_values(doc):
			return [(d.attribute, cstr(d.attribute_value)) for d in doc.get("attributes")]

		return get_values(self) != get_values(doc_before_save)

	def cant_change(self):
		if self.is_new():
			return

		restricted_fields = (
			"has_serial_no",
			"is_stock_item",
			"valuation_method",
			"has_batch_no",
		)

		values = frappe.db.get_value("Item", self.name, restricted_fields, as_dict=True)
		if not values:
			return

		if not values.get("valuation_method") and self.get("valuation_method"):
			values["valuation_method"] = (
				frappe.get_single_value("Stock Settings", "valuation_method") or "FIFO"
			)

		changed_fields = [
			field for field in restricted_fields if cstr(self.get(field)) != cstr(values.get(field))
		]

		# Allow to change valuation method from FIFO to Moving Average not vice versa
		if self.valuation_method == "Moving Average" and "valuation_method" in changed_fields:
			changed_fields.remove("valuation_method")

		if not changed_fields:
			return

		if linked_doc := self._get_linked_submitted_documents(changed_fields):
			changed_field_labels = [frappe.bold(_(self.meta.get_label(f))) for f in changed_fields]
			msg = _(
				"As there are existing submitted transactions against item {0}, you can not change the value of {1}."
			).format(self.name, ", ".join(changed_field_labels))

			if linked_doc and isinstance(linked_doc, dict):
				msg += "<br>"
				msg += _("Example of a linked document: {0}").format(
					frappe.get_desk_link(linked_doc.doctype, linked_doc.docname)
				)

			frappe.throw(msg, title=_("Linked with submitted documents"))

	def validate_serial_no_wise_valuation(self):
		if self.is_new() or not self._doc_before_save:
			return

		if not self.use_serial_no_wise_valuation or self._doc_before_save.use_serial_no_wise_valuation:
			return

		if frappe.db.exists("Serial No", {"item_code": self.name}):
			frappe.throw(
				_(
					"Serial No Wise Valuation cannot be enabled for Item {0} because Serial Nos already exist for it. Valuation for those Serial Nos was not tracked, so enabling it now would value outward entries incorrectly."
				).format(frappe.bold(self.name)),
				title=_("Serial Nos Exist"),
			)

	def set_valuation_method_for_serial_no_wise_valuation(self):
		if not self.has_serial_no or self.use_serial_no_wise_valuation:
			return

		# Only the switch turning off forces Moving Average, because the per serial costs already in the
		# ledger cannot be replayed as a FIFO queue. An item that has always had the switch off keeps its
		# own method, so an unrelated save cannot silently revalue a ledger nothing reposts.
		if self._doc_before_save and not self._doc_before_save.use_serial_no_wise_valuation:
			return

		if not frappe.db.exists("Stock Ledger Entry", {"item_code": self.name, "is_cancelled": 0}):
			return

		if (
			not self.is_new()
			and self._doc_before_save
			and self.has_value_changed("valuation_method")
			and self.valuation_method in ("FIFO", "LIFO")
		):
			frappe.throw(
				_(
					"Valuation Method for Item {0} must be Moving Average because Serial No Wise Valuation is disabled. Enable Serial No Wise Valuation to use FIFO or LIFO."
				).format(frappe.bold(self.name)),
				title=_("Invalid Valuation Method"),
			)

		self.valuation_method = "Moving Average"

	def validate_serialized_change_with_bundle(self):
		"""Block turning a serialized item non-serialized while any Serial and Batch Bundle still exists
		for it. Such bundles carry the item's serial numbers; the user must delete or cancel them first."""
		if self.is_new() or self.has_serial_no or not self._doc_before_save:
			return

		# Only relevant when the item was serialized before and is now being unset.
		if not self._doc_before_save.has_serial_no:
			return

		# Draft (docstatus 0) or submitted (docstatus 1) bundles block the change; cancelled ones don't.
		if frappe.db.count("Serial and Batch Bundle", {"item_code": self.name, "docstatus": ("<", 2)}):
			frappe.throw(
				_(
					"Cannot change Item {0} from serialized to non-serialized because a Serial and Batch Bundle exists for it. Please delete or cancel the Serial and Batch Bundle first."
				).format(frappe.bold(self.name)),
				title=_("Serial and Batch Bundle Exists"),
			)

	def _get_linked_submitted_documents(self, changed_fields: list[str]) -> dict[str, str] | None:
		linked_doctypes = [
			"Delivery Note Item",
			"Sales Invoice Item",
			"POS Invoice Item",
			"Purchase Receipt Item",
			"Purchase Invoice Item",
			"Stock Entry Detail",
			"Stock Reconciliation Item",
		]

		# For "Is Stock Item", following doctypes is important
		# because reserved_qty, ordered_qty and requested_qty updated from these doctypes
		if "is_stock_item" in changed_fields:
			linked_doctypes += [
				"Sales Order Item",
				"Purchase Order Item",
				"Material Request Item",
				"Work Order Item",
				"Product Bundle",
				"BOM",
			]

		for doctype in linked_doctypes:
			filters = {"item_code": self.name, "docstatus": 1}

			if doctype in ("Product Bundle", "BOM"):
				if doctype == "Product Bundle":
					filters = {"new_item_code": self.name}
					fieldname = "new_item_code as docname"
				else:
					filters = {"item": self.name, "docstatus": 1}
					fieldname = "name as docname"

				if linked_doc := frappe.db.get_value(doctype, filters, fieldname, as_dict=True):
					return linked_doc.update({"doctype": doctype})

			elif (
				doctype
				in (
					"Purchase Invoice Item",
					"Sales Invoice Item",
				)
				and "is_stock_item" not in changed_fields
			):
				# If Invoice has Stock impact, only then consider it.
				if linked_doc := frappe.db.get_value(
					"Stock Ledger Entry",
					{"item_code": self.name, "is_cancelled": 0},
					["voucher_no as docname", "voucher_type as doctype"],
					as_dict=True,
				):
					return linked_doc

			elif linked_doc := frappe.db.get_value(
				doctype,
				filters,
				["parent as docname", "parenttype as doctype"],
				as_dict=True,
			):
				return linked_doc

	def validate_auto_reorder_enabled_in_stock_settings(self):
		if self.reorder_levels:
			enabled = frappe.get_single_value("Stock Settings", "auto_indent")
			if not enabled:
				frappe.msgprint(
					msg=_("You have to enable auto re-order in Stock Settings to maintain re-order levels."),
					title=_("Enable Auto Re-Order"),
					indicator="orange",
				)

	def validate_serial_and_batch_no_enabled_in_stock_settings(self):
		if frappe.get_single_value("Stock Settings", "enable_serial_and_batch_no_for_item"):
			return

		doc_before_save = self.get_doc_before_save()
		for fieldname in ("has_serial_no", "has_batch_no"):
			if cint(self.get(fieldname)) and not (doc_before_save and cint(doc_before_save.get(fieldname))):
				frappe.throw(
					_("Cannot enable {0} as {1} is disabled in {2}").format(
						bold(self.meta.get_label(fieldname)),
						bold(_("Activate Serial / Batch No for Item")),
						get_link_to_form("Stock Settings", "Stock Settings"),
					),
					title=_("Serial / Batch No Not Activated"),
				)


def convert_erpnext_to_barcodenumber(erpnext_number, barcode):
	if erpnext_number == "EAN":
		ean_type = {
			8: "EAN8",
			13: "EAN13",
		}
		barcode_length = len(barcode)
		if barcode_length in ean_type:
			return ean_type[barcode_length]

		return erpnext_number

	convert = {"UPCA": "UPC"}

	if erpnext_number in convert:
		return convert[erpnext_number]

	return erpnext_number


def make_item_price(item, price_list_name, item_price):
	frappe.get_doc(
		{
			"doctype": "Item Price",
			"price_list": price_list_name,
			"item_code": item,
			"price_list_rate": item_price,
		}
	).insert()


def get_timeline_data(doctype: str, name: str) -> dict[int, int]:
	"""get timeline data based on Stock Ledger Entry. This is displayed as heatmap on the item page."""

	sle = frappe.qb.DocType("Stock Ledger Entry")

	return dict(
		frappe.qb.from_(sle)
		.select(UnixTimestamp(sle.posting_date), Count("*"))
		.where(sle.item_code == name)
		.where(sle.posting_date > CurDate() - Interval(years=1))
		.groupby(sle.posting_date)
		.run()
	)


def validate_end_of_life(item_code, end_of_life=None, disabled=None):
	if (not end_of_life) or (disabled is None):
		end_of_life, disabled = frappe.db.get_value("Item", item_code, ["end_of_life", "disabled"])

	if end_of_life and end_of_life != "0000-00-00" and getdate(end_of_life) <= now_datetime().date():
		frappe.throw(
			_("Item {0} has reached its end of life on {1}").format(item_code, formatdate(end_of_life))
		)

	if disabled:
		frappe.throw(_("Item {0} is disabled").format(item_code))


def validate_is_stock_item(item_code, is_stock_item=None):
	if not is_stock_item:
		is_stock_item = frappe.db.get_value("Item", item_code, "is_stock_item")

	if is_stock_item != 1:
		frappe.throw(_("Item {0} is not a stock Item").format(item_code))


def validate_cancelled_item(item_code, docstatus=None):
	if docstatus is None:
		docstatus = frappe.db.get_value("Item", item_code, "docstatus")

	if docstatus == 2:
		frappe.throw(_("Item {0} is cancelled").format(item_code))


@frappe.request_cache
def get_last_purchase_details(item_code, doc_name=None, conversion_rate=1.0):
	"""returns last purchase details in stock uom"""
	# get last purchase order item details

	last_purchase_order = get_purchase_voucher_details("Purchase Order", item_code, doc_name)

	# get last purchase receipt item details
	last_purchase_receipt = get_purchase_voucher_details("Purchase Receipt", item_code, doc_name)

	purchase_order_date = getdate(
		last_purchase_order and last_purchase_order[0].transaction_date or "1900-01-01"
	)
	purchase_receipt_date = getdate(
		last_purchase_receipt and last_purchase_receipt[0].posting_date or "1900-01-01"
	)

	if last_purchase_order and (purchase_order_date >= purchase_receipt_date or not last_purchase_receipt):
		# use purchase order

		last_purchase = last_purchase_order[0]
		purchase_date = purchase_order_date

	elif last_purchase_receipt and (purchase_receipt_date > purchase_order_date or not last_purchase_order):
		# use purchase receipt
		last_purchase = last_purchase_receipt[0]
		purchase_date = purchase_receipt_date

	else:
		last_purchase_invoice = get_purchase_voucher_details("Purchase Invoice", item_code, doc_name)

		if last_purchase_invoice:
			last_purchase = last_purchase_invoice[0]
			purchase_date = getdate(last_purchase.posting_date)
		else:
			return frappe._dict()

	conversion_factor = flt(last_purchase.conversion_factor)
	out = frappe._dict(
		{
			"base_price_list_rate": flt(last_purchase.base_price_list_rate) / conversion_factor,
			"base_rate": flt(last_purchase.base_rate) / conversion_factor,
			"base_net_rate": flt(last_purchase.base_net_rate) / conversion_factor,
			"discount_percentage": flt(last_purchase.discount_percentage),
			"purchase_date": purchase_date,
		}
	)

	conversion_rate = flt(conversion_rate) or 1.0
	out.update(
		{
			"price_list_rate": out.base_price_list_rate / conversion_rate,
			"rate": out.base_rate / conversion_rate,
			"base_rate": out.base_rate,
			"base_net_rate": out.base_net_rate,
		}
	)

	return out


def get_purchase_voucher_details(doctype, item_code, document_name=None):
	parent_doc = frappe.qb.DocType(doctype)
	child_doc = frappe.qb.DocType(doctype + " Item")

	query = (
		frappe.qb.from_(parent_doc)
		.inner_join(child_doc)
		.on(parent_doc.name == child_doc.parent)
		.select(
			parent_doc.name,
			parent_doc.conversion_rate,
			child_doc.conversion_factor,
			child_doc.base_price_list_rate,
			child_doc.discount_percentage,
			child_doc.base_rate,
			child_doc.base_net_rate,
		)
		.where(parent_doc.docstatus == 1)
		.where(child_doc.item_code == item_code)
	)

	if document_name:
		query = query.where(parent_doc.name != document_name)

	if doctype in ("Purchase Receipt", "Purchase Invoice"):
		query = query.select(parent_doc.posting_date, parent_doc.posting_time)
		query = query.orderby(
			parent_doc.posting_date, parent_doc.posting_time, parent_doc.name, order=Order.desc
		)
	else:
		query = query.select(parent_doc.transaction_date)
		query = query.orderby(parent_doc.transaction_date, parent_doc.name, order=Order.desc)

	return query.run(as_dict=1)


def rebuild_bin_reservations(item_code, warehouse):
	from erpnext.stock.utils import get_or_make_bin

	bin_doc = frappe.get_doc("Bin", get_or_make_bin(item_code, warehouse))
	bin_doc.update_reserved_qty_for_sub_contracting()
	bin_doc.update_reserved_stock()
	bin_doc.update_reserved_qty_for_production()


def check_stock_uom_with_bin(item, stock_uom):
	if stock_uom == frappe.db.get_value("Item", item, "stock_uom"):
		return

	ref_uom = frappe.db.get_value("Stock Ledger Entry", {"item_code": item}, "stock_uom")

	if ref_uom:
		if cstr(ref_uom) != cstr(stock_uom):
			frappe.throw(
				_(
					"Default Unit of Measure for Item {0} cannot be changed directly because you have already made some transaction(s) with another UOM. You will need to create a new Item to use a different Default UOM."
				).format(item)
			)

	bin_list = frappe.get_all(
		"Bin",
		filters={"item_code": item, "stock_uom": ["!=", stock_uom]},
		or_filters=[
			["reserved_qty", ">", 0],
			["ordered_qty", ">", 0],
			["indented_qty", ">", 0],
			["planned_qty", ">", 0],
			["reserved_qty_for_production", ">", 0],
			["reserved_qty_for_sub_contract", ">", 0],
			["reserved_qty_for_production_plan", ">", 0],
			["reserved_stock", ">", 0],
		],
		pluck="name",  # only used as an existence check below
		limit=1,
	)

	if bin_list:
		frappe.throw(
			_(
				"Default Unit of Measure for Item {0} cannot be changed directly because you have already made some transaction(s) with another UOM. You need to either cancel the linked documents or create a new Item."
			).format(item)
		)

	# No SLE or documents against item. Bin UOM can be changed safely.
	frappe.db.sql("""update `tabBin` set stock_uom=%s where item_code=%s""", (stock_uom, item))


def get_item_defaults(item_code, company):
	item = frappe.get_cached_doc("Item", item_code)

	out = item.as_dict()

	for d in item.item_defaults:
		if d.company == company:
			row = d.as_dict(no_private_properties=True)
			row.pop("name")
			out.update(row)
	return out


def set_item_default(item_code, company, fieldname, value):
	item = frappe.get_cached_doc("Item", item_code)

	for d in item.item_defaults:
		if d.company == company:
			if not d.get(fieldname):
				frappe.db.set_value(d.doctype, d.name, fieldname, value)
				item.clear_cache()
			return

	# no row found, add a new row for the company
	d = item.append("item_defaults", {fieldname: value, "company": company})
	d.db_insert()
	item.clear_cache()


@frappe.whitelist()
def get_item_details(item_code, company=None):
	# The whitelisted entry point authorises; _get_item_details does not. Deliberately not an
	# `ignore_permissions` argument: this is whitelisted, so a caller could pass it and skip the check.
	return _get_item_details(item_code, company, ignore_permissions=False)


def _get_item_details(item_code, company=None, ignore_permissions=True):
	doc = frappe.get_cached_doc("Item", item_code)
	if not ignore_permissions:
		# the whole Item document is returned below, so the record itself has to be authorised.
		doc.check_permission()

	out = frappe._dict()
	if company:
		out = get_item_defaults(item_code, company) or frappe._dict()

	out.update(doc.as_dict())

	return out


@frappe.whitelist()
def get_uom_conv_factor(uom: str | None, stock_uom: str | None):
	"""Get UOM conversion factor from uom to stock_uom
	e.g. uom = "Kg", stock_uom = "Gram" then returns 1000.0
	"""
	if uom == stock_uom:
		return 1.0

	from_uom, to_uom = uom, stock_uom  # renaming for readability

	exact_match = frappe.db.get_value(
		"UOM Conversion Factor", {"to_uom": to_uom, "from_uom": from_uom}, ["value"], as_dict=1
	)
	if exact_match:
		return exact_match.value

	inverse_match = frappe.db.get_value(
		"UOM Conversion Factor", {"to_uom": from_uom, "from_uom": to_uom}, ["value"], as_dict=1
	)
	if inverse_match:
		return flt(1 / inverse_match.value, frappe.get_precision("UOM Conversion Factor", "value"))

	# This attempts to try and get conversion from intermediate UOM.
	# case:
	# 			 g -> mg = 1000
	# 			 g -> kg = 0.001
	# therefore	 kg -> mg = 1000  / 0.001 = 1,000,000
	first = frappe.qb.DocType("UOM Conversion Factor").as_("first")
	second = frappe.qb.DocType("UOM Conversion Factor").as_("second")
	intermediate_match = (
		frappe.qb.from_(first)
		.join(second)
		.on((first.from_uom == second.from_uom) & (first.category == second.category))
		.select((first.value / second.value).as_("value"))
		.where((first.to_uom == to_uom) & (second.to_uom == from_uom))
		.limit(1)
		.run(as_dict=1)
	)

	if intermediate_match:
		return flt(intermediate_match[0].value, frappe.get_precision("UOM Conversion Factor", "value"))


@frappe.whitelist()
def get_item_attribute(parent, attribute_value=""):
	"""Used for providing auto-completions in child table."""
	if not frappe.has_permission("Item"):
		frappe.throw(_("No Permission"))

	return frappe.get_all(
		"Item Attribute Value",
		fields=["attribute_value"],
		filters={"parent": parent, "attribute_value": ("like", f"%{attribute_value}%")},
	)


def update_variants(variants, template, publish_progress=True):
	total = len(variants)
	for count, d in enumerate(variants, start=1):
		variant = frappe.get_doc("Item", d)
		copy_attributes_to_variant(template, variant)
		variant.save()
		if publish_progress:
			frappe.publish_progress(count / total * 100, title=_("Updating Variants..."))


def validate_item_default_company_links(item_defaults: list[ItemDefault]) -> None:
	company_links = [
		(df.options, df.fieldname)
		for df in frappe.get_meta("Item Default").get_link_fields()
		if df.options in ("Warehouse", "Cost Center", "Account")
	]
	for item_default in item_defaults:
		for doctype, field in company_links:
			if item_default.get(field):
				company = frappe.db.get_value(doctype, item_default.get(field), "company", cache=True)
				if company and company != item_default.company:
					frappe.throw(
						_("Row #{}: {} {} doesn't belong to Company {}. Please select valid {}.").format(
							item_default.idx,
							doctype,
							frappe.bold(item_default.get(field)),
							frappe.bold(item_default.company),
							frappe.bold(frappe.unscrub(field)),
						),
						title=_("Invalid Item Defaults"),
					)

		validate_default_inventory_account(item_default)


def validate_default_inventory_account(item_default) -> None:
	if not item_default.default_inventory_account:
		return

	account_type, is_group = frappe.db.get_value(
		"Account", item_default.default_inventory_account, ["account_type", "is_group"], cache=True
	)
	if account_type != "Stock" or is_group:
		frappe.throw(
			_("Row #{0}: Default Inventory Account {1} must be a Stock account that is not a group.").format(
				item_default.idx, frappe.bold(item_default.default_inventory_account)
			),
			title=_("Invalid Item Defaults"),
		)


@frappe.whitelist()
def get_asset_naming_series():
	from erpnext.assets.doctype.asset.asset import get_asset_naming_series

	return get_asset_naming_series()


@frappe.request_cache
def get_child_warehouses(warehouse):
	from erpnext.stock.doctype.warehouse.warehouse import get_child_warehouses

	return get_child_warehouses(warehouse)


ITEM_PRICES_LIMIT = 10


@frappe.whitelist()
def get_item_prices(item_code: str):
	"""Fetch valid item prices for the item prices tab."""
	frappe.has_permission("Item Price", "read", throw=True)
	today = getdate()

	# get_list, not get_all: otherwise a caller restricted to one Price List sees every party's negotiated rate
	prices = frappe.get_list(
		"Item Price",
		filters={"item_code": item_code, "docstatus": ["!=", 2]},
		or_filters=[["valid_upto", "is", "not set"], ["valid_upto", ">=", today]],
		fields=[
			"name",
			"price_list",
			"price_list_rate",
			"currency",
			"uom",
			"customer",
			"supplier",
			"buying",
			"selling",
			"valid_upto",
		],
		order_by="price_list",
		limit=ITEM_PRICES_LIMIT + 1,
	)

	return {
		"prices": prices[:ITEM_PRICES_LIMIT],
		"has_more": len(prices) > ITEM_PRICES_LIMIT,
	}


@frappe.whitelist()
def make_opening_stock_entry(
	item_code: str,
	company: str,
	qty: float,
	valuation_rate: float,
	warehouse: str | None = None,
):
	if not frappe.has_permission("Item", "write", item_code):
		frappe.throw(_("Not permitted"), frappe.PermissionError)

	item = frappe.get_doc("Item", item_code)

	if not item.is_stock_item:
		frappe.throw(_("Opening Stock can only be set for stock items."))
	if item.has_serial_no or item.has_batch_no:
		frappe.throw(
			_("Opening Stock for serialised or batch items must be set via the Stock Reconciliation form.")
		)
	if item.stock_ledger_created():
		frappe.throw(
			_("Opening Stock cannot be created as stock transactions already exist for item {0}.").format(
				frappe.bold(item_code)
			)
		)

	if flt(qty) <= 0:
		frappe.throw(_("Quantity must be greater than zero."))

	if flt(valuation_rate) < 0:
		frappe.throw(_("Valuation Rate cannot be negative."))

	if warehouse:
		warehouse_company = frappe.db.get_value("Warehouse", warehouse, "company")
		if warehouse_company != company:
			frappe.throw(_("Warehouse {0} does not belong to Company {1}.").format(warehouse, company))

	target_warehouse = get_default_warehouse_for_opening_stock(item, company, warehouse)

	opening_account = frappe.db.get_value(
		"Account",
		{"company": company, "account_type": "Temporary", "is_group": 0},
		"name",
	)

	if not opening_account:
		frappe.throw(
			_(
				"Please set a Temporary Opening account for company {0} to create an Opening Stock reconciliation."
			).format(frappe.bold(company))
		)

	stock_reco = create_opening_stock_reconciliation(
		item_code=item_code,
		company=company,
		qty=qty,
		valuation_rate=valuation_rate,
		warehouse=target_warehouse,
		expense_account=opening_account,
	)
	stock_reco.add_comment("Comment", _("Opening Stock"))

	frappe.msgprint(
		_("Opening Stock reconciliation created: {0}").format(
			get_link_to_form("Stock Reconciliation", stock_reco.name)
		),
		indicator="green",
		alert=True,
	)

	return stock_reco.name


def create_opening_stock_reconciliation(
	item_code: str,
	company: str,
	qty: float,
	valuation_rate: float,
	warehouse: str,
	expense_account: str,
):
	stock_reco = frappe.get_doc(
		{
			"doctype": "Stock Reconciliation",
			"purpose": "Opening Stock",
			"company": company,
			"expense_account": expense_account,
			"items": [
				{
					"item_code": item_code,
					"warehouse": warehouse,
					"qty": flt(qty),
					"valuation_rate": flt(valuation_rate),
					"allow_zero_valuation_rate": 1 if flt(valuation_rate) == 0 else 0,
					"reconcile_all_serial_batch": 1,
				}
			],
		}
	)

	stock_reco.insert()
	set_opening_stock_serial_batch_bundle(stock_reco)
	stock_reco.submit()

	return stock_reco


def set_opening_stock_serial_batch_bundle(stock_reco):
	row = stock_reco.items[0]
	item_details = frappe.get_cached_value(
		"Item", row.item_code, ["has_serial_no", "has_batch_no"], as_dict=1
	)

	if not (item_details.has_serial_no or item_details.has_batch_no):
		return

	bundle = SerialBatchCreation(
		{
			"item_code": row.item_code,
			"warehouse": row.warehouse,
			"voucher_type": stock_reco.doctype,
			"voucher_no": stock_reco.name,
			"voucher_detail_no": row.name,
			"posting_date": stock_reco.posting_date,
			"posting_time": stock_reco.posting_time,
			"qty": row.qty,
			"avg_rate": row.valuation_rate,
			"type_of_transaction": "Inward",
			"company": stock_reco.company,
			"do_not_submit": True,
		}
	).make_serial_and_batch_bundle()

	if not bundle:
		return

	row.db_set("serial_and_batch_bundle", bundle.name, update_modified=False)
	row.serial_and_batch_bundle = bundle.name


def get_default_warehouse_for_opening_stock(item, company: str, warehouse: str | None):
	if warehouse:
		return warehouse

	for default in item.item_defaults:
		if default.company == company and default.default_warehouse:
			return default.default_warehouse

	if company_warehouse := frappe.get_cached_value("Company", company, "default_warehouse"):
		return company_warehouse

	stores_warehouse = frappe.db.get_value("Warehouse", {"warehouse_name": _("Stores"), "company": company})

	if stores_warehouse:
		return stores_warehouse

	frappe.throw(
		_(
			"No warehouse found for company {0}. Please set a Default Warehouse in Item Defaults or Company."
		).format(frappe.bold(company))
	)