# Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt

from collections import defaultdict

import frappe
from frappe import _
from frappe.utils import flt


def execute(filters: dict | None = None):
	report = ReportData(filters)
	report.validate_filters()
	data = report.get_data()
	has_serial_no, has_batch_no = check_has_serial_no_in_data(data)
	columns = report.get_columns(has_serial_no, has_batch_no)

	return columns, data


def check_has_serial_no_in_data(data):
	has_serial_no = False
	has_batch_no = False

	for row in data:
		if row.get("serial_no"):
			has_serial_no = True
		if row.get("batch_no"):
			has_batch_no = True

		if has_serial_no and has_batch_no:
			break

	return has_serial_no, has_batch_no


class ReportData:
	def __init__(self, filters):
		self.filters = filters
		self.doctype_name = self.get_doctype()
		self.mapped_fg_entries = {}
		self.forward_trace_path = set()
		self.backward_trace_path = set()

	def validate_filters(self):
		if not self.filters.item_code and not self.filters.batches and not self.filters.serial_nos:
			frappe.throw(
				_("Please select at least one filter: Item Code, Batch, or Serial No."),
				title=_("Missing Filters"),
			)

	def get_data(self):
		result_data = []

		if self.filters.get("traceability_direction") in ["Backward", "Both"]:
			data = self.split_by_production_entries(self.get_serial_no_batches())
			source_data = self.prepare_source_data(data)

			# Prepare source data with raw materials
			for key in source_data:
				sabb_data = source_data[key]
				if sabb_data.reference_doctype != "Stock Entry":
					continue

				self.set_backward_data(sabb_data)

			# Source data has all the details including raw materials
			self.parse_batch_details(source_data, result_data, "Backward")

		if self.filters.get("traceability_direction") in ["Forward", "Both"]:
			data = self.get_serial_no_batches()
			batch_details = frappe._dict({})
			for row in data:
				value = row.serial_no or row.batch_no
				self.set_forward_data(value, batch_details)

			self.parse_batch_details(batch_details, result_data, "Forward")

		return result_data

	def parse_batch_details(self, sabb_data_details, data, direction, indent=0):
		for key in sabb_data_details:
			sabb = sabb_data_details[key]
			row = {
				"item_code": sabb.item_code,
				"batch_no": sabb.batch_no,
				"serial_no": sabb.serial_no,
				"warehouse": sabb.warehouse,
				"qty": sabb.qty,
				"reference_doctype": sabb.reference_doctype,
				"reference_name": sabb.reference_name,
				"item_name": sabb.item_name,
				"posting_datetime": sabb.posting_datetime,
				"indent": indent,
				"direction": direction,
				"batch_expiry_date": sabb.get("batch_expiry_date"),
				"warranty_expiry_date": sabb.get("warranty_expiry_date"),
				"amc_expiry_date": sabb.get("amc_expiry_date"),
			}

			if data and indent == 0:
				data.append({})

			if sabb.reference_doctype == "Purchase Receipt":
				row["supplier"] = frappe.db.get_value(
					"Purchase Receipt",
					sabb.reference_name,
					"supplier",
				)
			elif sabb.reference_doctype == "Stock Entry":
				row["work_order"] = frappe.db.get_value(
					"Stock Entry",
					sabb.reference_name,
					"work_order",
				)
			elif sabb.reference_doctype == "Delivery Note":
				row["customer"] = frappe.db.get_value(
					"Delivery Note",
					sabb.reference_name,
					"customer",
				)

			data.append(row)

			raw_materials = sabb.get("raw_materials")
			if raw_materials:
				self.parse_batch_details(raw_materials, data, direction, indent + 1)

		return data

	def split_by_production_entries(self, rows):
		production_entries = self.get_production_entries([row.serial_no or row.batch_no for row in rows])

		result = []
		for row in rows:
			entries = production_entries.get(row.serial_no or row.batch_no)
			if not entries:
				result.append(row)
				continue

			for stock_entry in entries:
				result.append(frappe._dict(row, reference_doctype="Stock Entry", reference_name=stock_entry))

		return result

	def get_production_entries(self, values):
		if not values:
			return {}

		sabb = frappe.qb.DocType("Serial and Batch Bundle")
		sabe = frappe.qb.DocType("Serial and Batch Entry")
		stock_entry = frappe.qb.DocType("Stock Entry")
		value_field = sabe.serial_no if self.doctype_name == "Serial No" else sabe.batch_no

		query = (
			frappe.qb.from_(sabb)
			.inner_join(sabe)
			.on(sabb.name == sabe.parent)
			.inner_join(stock_entry)
			.on(stock_entry.name == sabb.voucher_no)
			.select(value_field.as_("value"), sabb.voucher_no)
			.where(
				(sabb.voucher_type == "Stock Entry")
				& (sabb.type_of_transaction == "Inward")
				& (sabb.is_cancelled == 0)
				& (sabe.docstatus == 1)
				& (stock_entry.purpose.isin(["Manufacture", "Repack"]))
				& (value_field.isin(values))
			)
			.orderby(sabb.posting_datetime)
		)

		production_entries = defaultdict(list)
		for row in query.run(as_dict=True):
			if row.voucher_no not in production_entries[row.value]:
				production_entries[row.value].append(row.voucher_no)

		return production_entries

	def prepare_source_data(self, data):
		source_data = frappe._dict({})
		for row in data:
			key = (row.item_code, row.reference_name)

			value = row.serial_no or row.batch_no
			if value:
				key = (row.item_code, row.reference_name, value)

			sabb_details = self.get_data_from_sabb(row)
			row.update(sabb_details)

			if key not in source_data:
				row["raw_materials"] = frappe._dict({})
				source_data[key] = row

		return source_data

	def get_data_from_sabb(self, row):
		sabb = frappe.qb.DocType("Serial and Batch Bundle")
		sabb_entry = frappe.qb.DocType("Serial and Batch Entry")

		query = (
			frappe.qb.from_(sabb)
			.inner_join(sabb_entry)
			.on(sabb.name == sabb_entry.parent)
			.select(
				sabb_entry.qty,
				sabb_entry.warehouse,
				sabb_entry.posting_datetime,
			)
			.where(
				(sabb.voucher_type == row.reference_doctype)
				& (sabb.voucher_no == row.reference_name)
				& (sabb.is_cancelled == 0)
				& (sabb_entry.docstatus == 1)
			)
		)

		if row.batch_no:
			query = query.where(sabb_entry.batch_no == row.batch_no)
		else:
			query = query.where(sabb_entry.serial_no == row.serial_no)

		results = query.run(as_dict=True)
		if not results:
			return {}

		return frappe._dict(results[0], qty=sum(flt(result.qty) for result in results))

	def set_backward_data(self, sabb_data, qty=None):
		if qty:
			sabb_data.qty = qty

		if "raw_materials" not in sabb_data:
			sabb_data.raw_materials = frappe._dict({})

		self.backward_trace_path.add(sabb_data.reference_name)
		try:
			for material in self.get_materials(sabb_data):
				self.add_backward_material(sabb_data, material)
		finally:
			self.backward_trace_path.discard(sabb_data.reference_name)

		return sabb_data

	def add_backward_material(self, sabb_data, material):
		# raw material mapped to a different finished good serial / batch of the same entry
		if material.fg_serial_no and material.fg_serial_no != sabb_data.serial_no:
			return

		if material.fg_batch_no and material.fg_batch_no != sabb_data.batch_no:
			return

		# Recursive: batch has sub-components
		if material.serial_no or material.batch_no:
			self.set_backward_data_for_component(sabb_data, material)
			return

		sub_key = (material.item_code, material.name)
		if sub_key not in sabb_data.raw_materials:
			sabb_data.raw_materials[sub_key] = frappe._dict(
				{
					"item_code": material.item_code,
					"item_name": material.item_name,
					"qty": material.qty or material.quantity,
					"warehouse": material.warehouse,
				}
			)

	def set_backward_data_for_component(self, sabb_data, material):
		value = material.serial_no or material.batch_no
		key = (material.item_code, material.reference_name, material.name, value)
		if key not in sabb_data.raw_materials:
			for idx, details in enumerate(self.get_component_sources(value)):
				sabb_data.raw_materials[key if idx == 0 else (*key, details.reference_name)] = details

		sources = [details for node_key, details in sabb_data.raw_materials.items() if node_key[:4] == key]
		produced_qty = sum(abs(flt(details.qty)) for details in sources)
		for details in sources:
			share = abs(flt(details.qty)) / produced_qty if produced_qty else 1 / len(sources)
			self.set_backward_data(details, flt(material.qty) * share)

	def get_component_sources(self, value):
		details = self.get_serial_no_batches(value)
		if not details:
			inward_data = self.get_sabb_entries(value, "Inward")
			details = inward_data[-1] if inward_data else None

		if not details:
			return []

		sources = [
			source
			for source in self.split_by_production_entries([details])
			if source.reference_name not in self.backward_trace_path
		]
		if not sources and details.reference_name not in self.backward_trace_path:
			sources = [details]

		for source in sources:
			source.update(self.get_data_from_sabb(source))

		return sources

	def get_serial_no_batches(self, name=None):
		batches = self.filters.get("batches", [])
		serial_nos = self.filters.get("serial_nos", [])

		doctype = frappe.qb.DocType(self.doctype_name)
		query = frappe.qb.from_(doctype).select(
			doctype.reference_doctype,
			doctype.reference_name,
			doctype.item_name,
		)

		if self.doctype_name == "Batch":
			query = query.select(
				doctype.item.as_("item_code"),
				doctype.name.as_("batch_no"),
				doctype.expiry_date.as_("batch_expiry_date"),
			)
		else:
			query = query.select(
				doctype.item_code,
				doctype.name.as_("serial_no"),
				doctype.warranty_expiry_date,
				doctype.amc_expiry_date,
			)

		if name:
			query = query.where(doctype.name == name)
			data = query.run(as_dict=True)
			return data[0] if data else {}

		if batches and self.doctype_name == "Serial No":
			query = query.where(doctype.batch_no.isin(batches))
		elif batches:
			query = query.where(doctype.name.isin(batches))
		elif serial_nos:
			query = query.where(doctype.name.isin(serial_nos))

		if self.filters.get("item_code"):
			if self.doctype_name == "Serial No":
				query = query.where(doctype.item_code == self.filters.item_code)
			else:
				query = query.where(doctype.item == self.filters.item_code)

		return query.run(as_dict=True)

	def get_doctype(self):
		if self.filters.item_code:
			item_details = frappe.get_cached_value(
				"Item",
				self.filters.item_code,
				["has_batch_no", "has_serial_no"],
				as_dict=True,
			)

			if item_details.has_serial_no:
				return "Serial No"
			elif item_details.has_batch_no:
				return "Batch"

		elif self.filters.get("serial_nos"):
			return "Serial No"

		return "Batch"

	def get_materials(self, sabb_data):
		stock_entry = frappe.qb.DocType("Stock Entry")
		stock_entry_detail = frappe.qb.DocType("Stock Entry Detail")
		sabb_entry = frappe.qb.DocType("Serial and Batch Entry")

		query = (
			frappe.qb.from_(stock_entry)
			.inner_join(stock_entry_detail)
			.on(stock_entry.name == stock_entry_detail.parent)
			.left_join(sabb_entry)
			.on(
				(stock_entry_detail.serial_and_batch_bundle == sabb_entry.parent)
				& (sabb_entry.docstatus == 1)
			)
			.select(
				stock_entry_detail.s_warehouse.as_("warehouse"),
				stock_entry_detail.item_code,
				stock_entry_detail.name,
				stock_entry_detail.item_name,
				stock_entry_detail.parenttype.as_("reference_doctype"),
				stock_entry.name.as_("reference_name"),
				stock_entry_detail.qty.as_("row_qty"),
				stock_entry.fg_completed_qty,
				sabb_entry.batch_no,
				sabb_entry.serial_no,
				sabb_entry.qty.as_("quantity"),
				sabb_entry.fg_serial_no,
				sabb_entry.fg_batch_no,
			)
			.where(
				(stock_entry.docstatus == 1)
				& (stock_entry.purpose.isin(["Manufacture", "Repack"]))
				& (stock_entry.name == sabb_data.reference_name)
				& (stock_entry_detail.s_warehouse.isnotnull())
			)
		)

		materials = query.run(as_dict=True)
		for material in materials:
			material.qty = get_material_qty(material, sabb_data.qty, self.get_mapped_fg_batch_qty(material))

		return materials

	def get_mapped_fg_batch_qty(self, material):
		if not material.fg_batch_no or material.fg_serial_no:
			return 0

		return sum(
			abs(flt(entry.qty))
			for entry in self.get_mapped_fg_entries(material.reference_name, material.item_code)
			if entry.batch_no == material.fg_batch_no and not entry.serial_no
		)

	def set_forward_data(self, value, sabb_data):
		if value in self.forward_trace_path:
			return

		self.forward_trace_path.add(value)
		try:
			for row in self.get_sabb_entries(value):
				if row.reference_doctype == "Stock Entry":
					self.process_manufacture_or_repack_entry(row, sabb_data)
				else:
					self.add_direct_outward_entry(row, sabb_data)
		finally:
			self.forward_trace_path.discard(value)

	def add_direct_outward_entry(self, row, batch_details):
		key = (row.item_code, row.reference_name, row.serial_no, row.batch_no)
		if key not in batch_details:
			row["indent"] = 0
			batch_details[key] = row

	def get_sabb_entries(self, value, type_of_transaction=None):
		if not type_of_transaction:
			type_of_transaction = "Outward"

		SABB = frappe.qb.DocType("Serial and Batch Bundle")
		SABE = frappe.qb.DocType("Serial and Batch Entry")

		query = (
			frappe.qb.from_(SABB)
			.inner_join(SABE)
			.on(SABB.name == SABE.parent)
			.select(
				SABB.voucher_type.as_("reference_doctype"),
				SABB.voucher_no.as_("reference_name"),
				SABE.batch_no,
				SABE.serial_no,
				SABE.qty,
				SABB.item_code,
				SABB.item_name,
				SABB.posting_datetime,
				SABB.warehouse,
				SABE.fg_serial_no,
				SABE.fg_batch_no,
			)
			.where(
				(SABB.is_cancelled == 0)
				& (SABE.docstatus == 1)
				& (SABB.type_of_transaction == type_of_transaction)
			)
			.orderby(SABB.posting_datetime)
		)

		query = query.where((SABE.serial_no == value) | (SABE.batch_no == value))

		return query.run(as_dict=True)

	def process_manufacture_or_repack_entry(self, row, batch_details):
		ste = frappe.db.get_value("Stock Entry", row.reference_name, ["purpose", "work_order"], as_dict=True)

		if ste and ste.purpose in ["Manufacture", "Repack"]:
			fg_item = self.get_finished_item_from_stock_entry(row.reference_name)
			if not fg_item:
				return

			mapped_fg_entries = self.get_mapped_fg_entries(row.reference_name, row.item_code)
			fg_value = row.fg_serial_no or row.fg_batch_no
			if fg_value:
				# the finished good row that produced the mapped serial / batch
				fg_entries = [
					entry
					for entry in mapped_fg_entries
					if (row.fg_serial_no and entry.serial_no == row.fg_serial_no)
					or (row.fg_batch_no and entry.batch_no == row.fg_batch_no)
				]
				if fg_entries:
					fg_item = self.get_finished_item_from_stock_entry(
						row.reference_name, fg_entries[0].detail_name
					)

			key = (fg_item.item_code, row.reference_name, fg_value)

			if key not in batch_details:
				serial_no, batch_no = self.get_serial_batch_no(fg_item.serial_and_batch_bundle)
				if fg_value:
					serial_no, batch_no = row.fg_serial_no, row.fg_batch_no
					fg_item.qty = sum(abs(entry.qty) for entry in fg_entries) or fg_item.qty
				elif mapped_fg_entries:
					# unmapped raw materials of a partly mapped entry: only the unmapped production
					serial_no, batch_no = None, None
					fg_item.qty -= sum(
						abs(entry.qty) for entry in mapped_fg_entries if entry.detail_name == fg_item.name
					)

				fg_item.update(
					{
						"work_order": ste.work_order,
						"posting_datetime": row.posting_datetime,
						"serial_no": serial_no,
						"batch_no": batch_no,
						"indent": 0,
						"warehouse": fg_item.warehouse,
						"raw_materials": frappe._dict(
							{(row.item_code, row.reference_name, row.serial_no, row.batch_no): row}
						),
					}
				)
				batch_details[key] = fg_item
				if serial_no or batch_no:
					self.set_forward_data(serial_no or batch_no, fg_item.raw_materials)
			else:
				batch_details[key].raw_materials[
					(row.item_code, row.reference_name, row.serial_no, row.batch_no)
				] = row

	def get_finished_item_from_stock_entry(self, reference_name, detail_name=None):
		filters = {"parent": reference_name, "is_finished_item": 1}
		if detail_name:
			filters["name"] = detail_name

		return frappe.db.get_value(
			"Stock Entry Detail",
			filters,
			[
				"name",
				"item_code",
				"item_name",
				"serial_and_batch_bundle",
				"qty",
				"parenttype as reference_doctype",
				"parent as reference_name",
				"t_warehouse as warehouse",
			],
			as_dict=True,
		)

	def get_mapped_fg_entries(self, stock_entry, item_code):
		"""Finished good serial / batch entries of the stock entry that this raw material item is mapped to."""
		if (stock_entry, item_code) in self.mapped_fg_entries:
			return self.mapped_fg_entries[(stock_entry, item_code)]

		sed = frappe.qb.DocType("Stock Entry Detail")
		sabe = frappe.qb.DocType("Serial and Batch Entry")

		mapped_values = (
			frappe.qb.from_(sed)
			.inner_join(sabe)
			.on(sed.serial_and_batch_bundle == sabe.parent)
			.select(sabe.fg_serial_no, sabe.fg_batch_no)
			.distinct()
			.where(
				(sed.parent == stock_entry)
				& (sed.item_code == item_code)
				& (sed.is_finished_item == 0)
				& (sabe.fg_serial_no.isnotnull() | sabe.fg_batch_no.isnotnull())
			)
		).run(as_dict=True)

		mapped_entries = []
		if mapped_values:
			serial_nos = {row.fg_serial_no for row in mapped_values if row.fg_serial_no}
			batch_nos = {row.fg_batch_no for row in mapped_values if row.fg_batch_no}

			fg_entries = (
				frappe.qb.from_(sed)
				.inner_join(sabe)
				.on(sed.serial_and_batch_bundle == sabe.parent)
				.select(sed.name.as_("detail_name"), sabe.serial_no, sabe.batch_no, sabe.qty)
				.where((sed.parent == stock_entry) & (sed.is_finished_item == 1))
			).run(as_dict=True)

			mapped_entries = [
				entry
				for entry in fg_entries
				if (entry.serial_no and entry.serial_no in serial_nos)
				or (not entry.serial_no and entry.batch_no in batch_nos)
			]

		self.mapped_fg_entries[(stock_entry, item_code)] = mapped_entries
		return mapped_entries

	def get_serial_batch_no(self, serial_and_batch_bundle):
		sabb_details = frappe.get_all(
			"Serial and Batch Entry",
			filters={"parent": serial_and_batch_bundle},
			fields=["batch_no", "serial_no"],
		)

		# a bundle with several serial / batch nos can't be pinned to one of them
		serial_nos = {row.serial_no for row in sabb_details}
		batch_nos = {row.batch_no for row in sabb_details}
		return (
			serial_nos.pop() if len(serial_nos) == 1 else None,
			batch_nos.pop() if len(batch_nos) == 1 else None,
		)

	def get_columns(self, has_serial_no=None, has_batch_no=None):
		columns = [
			{
				"fieldname": "item_code",
				"label": _("Item Code"),
				"fieldtype": "Link",
				"options": "Item",
				"width": 180,
			},
			{
				"fieldname": "item_name",
				"label": _("Item Name"),
				"fieldtype": "Data",
				"width": 120,
			},
		]

		if has_serial_no:
			columns.append(
				{
					"fieldname": "serial_no",
					"label": _("Serial No"),
					"fieldtype": "Link",
					"options": "Serial No",
					"width": 120,
				}
			)

		if has_batch_no:
			columns.extend(
				[
					{
						"fieldname": "batch_no",
						"label": _("Batch No"),
						"fieldtype": "Link",
						"options": "Batch",
						"width": 120,
					},
					{
						"fieldname": "batch_expiry_date",
						"label": _("Batch Expiry Date"),
						"fieldtype": "Date",
						"width": 150,
					},
				]
			)

		columns.extend(
			[
				{
					"fieldname": "qty",
					"label": _("Quantity"),
					"fieldtype": "Float",
					"width": 90,
				},
				{
					"fieldname": "reference_doctype",
					"label": _("Voucher Type"),
					"fieldtype": "Link",
					"options": "DocType",
					"width": 130,
				},
				{
					"fieldname": "reference_name",
					"label": _("Source Document No"),
					"fieldtype": "Dynamic Link",
					"options": "reference_doctype",
					"width": 200,
				},
				{
					"fieldname": "warehouse",
					"label": _("Warehouse"),
					"fieldtype": "Link",
					"options": "Warehouse",
					"width": 120,
				},
				{
					"fieldname": "posting_datetime",
					"label": _("Posting Datetime"),
					"fieldtype": "Datetime",
					"width": 120,
				},
				{
					"fieldname": "work_order",
					"label": _("Work Order"),
					"fieldtype": "Link",
					"options": "Work Order",
					"width": 160,
				},
			]
		)

		if self.filters.get("traceability_direction") == "Backward":
			columns.append(
				{
					"fieldname": "supplier",
					"label": _("Supplier"),
					"fieldtype": "Link",
					"options": "Supplier",
					"width": 150,
				}
			)
		else:
			columns.append(
				{
					"fieldname": "customer",
					"label": _("Customer"),
					"fieldtype": "Link",
					"options": "Customer",
					"width": 150,
				}
			)

		if has_serial_no:
			columns.extend(
				[
					{
						"fieldname": "warranty_expiry_date",
						"label": _("Warranty Expiry (Serial)"),
						"fieldtype": "Date",
						"width": 200,
					},
					{
						"fieldname": "amc_expiry_date",
						"label": _("AMC Expiry (Serial)"),
						"fieldtype": "Date",
						"width": 160,
					},
				]
			)

		return columns


def get_material_qty(material, fg_qty, mapped_fg_qty=0):
	if material.serial_no or material.fg_serial_no:
		return abs(flt(material.quantity))

	if material.fg_batch_no:
		return abs(flt(material.quantity)) * (
			flt(fg_qty) / mapped_fg_qty if fg_qty and mapped_fg_qty else 1.0
		)

	ratio = flt(fg_qty) / (flt(material.fg_completed_qty) or flt(fg_qty)) if fg_qty else 1.0

	if material.batch_no:
		return abs(flt(material.quantity)) * ratio

	return flt(material.row_qty) * ratio