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


import frappe
from frappe import _, bold, throw
from frappe.query_builder.functions import Min
from frappe.utils import cint, flt, get_link_to_form, nowtime

from erpnext.accounts.party import render_address
from erpnext.controllers.accounts_controller import _get_taxes_and_charges
from erpnext.controllers.sales_and_purchase_return import get_rate_for_return, is_batch_expired
from erpnext.controllers.stock_controller import StockController
from erpnext.stock.doctype.item.item import set_item_default
from erpnext.stock.get_item_details import get_bin_details, get_conversion_factor
from erpnext.stock.utils import _get_incoming_rate, get_combine_datetime, get_valuation_method


class SellingController(StockController):
	def __setup__(self):
		self.flags.ignore_permlevel_for_fields = ["selling_price_list", "price_list_currency"]

	def onload(self):
		super().onload()
		if (
			self.doctype in ("Sales Order", "Delivery Note", "Sales Invoice", "Quotation")
			and self.docstatus.is_draft()
			and not hasattr(self, "_action")
		):
			for item in self.get("items") + (self.get("packed_items") or []):
				company = self.company

				item.update(
					get_bin_details(
						item.item_code, item.warehouse, company=company, include_child_warehouses=True
					)
				)

		if self.docstatus == 1 and self.doctype in ["Delivery Note", "Sales Invoice"]:
			self.set_onload(
				"allow_to_make_qc_after_submission",
				frappe.get_single_value(
					"Stock Settings", "allow_to_make_quality_inspection_after_purchase_or_delivery"
				),
			)

		if self.get("company") and not self.get("terms"):
			if not self.get("tc_name"):
				self.tc_name = frappe.get_value("Company", self.company, "default_selling_terms")
			self.set_missing_terms()

	def validate(self):
		super().validate()
		self.validate_items()
		if not (self.get("is_debit_note") or self.get("is_return")):
			self.validate_max_discount()
		self.validate_selling_price()
		self.set_qty_as_per_stock_uom()
		self.set_po_nos(for_validate=True)
		self.set_gross_profit()
		set_default_income_account_for_item(self)
		self.set_customer_address()
		self.validate_price_list_country()
		self.validate_for_duplicate_items()
		self.validate_target_warehouse()
		if self.doctype == "Sales Invoice":
			self.validate_against_stock_reservation_entries()
		self.validate_auto_repeat_subscription_dates()
		self.validate_sample_retention_warehouse()
		for table_field in ["items", "packed_items"]:
			if self.get(table_field):
				self.set_serial_and_batch_bundle(table_field)

	def validate_against_stock_reservation_entries(self):
		if self.get("is_return") or (self.doctype == "Sales Invoice" and not cint(self.get("update_stock"))):
			return

		so_field = "against_sales_order" if self.doctype == "Delivery Note" else "sales_order"
		rows = [item for item in self.get("items") if item.get(so_field) and item.so_detail]
		reserved_warehouses = get_reserved_warehouses_by_order_row([item.so_detail for item in rows])
		for item in rows:
			warehouses = reserved_warehouses.get(item.so_detail)
			if not warehouses:
				continue

			if not item.warehouse:
				item.warehouse = warehouses[0]
			elif item.warehouse not in warehouses:
				frappe.throw(
					_("Row #{0}: Stock is reserved for item {1} in warehouse {2}.").format(
						item.idx,
						frappe.bold(item.item_code),
						frappe.bold(warehouses[0])
						if len(warehouses) == 1
						else _("{0} and {1}").format(
							frappe.bold(", ".join(warehouses[:-1])), frappe.bold(warehouses[-1])
						),
					),
					title=_("Stock Reservation Warehouse Mismatch"),
				)

	def validate_price_list_country(self):
		from erpnext.stock.doctype.price_list.price_list import validate_price_list_country

		if self.get("is_return") or not self.get("customer_address"):
			return

		country = frappe.db.get_value("Address", self.customer_address, "country")
		validate_price_list_country(self.get("selling_price_list"), country)

	def validate_standalone_serial_nos_customer(self):
		if not self.is_return or self.return_against:
			return

		if self.doctype in ["Sales Invoice", "Delivery Note"]:
			bundle_ids = [d.serial_and_batch_bundle for d in self.get("items") if d.serial_and_batch_bundle]
			if not bundle_ids:
				return

			serial_nos = frappe.get_all(
				"Serial and Batch Entry",
				filters={"parent": ("in", bundle_ids), "serial_no": ("is", "set")},
				pluck="serial_no",
			)

			if not serial_nos:
				return

			if serial_nos := frappe.get_all(
				"Serial No",
				filters={"name": ("in", serial_nos), "customer": ("is", "set")},
				fields=["name", "customer"],
			):
				for sn in serial_nos:
					if sn.customer and sn.customer != self.customer:
						frappe.throw(
							_(
								"Serial No {0} is already assigned to customer {1}. Can only be returned against the customer {1}"
							).format(frappe.bold(sn.name), frappe.bold(sn.customer)),
							title=_("Serial No Already Assigned"),
						)

	def set_missing_values(self, for_validate=False):
		super().set_missing_values(for_validate)

		# set contact and address details for customer, if they are not mentioned
		self.set_missing_lead_customer_details(for_validate=for_validate)
		self.set_price_list_and_item_details(for_validate=for_validate)
		self.set_company_contact_person()

	def set_missing_lead_customer_details(self, for_validate=False):
		customer, lead = None, None
		if getattr(self, "customer", None):
			customer = self.customer
		elif self.doctype == "Opportunity" and self.party_name:
			if self.opportunity_from == "Customer":
				customer = self.party_name
			else:
				lead = self.party_name
		elif self.doctype == "Quotation" and self.party_name:
			if self.quotation_to == "Customer":
				customer = self.party_name
			elif self.quotation_to == "Lead":
				lead = self.party_name

		if customer:
			from erpnext.accounts.party import _get_party_details

			party_details = _get_party_details(
				customer,
				ignore_permissions=self.flags.ignore_permissions,
				doctype=self.doctype,
				company=self.company,
				posting_date=self.get("posting_date"),
				fetch_payment_terms_template=self.has_value_changed("company"),
				party_address=self.customer_address,
				shipping_address=self.shipping_address_name,
				company_address=self.get("company_address"),
			)
			if not self.meta.get_field("sales_team"):
				party_details.pop("sales_team")
			self.update_if_missing(party_details)

		elif lead:
			from erpnext.crm.doctype.lead.lead import get_lead_details

			self.update_if_missing(
				get_lead_details(
					lead,
					posting_date=self.get("transaction_date") or self.get("posting_date"),
					company=self.company,
					doctype=self.doctype,
				)
			)

		if self.get("taxes_and_charges") and not self.get("taxes") and not for_validate:
			taxes = _get_taxes_and_charges("Sales Taxes and Charges Template", self.taxes_and_charges)
			for tax in taxes:
				self.append("taxes", tax)

	def set_price_list_and_item_details(self, for_validate=False):
		self.set_price_list_currency("Selling")
		self.set_missing_item_details(for_validate=for_validate)

	def set_company_contact_person(self):
		"""Set the Company's Default Sales Contact as Company Contact Person."""
		if self.company and self.meta.has_field("company_contact_person") and not self.company_contact_person:
			self.company_contact_person = frappe.get_cached_value(
				"Company", self.company, "default_sales_contact"
			)

	def remove_shipping_charge(self):
		if self.shipping_rule:
			shipping_rule = frappe.get_last_doc("Shipping Rule", self.shipping_rule)
			existing_shipping_charge = self.get(
				"taxes",
				{
					"doctype": "Sales Taxes and Charges",
					"charge_type": "Actual",
					"account_head": shipping_rule.account,
					"cost_center": shipping_rule.cost_center,
				},
			)
			if existing_shipping_charge:
				self.get("taxes").remove(existing_shipping_charge[-1])
				self.calculate_taxes_and_totals()

	def set_total_in_words(self):
		from frappe.utils import money_in_words

		if self.meta.get_field("base_in_words"):
			base_amount = abs(
				self.base_grand_total if self.is_rounded_total_disabled() else self.base_rounded_total
			)
			self.base_in_words = money_in_words(base_amount, self.company_currency)

		if self.meta.get_field("in_words"):
			amount = abs(self.grand_total if self.is_rounded_total_disabled() else self.rounded_total)
			self.in_words = money_in_words(amount, self.currency)

	def calculate_commission(self):
		if not self.meta.get_field("commission_rate"):
			return

		self.round_floats_in(self, ("amount_eligible_for_commission", "commission_rate"))

		if not (0 <= self.commission_rate <= 100.0):
			throw(
				"{} {}".format(
					_(self.meta.get_label("commission_rate")),
					_("must be between 0 and 100"),
				)
			)

		self.amount_eligible_for_commission = sum(
			item.base_net_amount for item in self.items if item.grant_commission
		)

		self.total_commission = flt(
			self.amount_eligible_for_commission * self.commission_rate / 100.0,
			self.precision("total_commission"),
		)

	def calculate_contribution(self):
		if not self.meta.get_field("sales_team"):
			return

		total = 0.0
		sales_team = self.get("sales_team")

		self.validate_sales_team(sales_team)

		for sales_person in sales_team:
			self.round_floats_in(sales_person)

			sales_person.allocated_amount = flt(
				flt(self.amount_eligible_for_commission) * sales_person.allocated_percentage / 100.0,
				self.precision("allocated_amount", sales_person),
			)

			if sales_person.commission_rate:
				sales_person.incentives = flt(
					sales_person.allocated_amount * flt(sales_person.commission_rate) / 100.0,
					self.precision("incentives", sales_person),
				)

			total += sales_person.allocated_percentage

		if sales_team and flt(total, self.precision("allocated_percentage", "sales_team")) != 100.0:
			throw(_("Total allocated percentage for sales team should be 100"))

	def validate_sales_team(self, sales_team):
		sales_persons = [d.sales_person for d in sales_team]

		if not sales_persons:
			return

		sales_person_status = frappe.db.get_all(
			"Sales Person", filters={"name": ["in", sales_persons]}, fields=["name", "enabled"]
		)

		for row in sales_person_status:
			if not row.enabled:
				frappe.throw(_("Sales Person <b>{0}</b> is disabled.").format(row.name))

	def validate_max_discount(self):
		for d in self.get("items"):
			if d.item_code:
				discount = flt(frappe.get_cached_value("Item", d.item_code, "max_discount"))

				if discount and flt(d.discount_percentage) > discount:
					frappe.throw(_("Maximum discount for Item {0} is {1}%").format(d.item_code, discount))

	def set_qty_as_per_stock_uom(self):
		allow_to_edit_stock_qty = frappe.get_single_value(
			"Stock Settings", "allow_to_edit_stock_uom_qty_for_sales"
		)

		for d in self.get("items"):
			if d.meta.get_field("stock_qty"):
				if not d.conversion_factor:
					frappe.throw(_("Row {0}: Conversion Factor is mandatory").format(d.idx))
				d.stock_qty = flt(d.qty) * flt(d.conversion_factor)
				if allow_to_edit_stock_qty:
					d.stock_qty = flt(d.stock_qty, d.precision("stock_qty"))

	def validate_selling_price(self):
		def throw_message(idx, item_name, rate, ref_rate_field):
			throw(
				_(
					"""Row #{0}: Selling rate for item {1} is lower than its {2}.
					Selling {3} should be atleast {4}.<br><br>Alternatively,
					you can disable '{5}' in {6} to bypass
					this validation."""
				).format(
					idx,
					bold(item_name),
					bold(ref_rate_field),
					bold("net rate"),
					bold(rate),
					bold(frappe.get_meta("Selling Settings").get_label("validate_selling_price")),
					get_link_to_form("Selling Settings"),
				),
				title=_("Invalid Selling Price"),
			)

		if self.get("is_return") or not frappe.get_single_value("Selling Settings", "validate_selling_price"):
			return

		is_internal_customer = self.get("is_internal_customer")

		for item in self.items:
			if not item.item_code or item.is_free_item:
				continue

			last_purchase_rate, is_stock_item = frappe.get_cached_value(
				"Item", item.item_code, ("last_purchase_rate", "is_stock_item")
			)

			last_purchase_rate_in_sales_uom = flt(
				last_purchase_rate * (item.conversion_factor or 1), item.precision("base_net_rate")
			)

			if flt(item.base_net_rate) < flt(last_purchase_rate_in_sales_uom):
				throw_message(item.idx, item.item_name, last_purchase_rate_in_sales_uom, "last purchase rate")

			if is_internal_customer or not is_stock_item:
				continue

			rate_field = "valuation_rate" if self.doctype in ["Sales Order", "Quotation"] else "incoming_rate"
			if item.get(rate_field) and item.base_net_rate < (
				valuation_rate := flt(
					item.get(rate_field) * (item.conversion_factor or 1), item.precision("base_net_rate")
				)
			):
				throw_message(
					item.idx,
					item.item_name,
					valuation_rate,
					"valuation rate",
				)

	def get_item_list(self):
		il = []
		for d in self.get("items"):
			if self.has_product_bundle(d.item_code):
				for p in self.get("packed_items"):
					if p.parent_detail_docname == d.name and p.parent_item == d.item_code:
						# the packing details table's qty is already multiplied with parent's qty
						il.append(
							frappe._dict(
								{
									"warehouse": p.warehouse or d.warehouse,
									"item_code": p.item_code,
									"qty": flt(p.qty),
									"serial_no": p.serial_no if self.docstatus == 2 else None,
									"batch_no": p.batch_no if self.docstatus == 2 else None,
									"uom": p.uom,
									"serial_and_batch_bundle": p.serial_and_batch_bundle
									or get_serial_and_batch_bundle(p, self, d),
									"name": d.name,
									"project": d.get("project"),
									"target_warehouse": p.target_warehouse,
									"company": self.company,
									"voucher_type": self.doctype,
									"allow_zero_valuation": d.allow_zero_valuation_rate,
									"sales_invoice_item": d.get("sales_invoice_item"),
									"dn_detail": d.get("dn_detail"),
									"incoming_rate": p.get("incoming_rate"),
									"item_row": p,
								}
							)
						)
			else:
				il.append(
					frappe._dict(
						{
							"warehouse": d.warehouse,
							"item_code": d.item_code,
							"qty": d.stock_qty,
							"serial_no": d.serial_no if self.docstatus == 2 else None,
							"batch_no": d.batch_no if self.docstatus == 2 else None,
							"uom": d.uom,
							"stock_uom": d.stock_uom,
							"conversion_factor": d.conversion_factor,
							"serial_and_batch_bundle": d.serial_and_batch_bundle,
							"name": d.name,
							"project": d.get("project"),
							"target_warehouse": d.target_warehouse,
							"company": self.company,
							"voucher_type": self.doctype,
							"allow_zero_valuation": d.allow_zero_valuation_rate,
							"sales_invoice_item": d.get("sales_invoice_item"),
							"dn_detail": d.get("dn_detail"),
							"incoming_rate": d.get("incoming_rate"),
							"item_row": d,
						}
					)
				)

		return il

	def has_product_bundle(self, item_code):
		product_bundle_items = getattr(self, "_product_bundle_items", None)
		if product_bundle_items is None:
			self._product_bundle_items = product_bundle_items = {}

		if item_code not in product_bundle_items:
			self._fetch_product_bundle_items(item_code)

		return product_bundle_items[item_code]

	def _fetch_product_bundle_items(self, item_code):
		product_bundle_items = self._product_bundle_items
		items_to_fetch = {row.item_code for row in self.items if row.item_code not in product_bundle_items}
		# fetch for requisite item_code even if it is not in items
		items_to_fetch.add(item_code)

		items_with_product_bundle = {
			row.new_item_code
			for row in frappe.get_all(
				"Product Bundle",
				filters={"new_item_code": ("in", items_to_fetch), "disabled": 0},
				fields="new_item_code",
			)
		}

		for item_code in items_to_fetch:
			product_bundle_items[item_code] = item_code in items_with_product_bundle

	def get_already_delivered_qty(self, current_docname, so, so_detail):
		delivered_via_dn = frappe.db.sql(
			"""select sum(qty) from `tabDelivery Note Item`
			where so_detail = %s and docstatus = 1
			and against_sales_order = %s
			and parent != %s""",
			(so_detail, so, current_docname),
		)

		delivered_via_si = frappe.db.sql(
			"""select sum(si_item.qty)
			from `tabSales Invoice Item` si_item, `tabSales Invoice` si
			where si_item.parent = si.name and si.update_stock = 1
			and si_item.so_detail = %s and si.docstatus = 1
			and si_item.sales_order = %s
			and si.name != %s""",
			(so_detail, so, current_docname),
		)

		total_delivered_qty = (flt(delivered_via_dn[0][0]) if delivered_via_dn else 0) + (
			flt(delivered_via_si[0][0]) if delivered_via_si else 0
		)

		return total_delivered_qty

	def get_so_qty_and_warehouse(self, so_detail):
		so_item = frappe.db.sql(
			"""select qty, warehouse from `tabSales Order Item`
			where name = %s and docstatus = 1""",
			so_detail,
			as_dict=1,
		)
		so_qty = so_item and flt(so_item[0]["qty"]) or 0.0
		so_warehouse = so_item and so_item[0]["warehouse"] or ""
		return so_qty, so_warehouse

	def check_sales_order_on_hold_or_close(self, ref_fieldname):
		if self.is_return:
			return
		self.check_for_on_hold_or_closed_status("Sales Order", ref_fieldname)

	def update_reserved_qty(self):
		so_map = {}
		for d in self.get("items"):
			if d.so_detail:
				if self.doctype == "Delivery Note" and d.against_sales_order:
					so_map.setdefault(d.against_sales_order, []).append(d.so_detail)
				elif self.doctype == "Sales Invoice" and d.sales_order and self.update_stock:
					so_map.setdefault(d.sales_order, []).append(d.so_detail)

		for so, so_item_rows in so_map.items():
			if so and so_item_rows:
				sales_order = frappe.get_lazy_doc("Sales Order", so)

				if (sales_order.status == "Closed" and not self.is_return) or sales_order.status in [
					"Cancelled"
				]:
					frappe.throw(
						_("{0} {1} is cancelled or closed").format(_("Sales Order"), so),
						frappe.InvalidStatusError,
					)

				sales_order.update_reserved_qty(so_item_rows)

	def set_incoming_rate(self):
		def reset_incoming_rate():
			old_item = next(
				(
					item
					for item in (old_doc.get("items") + (old_doc.get("packed_items") or []))
					if item.name == d.name
				),
				None,
			)
			if old_item:
				old_qty = flt(old_item.get("stock_qty") or old_item.get("actual_qty") or old_item.get("qty"))
				if (
					old_item.item_code != d.item_code
					or old_item.warehouse != d.warehouse
					or old_qty != qty
					or old_item.serial_no != d.serial_no
					or get_serial_nos(old_item.serial_and_batch_bundle)
					!= get_serial_nos(d.serial_and_batch_bundle)
					or old_item.batch_no != d.batch_no
					or get_batch_nos(old_item.serial_and_batch_bundle)
					!= get_batch_nos(d.serial_and_batch_bundle)
				):
					d.incoming_rate = 0

		if self.doctype not in ("Delivery Note", "Sales Invoice"):
			return

		if self.doctype == "Sales Invoice" and not self.update_stock and not self.is_internal_transfer():
			return

		from erpnext.stock.serial_batch_bundle import get_batch_nos, get_serial_nos

		allow_at_arms_length_price = frappe.get_cached_value(
			"Stock Settings", None, "allow_internal_transfer_at_arms_length_price"
		)
		set_zero_rate_for_expired_batch = frappe.db.get_single_value(
			"Selling Settings", "set_zero_rate_for_expired_batch"
		)

		is_standalone = self.is_return and not self.return_against

		old_doc = self.get_doc_before_save()
		items = self.get("items") + (self.get("packed_items") or [])
		for d in items:
			if not frappe.get_cached_value("Item", d.item_code, "is_stock_item"):
				continue

			item_details = frappe.get_cached_value(
				"Item", d.item_code, ["has_serial_no", "has_batch_no", "has_expiry_date"], as_dict=1
			)

			if (
				set_zero_rate_for_expired_batch
				and item_details.has_batch_no
				and item_details.has_expiry_date
				and self.get("is_return")
				and not self.get("return_against")
				and is_batch_expired(d.batch_no, self.get("posting_date"))
			):
				# set incoming rate as zero for stand-lone credit note with expired batch
				d.incoming_rate = 0

			elif not self.get("return_against") or (
				get_valuation_method(d.item_code, self.company) == "Moving Average"
				and self.get("is_return")
				and not item_details.has_serial_no
				and not item_details.has_batch_no
			):
				# Get incoming rate based on original item cost based on valuation method
				qty = flt(d.get("stock_qty") or d.get("actual_qty") or d.get("qty"))

				if old_doc:
					reset_incoming_rate()

				if (
					not d.incoming_rate
					or (self.is_new() and not is_standalone)
					or self.is_internal_transfer()
					or (
						get_valuation_method(d.item_code, self.company) == "Moving Average"
						and self.get("is_return")
						and not is_standalone
					)
				):
					d.incoming_rate = _get_incoming_rate(
						{
							"item_code": d.item_code,
							"warehouse": d.warehouse,
							"posting_date": self.get("posting_date") or self.get("transaction_date"),
							"posting_time": self.get("posting_time") or nowtime(),
							"qty": qty if cint(self.get("is_return")) else (-1 * qty),
							"serial_and_batch_bundle": d.serial_and_batch_bundle,
							"company": self.company,
							"voucher_type": self.doctype,
							"voucher_no": self.name,
							"voucher_detail_no": d.name,
							"allow_zero_valuation": d.get("allow_zero_valuation_rate"),
							"batch_no": d.batch_no,
							"serial_no": d.serial_no,
						},
						raise_error_if_no_rate=is_standalone,
						fallbacks=not is_standalone,
					)

				if (
					not d.incoming_rate
					and self.get("return_against")
					and self.get("is_return")
					and get_valuation_method(d.item_code, self.company) == "Moving Average"
				):
					d.incoming_rate = get_rate_for_return(
						self.doctype, self.name, d.item_code, self.return_against, item_row=d
					)

				if (
					self.get("is_return")
					and not d.incoming_rate
					and not self.get("return_against")
					and not self.is_internal_transfer()
					and not d.get("allow_zero_valuation_rate")
				):
					d.incoming_rate = d.rate

				# For internal transfers use incoming rate as the valuation rate
				if self.is_internal_transfer():
					if self.doctype == "Delivery Note" or self.get("update_stock"):
						if d.doctype == "Packed Item":
							incoming_rate = flt(
								flt(d.incoming_rate, d.precision("incoming_rate")) * d.conversion_factor,
								d.precision("incoming_rate"),
							)
							if d.incoming_rate != incoming_rate:
								d.incoming_rate = incoming_rate
						else:
							if allow_at_arms_length_price:
								continue

							rate = flt(flt(d.incoming_rate) * flt(d.conversion_factor or 1.0))

							if flt(d.rate, d.precision("incoming_rate")) != flt(
								rate, d.precision("incoming_rate")
							):
								d.rate = rate
								frappe.msgprint(
									_(
										"Row {0}: Item rate has been updated as per valuation rate since its an internal stock transfer"
									).format(d.idx),
									alert=1,
								)

							d.discount_percentage = 0.0
							d.discount_amount = 0.0
							d.margin_rate_or_amount = 0.0

			elif self.get("return_against"):
				# Get incoming rate of return entry from reference document
				# based on original item cost as per valuation method
				d.incoming_rate = get_rate_for_return(
					self.doctype, self.name, d.item_code, self.return_against, item_row=d
				)

	def update_stock_ledger(self, allow_negative_stock=False):
		self.update_reserved_qty()

		sl_entries = []
		# Loop over items and packed items table
		for d in self.get_item_list():
			if frappe.get_cached_value("Item", d.item_code, "is_stock_item") == 1 and flt(d.qty):
				if flt(d.conversion_factor) == 0.0:
					d.conversion_factor = (
						get_conversion_factor(d.item_code, d.uom).get("conversion_factor") or 1.0
					)

				# On cancellation or return entry submission, make stock ledger entry for
				# target warehouse first, to update serial no values properly

				if d.warehouse and (
					(not cint(self.is_return) and self.docstatus == 1)
					or (cint(self.is_return) and self.docstatus == 2)
				):
					sl_entries.append(self.get_sle_for_source_warehouse(d))

				if d.target_warehouse:
					sl_entries.append(self.get_sle_for_target_warehouse(d))

				if d.warehouse and (
					(not cint(self.is_return) and self.docstatus == 2)
					or (cint(self.is_return) and self.docstatus == 1)
				):
					sl_entries.append(self.get_sle_for_source_warehouse(d))

		self.make_sl_entries(sl_entries, allow_negative_stock=allow_negative_stock)

	def get_sle_for_source_warehouse(self, item_row):
		serial_and_batch_bundle = (
			item_row.serial_and_batch_bundle
			if not self.is_internal_transfer() or self.docstatus == 1
			else None
		)

		if self.is_internal_transfer():
			if serial_and_batch_bundle and self.docstatus == 1 and self.is_return:
				serial_and_batch_bundle = self.make_package_for_transfer(
					serial_and_batch_bundle, item_row.warehouse, type_of_transaction="Inward"
				)
			elif not serial_and_batch_bundle:
				serial_and_batch_bundle = frappe.db.get_value(
					"Stock Ledger Entry",
					{"voucher_detail_no": item_row.name, "warehouse": item_row.warehouse},
					"serial_and_batch_bundle",
				)

		sle = self.get_sl_entries(
			item_row,
			{
				"actual_qty": -1 * get_ledger_qty(item_row.qty),
				"incoming_rate": item_row.incoming_rate,
				"recalculate_rate": cint(self.is_return),
				"serial_and_batch_bundle": serial_and_batch_bundle,
			},
		)
		if item_row.target_warehouse and not cint(self.is_return):
			sle.dependant_sle_voucher_detail_no = item_row.name

		return sle

	def get_sle_for_target_warehouse(self, item_row):
		sle = self.get_sl_entries(
			item_row, {"actual_qty": get_ledger_qty(item_row.qty), "warehouse": item_row.target_warehouse}
		)

		if self.docstatus == 1:
			if not cint(self.is_return):
				sle.update({"incoming_rate": item_row.incoming_rate, "recalculate_rate": 1})
			else:
				sle.update({"outgoing_rate": item_row.incoming_rate})
				if item_row.warehouse:
					sle.dependant_sle_voucher_detail_no = item_row.name

			if item_row.serial_and_batch_bundle and not cint(self.is_return):
				type_of_transaction = "Inward"
				if cint(self.is_return):
					type_of_transaction = "Outward"

				sle["serial_and_batch_bundle"] = self.make_package_for_transfer(
					item_row.serial_and_batch_bundle,
					item_row.target_warehouse,
					type_of_transaction=type_of_transaction,
				)

		return sle

	def set_po_nos(self, for_validate=False):
		if self.doctype == "Sales Invoice" and hasattr(self, "items"):
			if for_validate and self.po_no:
				return
			self.set_pos_for_sales_invoice()
		if self.doctype == "Delivery Note" and hasattr(self, "items"):
			if for_validate and self.po_no:
				return
			self.set_pos_for_delivery_note()

	def set_pos_for_sales_invoice(self):
		po_nos = []
		if self.po_no:
			po_nos.append(self.po_no)
		self.get_po_nos("Sales Order", "sales_order", po_nos)
		self.get_po_nos("Delivery Note", "delivery_note", po_nos)
		self.po_no = ", ".join(list(set(x.strip() for x in ",".join(po_nos).split(","))))

	def set_pos_for_delivery_note(self):
		po_nos = []
		if self.po_no:
			po_nos.append(self.po_no)
		self.get_po_nos("Sales Order", "against_sales_order", po_nos)
		self.get_po_nos("Sales Invoice", "against_sales_invoice", po_nos)
		self.po_no = ", ".join(list(set(x.strip() for x in ",".join(po_nos).split(","))))

	def get_po_nos(self, ref_doctype, ref_fieldname, po_nos):
		doc_list = list(set(d.get(ref_fieldname) for d in self.items if d.get(ref_fieldname)))
		if doc_list:
			po_nos += [
				d.po_no
				for d in frappe.get_all(ref_doctype, "po_no", filters={"name": ("in", doc_list)})
				if d.get("po_no")
			]

	def set_gross_profit(self):
		if self.doctype in ["Sales Order", "Quotation"]:
			for item in self.items:
				item.gross_profit = flt(
					((flt(item.stock_uom_rate) - flt(item.valuation_rate)) * item.stock_qty),
					self.precision("amount", item),
				)

	def set_customer_address(self):
		address_dict = {
			"customer_address": "address_display",
			"shipping_address_name": "shipping_address",
			"company_address": "company_address_display",
			"dispatch_address_name": "dispatch_address",
		}

		for address_field, address_display_field in address_dict.items():
			if self.get(address_field):
				self.set(
					address_display_field, render_address(self.get(address_field), check_permissions=False)
				)

	def validate_for_duplicate_items(self):
		check_list, chk_dupl_itm = [], []
		if cint(frappe.get_single_value("Selling Settings", "allow_multiple_items")):
			return
		if self.doctype == "Sales Invoice" and self.is_consolidated:
			return
		if self.doctype == "POS Invoice":
			return

		items = [item.item_code for item in self.get("items")]
		item_stock_map = frappe._dict(
			frappe.get_all(
				"Item",
				filters={"item_code": ["in", items]},
				fields=["item_code", "is_stock_item"],
				as_list=True,
			)
		)

		for d in self.get("items"):
			if self.doctype == "Sales Invoice":
				stock_items = [
					d.item_code,
					d.description,
					d.warehouse,
					d.sales_order or d.delivery_note,
					d.batch_no or "",
				]
				non_stock_items = [d.item_code, d.description, d.sales_order or d.delivery_note]
			elif self.doctype == "Delivery Note":
				stock_items = [
					d.item_code,
					d.description,
					d.warehouse,
					d.against_sales_order or d.against_sales_invoice,
					d.batch_no or "",
				]
				non_stock_items = [
					d.item_code,
					d.description,
					d.against_sales_order or d.against_sales_invoice,
				]
			elif self.doctype in ["Sales Order", "Quotation"]:
				stock_items = [d.item_code, d.description, d.warehouse, ""]
				non_stock_items = [d.item_code, d.description]

			duplicate_items_msg = _("Item {0} entered multiple times.").format(frappe.bold(d.item_code))
			duplicate_items_msg += "<br><br>"
			duplicate_items_msg += _("Please enable {} in {} to allow same item in multiple rows").format(
				frappe.bold(_("Allow Item to Be Added Multiple Times in a Transaction")),
				get_link_to_form("Selling Settings", "Selling Settings"),
			)
			if item_stock_map.get(d.item_code):
				if stock_items in check_list:
					frappe.throw(duplicate_items_msg)
				else:
					check_list.append(stock_items)
			else:
				if non_stock_items in chk_dupl_itm:
					frappe.throw(duplicate_items_msg)
				else:
					chk_dupl_itm.append(non_stock_items)

	def validate_target_warehouse(self):
		items = self.get("items") + (self.get("packed_items") or [])

		for d in items:
			if d.get("target_warehouse") and d.get("warehouse") == d.get("target_warehouse"):
				warehouse = frappe.bold(d.get("target_warehouse"))
				frappe.throw(
					_(
						"Row {0}: Delivery Warehouse ({1}) and Customer Warehouse ({2}) can not be same"
					).format(d.idx, warehouse, warehouse)
				)

		if not self.get("is_internal_customer") and any(d.get("target_warehouse") for d in items):
			msg = _("Target Warehouse is set for some items but the customer is not an internal customer.")
			msg += " " + _("This {} will be treated as material transfer.").format(_(self.doctype))
			frappe.msgprint(msg, title="Internal Transfer", alert=True)

	def validate_items(self):
		# validate items to see if they have is_sales_item enabled
		from erpnext.controllers.buying_controller import validate_item_type

		validate_item_type(self, "is_sales_item", "sales")

	def validate_sample_retention_warehouse(self):
		if self.get("is_return"):
			return

		sample_retention_warehouse = frappe.get_cached_value(
			"Company", self.company, "sample_retention_warehouse"
		)
		if not sample_retention_warehouse:
			return

		items = self.get("items") + (self.get("packed_items"))
		for item in items:
			if item.get("warehouse") == sample_retention_warehouse:
				frappe.throw(
					_("Row {0}: Cannot sell item {1} from Sample Retention Warehouse {2}").format(
						item.idx, frappe.bold(item.item_code), frappe.bold(sample_retention_warehouse)
					),
					title=_("Not Allowed"),
				)

	def validate_produced_serial_nos_against_reservation(self):
		"""Restrict delivery to the serial nos reserved for a Sales Order Item with ensure delivery by serial no."""

		from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
		from erpnext.stock.doctype.stock_reservation_entry.stock_reservation_entry import (
			get_sre_reserved_serial_nos_for_voucher_detail_nos,
		)

		if self.is_return or not frappe.db.get_single_value("Stock Settings", "enable_stock_reservation"):
			return

		so_field = "sales_order" if self.doctype == "Sales Invoice" else "against_sales_order"
		rows = [d for d in self.items if d.get(so_field) and d.so_detail]
		if not rows:
			return

		flagged_so_details = frappe.get_all(
			"Sales Order Item",
			filters={
				"name": ("in", [d.so_detail for d in rows]),
				"ensure_delivery_based_on_produced_serial_no": 1,
			},
			pluck="name",
		)
		rows = [d for d in rows if d.so_detail in flagged_so_details]
		if not rows:
			return

		reserved_serial_nos = get_sre_reserved_serial_nos_for_voucher_detail_nos(
			"Sales Order", flagged_so_details
		)
		bundle_map = dict(
			frappe.get_all(
				rows[0].doctype,
				filters={"name": ("in", [d.name for d in rows])},
				fields=["name", "serial_and_batch_bundle"],
				as_list=True,
			)
		)
		bundle_serial_nos = frappe._dict()
		if bundles := [b for b in bundle_map.values() if b]:
			for entry in frappe.get_all(
				"Serial and Batch Entry",
				filters={"parent": ("in", bundles), "serial_no": ("is", "set")},
				fields=["parent", "serial_no"],
			):
				bundle_serial_nos.setdefault(entry.parent, []).append(entry.serial_no)

		for row in rows:
			if not reserved_serial_nos.get(row.so_detail):
				frappe.throw(
					_(
						"Row #{0}: Delivery of Item {1} is ensured by produced Serial No, but no Serial No is reserved against Sales Order {2}. Reserve the produced Serial Nos from the Sales Order."
					).format(row.idx, frappe.bold(row.item_code), frappe.bold(row.get(so_field))),
					title=_("Serial No Not Reserved"),
				)

			bundle = bundle_map.get(row.name)
			serial_nos = bundle_serial_nos.get(bundle, []) if bundle else get_serial_nos(row.serial_no)
			if invalid_serial_nos := [
				sn for sn in serial_nos if sn not in reserved_serial_nos[row.so_detail]
			]:
				frappe.throw(
					_(
						"Row #{0}: Serial No {1} is not reserved against Sales Order {2}. Deliver only the Serial Nos produced and reserved for it."
					).format(
						row.idx, frappe.bold(", ".join(invalid_serial_nos)), frappe.bold(row.get(so_field))
					),
					title=_("Serial No Not Reserved"),
				)

	def update_stock_reservation_entries(self) -> None:
		"""Updates Delivered Qty in Stock Reservation Entries."""

		# Don't update Delivered Qty on Return.
		if self.is_return:
			return

		so_field = "sales_order" if self.doctype == "Sales Invoice" else "against_sales_order"

		if self._action == "submit":
			for item in self.get("items"):
				# Skip if `Sales Order` or `Sales Order Item` reference is not set.
				if not item.get(so_field) or not item.so_detail:
					continue

				table = frappe.qb.DocType("Stock Reservation Entry")
				query = (
					frappe.qb.from_(table)
					.select(table.name)
					.where(
						(table.docstatus == 1)
						& (table.voucher_type == "Sales Order")
						& (table.voucher_no == item.get(so_field))
						& (table.voucher_detail_no == item.so_detail)
						& (table.warehouse == item.warehouse)
						& (table.delivered_qty < table.reserved_qty)
					)
					.orderby(table.creation)
				)
				sre_list = query.run(pluck="name")

				# Skip if no Stock Reservation Entries.
				if not sre_list:
					continue

				qty_to_deliver = item.stock_qty
				for sre in sre_list:
					if qty_to_deliver <= 0:
						break

					sre_doc = frappe.get_doc("Stock Reservation Entry", sre)

					qty_can_be_deliver = 0
					if sre_doc.reservation_based_on == "Serial and Batch":
						sbb = frappe.get_doc("Serial and Batch Bundle", item.serial_and_batch_bundle)
						if sre_doc.has_serial_no:
							delivered_serial_nos = [d.serial_no for d in sbb.entries]
							for entry in sre_doc.sb_entries:
								if entry.serial_no in delivered_serial_nos:
									entry.delivered_qty = 1  # Qty will always be 0 or 1 for Serial No.
									entry.db_update()
									qty_can_be_deliver += 1
									delivered_serial_nos.remove(entry.serial_no)
						else:
							delivered_batch_qty = {d.batch_no: -1 * d.qty for d in sbb.entries}
							for entry in sre_doc.sb_entries:
								if entry.batch_no in delivered_batch_qty:
									delivered_qty = min(
										(entry.qty - entry.delivered_qty), delivered_batch_qty[entry.batch_no]
									)
									entry.delivered_qty += delivered_qty
									entry.db_update()
									qty_can_be_deliver += delivered_qty
									delivered_batch_qty[entry.batch_no] -= delivered_qty
					else:
						# `Delivered Qty` should be less than or equal to `Reserved Qty`.
						qty_can_be_deliver = min(
							(sre_doc.reserved_qty - sre_doc.delivered_qty), qty_to_deliver
						)

					sre_doc.delivered_qty += qty_can_be_deliver
					sre_doc.db_update()

					# Update Stock Reservation Entry `Status` based on `Delivered Qty`.
					sre_doc.update_status()

					# Update Reserved Stock in Bin.
					sre_doc.update_reserved_stock_in_bin()
					sre_doc.update_reserved_qty_in_voucher(update_modified=False)

					qty_to_deliver -= qty_can_be_deliver

		if self._action == "cancel":
			for item in self.get("items"):
				# Skip if `Sales Order` or `Sales Order Item` reference is not set.
				if not item.get(so_field) or not item.so_detail:
					continue

				sre_list = frappe.db.get_all(
					"Stock Reservation Entry",
					{
						"docstatus": 1,
						"voucher_type": "Sales Order",
						"voucher_no": item.get(so_field),
						"voucher_detail_no": item.so_detail,
						"warehouse": item.warehouse,
						"status": ["in", ["Partially Delivered", "Delivered", "Partially Used", "Closed"]],
					},
					order_by="creation",
				)

				# Skip if no Stock Reservation Entries.
				if not sre_list:
					continue

				qty_to_undelivered = item.stock_qty
				for sre in sre_list:
					if qty_to_undelivered <= 0:
						break

					sre_doc = frappe.get_doc("Stock Reservation Entry", sre)

					qty_can_be_undelivered = 0
					if sre_doc.reservation_based_on == "Serial and Batch":
						sbb = frappe.get_doc("Serial and Batch Bundle", item.serial_and_batch_bundle)
						if sre_doc.has_serial_no:
							serial_nos_to_undelivered = [d.serial_no for d in sbb.entries]
							for entry in sre_doc.sb_entries:
								if entry.serial_no in serial_nos_to_undelivered:
									entry.delivered_qty = 0  # Qty will always be 0 or 1 for Serial No.
									entry.db_update()
									qty_can_be_undelivered += 1
									serial_nos_to_undelivered.remove(entry.serial_no)
						else:
							batch_qty_to_undelivered = {d.batch_no: -1 * d.qty for d in sbb.entries}
							for entry in sre_doc.sb_entries:
								if entry.batch_no in batch_qty_to_undelivered:
									undelivered_qty = min(
										entry.delivered_qty, batch_qty_to_undelivered[entry.batch_no]
									)
									entry.delivered_qty -= undelivered_qty
									entry.db_update()
									qty_can_be_undelivered += undelivered_qty
									batch_qty_to_undelivered[entry.batch_no] -= undelivered_qty
					else:
						# `Qty to Undelivered` should be less than or equal to `Delivered Qty`.
						qty_can_be_undelivered = min(sre_doc.delivered_qty, qty_to_undelivered)

					sre_doc.delivered_qty -= qty_can_be_undelivered
					sre_doc.db_update()

					# Update Stock Reservation Entry `Status` based on `Delivered Qty`.
					sre_doc.update_status()

					# Update Reserved Stock in Bin.
					sre_doc.update_reserved_stock_in_bin()
					sre_doc.update_reserved_qty_in_voucher(update_modified=False)

					qty_to_undelivered -= qty_can_be_undelivered

	def set_serial_and_batch_bundle_from_pick_list(self):
		from erpnext.stock.serial_batch_bundle import SerialBatchCreation

		for item in self.items:
			if item.use_serial_batch_fields or not item.against_pick_list or not self.get("update_stock", 1):
				continue

			if item.pick_list_item and not item.serial_and_batch_bundle:
				filters = {
					"item_code": item.item_code,
					"voucher_type": "Pick List",
					"voucher_no": item.against_pick_list,
					"voucher_detail_no": item.pick_list_item,
				}

				bundle_id = frappe.db.get_value("Serial and Batch Bundle", filters, "name")

				if bundle_id:
					cls_obj = SerialBatchCreation(
						{
							"type_of_transaction": "Outward",
							"serial_and_batch_bundle": bundle_id,
							"item_code": item.get("item_code"),
							"warehouse": item.get("warehouse"),
						}
					)

					cls_obj.duplicate_package()

					item.serial_and_batch_bundle = cls_obj.serial_and_batch_bundle

	def update_pick_list_status(self):
		from erpnext.stock.doctype.pick_list.pick_list import update_pick_list_status

		pick_lists = {row.against_pick_list for row in self.items if row.against_pick_list}
		for pick_list in pick_lists:
			update_pick_list_status(pick_list)


def set_default_income_account_for_item(obj):
	"""Set income account as default for items in the transaction.

	Updates the item default income account for each item in the transaction
	if it differs from the company's default income account.

	Args:
	    obj: Transaction document containing items table with income_account field
	"""
	company_default = frappe.get_cached_value("Company", obj.company, "default_income_account")
	for d in sorted(obj.get("items", default=[]), key=lambda row: row.item_code or ""):
		income_account = getattr(d, "income_account", None)
		if d.item_code and income_account and income_account != company_default:
			set_item_default(d.item_code, obj.company, "income_account", income_account)


def get_serial_and_batch_bundle(child, parent, delivery_note_child=None):
	from erpnext.stock.serial_batch_bundle import SerialBatchCreation

	if parent.get("is_return") and parent.get("packed_items"):
		return

	if child.get("use_serial_batch_fields"):
		return

	if not frappe.get_single_value("Stock Settings", "auto_create_serial_and_batch_bundle_for_outward"):
		return

	item_details = frappe.db.get_value("Item", child.item_code, ["has_serial_no", "has_batch_no"], as_dict=1)

	if not item_details.has_serial_no and not item_details.has_batch_no:
		return

	sn_doc = SerialBatchCreation(
		{
			"item_code": child.item_code,
			"warehouse": child.warehouse,
			"voucher_type": parent.doctype,
			"voucher_no": parent.name if parent.docstatus < 2 else None,
			"voucher_detail_no": delivery_note_child.name if delivery_note_child else child.name,
			"posting_datetime": get_combine_datetime(parent.posting_date, parent.posting_time),
			"qty": child.qty,
			"type_of_transaction": "Outward" if child.qty > 0 and parent.docstatus < 2 else "Inward",
			"company": parent.company,
			"do_not_submit": "True",
		}
	)

	doc = sn_doc.make_serial_and_batch_bundle()
	child.db_set("serial_and_batch_bundle", doc.name)

	return doc.name


def get_reserved_warehouses_by_order_row(so_details: list) -> dict:
	if not so_details:
		return {}

	sre = frappe.qb.DocType("Stock Reservation Entry")
	rows = (
		frappe.qb.from_(sre)
		.select(sre.voucher_detail_no, sre.warehouse)
		.where(
			(sre.docstatus == 1)
			& (sre.voucher_type == "Sales Order")
			& (sre.voucher_detail_no.isin(so_details))
			& (sre.delivered_qty < sre.reserved_qty)
		)
		.groupby(sre.voucher_detail_no, sre.warehouse)
		.orderby(Min(sre.creation))
	).run(as_dict=True)

	reserved_warehouses = {}
	for row in rows:
		reserved_warehouses.setdefault(row.voucher_detail_no, []).append(row.warehouse)

	return reserved_warehouses


def get_ledger_qty(qty) -> float:
	"""Stock qty as the stock ledger keeps balances, rounded to its precision."""
	return flt(qty, frappe.get_precision("Stock Ledger Entry", "actual_qty"))