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

import frappe
from frappe import _, throw
from frappe.query_builder import Criterion
from frappe.query_builder.functions import Max
from frappe.utils import add_days, cint, cstr, date_diff, escape_html, formatdate, getdate

from erpnext.setup.doctype.employee.employee import get_holiday_list_for_employee
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos, get_serial_nos_from_sle_list
from erpnext.utilities.transaction_base import TransactionBase, delete_events


class MaintenanceSchedule(TransactionBase):
	# 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.maintenance.doctype.maintenance_schedule_detail.maintenance_schedule_detail import (
			MaintenanceScheduleDetail,
		)
		from erpnext.maintenance.doctype.maintenance_schedule_item.maintenance_schedule_item import (
			MaintenanceScheduleItem,
		)

		address_display: DF.TextEditor | None
		amended_from: DF.Link | None
		company: DF.Link
		contact_display: DF.SmallText | None
		contact_email: DF.Data | None
		contact_mobile: DF.Data | None
		contact_person: DF.Link | None
		customer: DF.Link | None
		customer_address: DF.Link | None
		customer_group: DF.Link | None
		customer_name: DF.Data | None
		items: DF.Table[MaintenanceScheduleItem]
		naming_series: DF.Literal["MAT-MSH-.YYYY.-"]
		schedules: DF.Table[MaintenanceScheduleDetail]
		status: DF.Literal["", "Draft", "Submitted", "Cancelled"]
		territory: DF.Link | None
		transaction_date: DF.Date
	# end: auto-generated types

	@frappe.whitelist()
	def generate_schedule(self):
		if self.docstatus != 0:
			return
		self.set("schedules", [])
		count = 1
		for d in self.get("items"):
			self.validate_maintenance_detail()
			s_list = []
			s_list = self.create_schedule_list(d.start_date, d.end_date, d.no_of_visits, d.sales_person)
			for i in range(d.no_of_visits):
				child = self.append("schedules")
				child.item_code = d.item_code
				child.item_name = d.item_name
				child.scheduled_date = s_list[i].strftime("%Y-%m-%d")
				if d.serial_no:
					child.serial_no = d.serial_no
				child.idx = count
				count = count + 1
				child.sales_person = d.sales_person
				child.completion_status = "Pending"
				child.item_reference = d.name

	@frappe.whitelist()
	def validate_end_date_visits(self):
		days_in_period = {"Weekly": 7, "Monthly": 30, "Quarterly": 91, "Half Yearly": 182, "Yearly": 365}
		for item in self.items:
			if item.periodicity and item.periodicity != "Random" and item.start_date:
				days = days_in_period[item.periodicity]
				if not item.end_date:
					item.end_date = add_days(item.start_date, (item.no_of_visits or 1) * days)

				no_of_visits = round((date_diff(item.end_date, item.start_date) + 1) / days)
				if not item.no_of_visits:
					item.no_of_visits = no_of_visits
				elif item.no_of_visits != no_of_visits:
					throw(
						_(
							"Row {0}: {1} {2} visits need an End Date of {3}. Change the End Date or the Number of Visits."
						).format(
							item.idx,
							item.no_of_visits,
							_(item.periodicity),
							formatdate(add_days(item.start_date, item.no_of_visits * days)),
						)
					)

	def on_submit(self):
		if not self.get("schedules"):
			throw(_("Please click on 'Generate Schedule' to get schedule"))
		self.check_serial_no_added()
		self.validate_schedule()

		email_map = {}
		for d in self.get("items"):
			if d.serial_and_batch_bundle:
				serial_nos = frappe.get_doc(
					"Serial and Batch Bundle", d.serial_and_batch_bundle
				).get_serial_nos()

				if serial_nos:
					self.validate_serial_no(d.item_code, serial_nos, d.start_date)
					self.update_amc_date(serial_nos, d.end_date)

			no_email_sp = []
			if d.sales_person and d.sales_person not in email_map:
				sp = frappe.get_doc("Sales Person", d.sales_person)
				try:
					email_map[d.sales_person] = sp.get_email_id()
				except frappe.ValidationError:
					no_email_sp.append(d.sales_person)

			if no_email_sp:
				frappe.msgprint(
					_(
						"Setting Events to {0}, since the Employee attached to the below Sales Persons does not have a User ID{1}"
					).format(self.owner, "<br>" + "<br>".join(no_email_sp))
				)

			scheduled_date = frappe.db.get_all(
				"Maintenance Schedule Detail",
				{"parent": self.name, "item_reference": d.name},
				["scheduled_date"],
				as_list=False,
			)

			for key in scheduled_date:
				description = frappe._("Reference: {0}, Item Code: {1} and Customer: {2}").format(
					self.name, d.item_code, self.customer
				)
				event = frappe.get_doc(
					{
						"doctype": "Event",
						"subject": description,
						"description": description,
						"starts_on": cstr(key["scheduled_date"]) + " 10:00:00",
						"event_type": "Private",
					}
				)
				event.add_participant(self.doctype, self.name)
				if email_map.get(d.sales_person):
					event.append(
						"event_participants",
						{
							"reference_doctype": "Sales Person",
							"reference_docname": d.sales_person,
							"email": email_map[d.sales_person],
						},
					)
				event.insert(ignore_permissions=1)

		self.db_set("status", "Submitted")

	def create_schedule_list(self, start_date, end_date, no_of_visit, sales_person):
		schedule_list = []
		start_date_copy = start_date
		date_diff = (getdate(end_date) - getdate(start_date)).days
		add_by = date_diff / no_of_visit

		for _visit in range(cint(no_of_visit)):
			if getdate(start_date_copy) < getdate(end_date):
				start_date_copy = add_days(start_date_copy, add_by)
				if len(schedule_list) < no_of_visit:
					schedule_date = self.validate_schedule_date_for_holiday_list(
						getdate(start_date_copy), sales_person, start_date
					)
					if schedule_date > getdate(end_date):
						if schedule_date != getdate(start_date_copy):
							throw(
								_("Every day from {0} to {1} is a holiday for Sales Person {2}.").format(
									formatdate(start_date), formatdate(end_date), sales_person
								)
							)
						schedule_date = getdate(end_date)
					schedule_list.append(schedule_date)

		return schedule_list

	def validate_schedule_date_for_holiday_list(self, schedule_date, sales_person, start_date):
		employee = frappe.db.get_value("Sales Person", sales_person, "employee")
		if employee:
			holiday_list = get_holiday_list_for_employee(employee, raise_exception=False)
		else:
			holiday_list = frappe.get_cached_value("Company", self.company, "default_holiday_list")

		holidays = frappe.db.sql_list(
			"""select holiday_date from `tabHoliday` where parent=%s""", holiday_list
		)

		working_date = schedule_date
		while working_date in holidays:
			working_date = add_days(working_date, -1)

		if working_date < getdate(start_date):
			working_date = schedule_date
			while working_date in holidays:
				working_date = add_days(working_date, 1)

		return working_date

	def validate_maintenance_detail(self):
		if not self.get("items"):
			throw(_("Please enter Maintenance Details first"))

		for d in self.get("items"):
			if not d.item_code:
				throw(_("Please select item code"))
			elif not d.start_date or not d.end_date:
				throw(_("Please select Start Date and End Date for Item {0}").format(d.item_code))
			elif not d.no_of_visits:
				throw(_("Please mention no of visits required"))

			if getdate(d.start_date) >= getdate(d.end_date):
				throw(_("Start date should be less than end date for Item {0}").format(d.item_code))

			if d.no_of_visits > date_diff(d.end_date, d.start_date):
				throw(
					_(
						"Row {0}: Number of visits can not be more than the {1} days between Start and End Date"
					).format(d.idx, date_diff(d.end_date, d.start_date))
				)

	def validate_sales_order(self):
		for d in self.get("items"):
			if d.sales_order:
				chk = frappe.db.sql(
					"""select ms.name from `tabMaintenance Schedule` ms,
					`tabMaintenance Schedule Item` msi where msi.parent=ms.name and
					msi.sales_order=%s and ms.docstatus=1""",
					d.sales_order,
				)
				if chk:
					throw(_("Maintenance Schedule {0} exists against {1}").format(chk[0][0], d.sales_order))

	def validate_items_table_change(self):
		doc_before_save = self.get_doc_before_save()
		if not doc_before_save:
			return
		for prev_item, item in zip(doc_before_save.items, self.items, strict=False):
			fields = [
				"item_code",
				"start_date",
				"end_date",
				"periodicity",
				"sales_person",
				"no_of_visits",
				"serial_no",
			]
			for field in fields:
				b_doc = prev_item.as_dict()
				doc = item.as_dict()
				if cstr(b_doc[field]) != cstr(doc[field]):
					return True

	def validate_no_of_visits(self):
		return len(self.schedules) != sum(d.no_of_visits for d in self.items)

	def validate(self):
		self.validate_end_date_visits()
		self.validate_maintenance_detail()
		self.validate_sales_order()
		self.validate_serial_no_bundle()
		if not self.schedules or self.validate_items_table_change() or self.validate_no_of_visits():
			self.generate_schedule()

	def validate_serial_no_bundle(self):
		ids = [d.serial_and_batch_bundle for d in self.items if d.serial_and_batch_bundle]

		if not ids:
			return

		voucher_nos = frappe.get_all(
			"Serial and Batch Bundle", fields=["name", "voucher_type"], filters={"name": ("in", ids)}
		)

		for row in voucher_nos:
			if row.voucher_type != "Maintenance Schedule":
				msg = f"""Serial and Batch Bundle {row.name}
					should have voucher type as 'Maintenance Schedule'"""

				frappe.throw(_(msg))

	def on_update(self):
		self.db_set("status", "Draft")

	def update_amc_date(self, serial_nos, amc_expiry_date=None):
		for serial_no in serial_nos:
			serial_no_doc = frappe.get_doc("Serial No", serial_no)
			serial_no_doc.amc_expiry_date = amc_expiry_date
			serial_no_doc.save(ignore_permissions=True)

	def validate_serial_no(self, item_code, serial_nos, amc_start_date):
		delivery_dates = self.get_delivery_dates(item_code, serial_nos)
		for serial_no in serial_nos:
			sr_details = frappe.db.get_value(
				"Serial No",
				serial_no,
				[
					"serial_no",
					"warranty_expiry_date",
					"amc_expiry_date",
					"warehouse",
					"item_code",
					"customer",
				],
				as_dict=1,
			)

			if not sr_details:
				frappe.throw(_("Serial No {0} not found").format(serial_no))

			if sr_details.get("item_code") != item_code:
				frappe.throw(
					_("Serial No {0} does not belong to Item {1}").format(
						frappe.bold(serial_no), frappe.bold(item_code)
					),
					title=_("Invalid"),
				)

			if sr_details.warranty_expiry_date and getdate(sr_details.warranty_expiry_date) >= getdate(
				amc_start_date
			):
				throw(
					_("Serial No {0} is under warranty upto {1}").format(
						serial_no, sr_details.warranty_expiry_date
					)
				)

			if sr_details.amc_expiry_date and getdate(sr_details.amc_expiry_date) >= getdate(amc_start_date):
				throw(
					_("Serial No {0} is under maintenance contract upto {1}").format(
						serial_no, sr_details.amc_expiry_date
					)
				)

			validate_serial_no_for_customer(sr_details, self.customer)

			delivery_date = delivery_dates.get(serial_no)
			if delivery_date and getdate(delivery_date) >= getdate(amc_start_date):
				throw(
					_("Maintenance start date can not be before delivery date for Serial No {0}").format(
						serial_no
					)
				)

	def get_delivery_dates(self, item_code, serial_nos):
		"""Latest Delivery Note or Sales Invoice posting per serial, read once for all serials."""
		bundle = frappe.qb.DocType("Serial and Batch Bundle")
		entry = frappe.qb.DocType("Serial and Batch Entry")
		rows = (
			frappe.qb.from_(bundle)
			.inner_join(entry)
			.on(entry.parent == bundle.name)
			.select(entry.serial_no, Max(bundle.posting_datetime).as_("posting_datetime"))
			.where(
				entry.serial_no.isin(serial_nos)
				& (bundle.item_code == item_code)
				& (bundle.docstatus == 1)
				& (bundle.is_cancelled == 0)
				& (bundle.type_of_transaction == "Outward")
				& bundle.voucher_type.isin(["Delivery Note", "Sales Invoice"])
			)
			.groupby(entry.serial_no)
		).run(as_dict=True)

		dates = {row.serial_no: row.posting_datetime for row in rows}
		if missing := [serial_no for serial_no in serial_nos if serial_no not in dates]:
			dates.update(get_legacy_delivery_dates(item_code, missing))
		return dates

	def validate_schedule(self):
		item_lst1 = []
		item_lst2 = []
		for d in self.get("items"):
			if d.item_code not in item_lst1:
				item_lst1.append(d.item_code)

		for m in self.get("schedules"):
			if m.item_code not in item_lst2:
				item_lst2.append(m.item_code)

		if len(item_lst1) != len(item_lst2):
			throw(
				_(
					"Maintenance Schedule is not generated for all the items. Please click on 'Generate Schedule'"
				)
			)
		else:
			for x in item_lst1:
				if x not in item_lst2:
					throw(_("Please click on 'Generate Schedule'"))

	def check_serial_no_added(self):
		serial_present = []
		for d in self.get("items"):
			if d.serial_no:
				serial_present.append(d.item_code)

		for m in self.get("schedules"):
			if serial_present:
				if m.item_code in serial_present and not m.serial_no:
					throw(
						_("Please click on 'Generate Schedule' to fetch Serial No added for Item {0}").format(
							m.item_code
						)
					)

	def on_cancel(self):
		for d in self.get("items"):
			if d.serial_and_batch_bundle:
				serial_nos = frappe.get_doc(
					"Serial and Batch Bundle", d.serial_and_batch_bundle
				).get_serial_nos()

				if serial_nos:
					expiry_dates = self.get_other_amc_expiry_dates(d.item_code, serial_nos)
					for serial_no in serial_nos:
						self.update_amc_date([serial_no], expiry_dates.get(serial_no))

		self.db_set("status", "Cancelled")
		delete_events(self.doctype, self.name)

	def get_other_amc_expiry_dates(self, item_code, serial_nos):
		"""Latest end date per serial across the item's other submitted schedules."""
		rows = frappe.get_all(
			"Maintenance Schedule Item",
			filters={
				"item_code": item_code,
				"docstatus": 1,
				"parent": ("!=", self.name),
				"serial_and_batch_bundle": ("is", "set"),
			},
			fields=["serial_and_batch_bundle", "end_date"],
		)
		if not rows:
			return {}

		bundle_serial_nos = get_serial_nos_from_sle_list([row.serial_and_batch_bundle for row in rows])
		expiry_dates = {}
		for row in rows:
			for serial_no in set(bundle_serial_nos.get(row.serial_and_batch_bundle, [])).intersection(
				serial_nos
			):
				expiry_dates[serial_no] = max(row.end_date, expiry_dates.get(serial_no, row.end_date))
		return expiry_dates

	def on_trash(self):
		delete_events(self.doctype, self.name)

	@frappe.whitelist()
	def get_pending_data(self, data_type, s_date=None, item_name=None):
		if data_type == "date":
			dates = ""
			for schedule in self.schedules:
				if schedule.item_name == item_name and schedule.completion_status == "Pending":
					dates = dates + "\n" + formatdate(schedule.scheduled_date, "dd-MM-yyyy")
			return dates
		elif data_type == "items":
			items = ""
			for item in self.items:
				for schedule in self.schedules:
					if item.item_name == schedule.item_name and schedule.completion_status == "Pending":
						items = items + "\n" + item.item_name
						break
			return items
		elif data_type == "id":
			for schedule in self.schedules:
				if (
					schedule.item_name == item_name
					and schedule.completion_status == "Pending"
					and s_date == formatdate(schedule.scheduled_date, "dd-mm-yyyy")
				):
					return schedule.name


def validate_serial_no_for_customer(serial, customer):
	"""Refuse a serial still in stock; warn when it was sold to another customer."""
	number = frappe.bold(escape_html(serial.serial_no))
	if serial.warehouse:
		frappe.throw(
			_("Serial No {0} is still in stock in Warehouse {1}").format(
				number, frappe.bold(serial.warehouse)
			)
		)
	if serial.customer and serial.customer != customer:
		frappe.msgprint(
			_("Serial No {0} was sold to Customer {1}, not {2}").format(
				number, frappe.bold(escape_html(serial.customer)), frappe.bold(escape_html(customer))
			),
			title=_("Different Customer"),
			indicator="orange",
		)


def get_legacy_delivery_dates(item_code: str, serial_nos: list[str]) -> dict:
	"""Latest delivery per serial kept on the Stock Ledger Entry itself, from before Serial and Batch Bundles."""
	ledger = frappe.qb.DocType("Stock Ledger Entry")
	rows = (
		frappe.qb.from_(ledger)
		.select(ledger.serial_no, ledger.posting_datetime)
		.where(
			(ledger.item_code == item_code)
			& (ledger.is_cancelled == 0)
			& (ledger.actual_qty < 0)
			& ledger.voucher_type.isin(["Delivery Note", "Sales Invoice"])
			& Criterion.any([ledger.serial_no.like(f"%{serial_no}%") for serial_no in serial_nos])
		)
	).run(as_dict=True)

	dates = {}
	for row in rows:
		for serial_no in set(get_serial_nos(row.serial_no)).intersection(serial_nos):
			dates[serial_no] = max(dates.get(serial_no, row.posting_datetime), row.posting_datetime)
	return dates


@frappe.whitelist()
def get_serial_nos_from_schedule(item_code, schedule=None):
	serial_nos = []
	if schedule:
		frappe.has_permission("Maintenance Schedule", "read", doc=schedule, throw=True)
		serial_nos = frappe.db.get_value(
			"Maintenance Schedule Item", {"parent": schedule, "item_code": item_code}, "serial_no"
		)

	if serial_nos:
		serial_nos = get_serial_nos(serial_nos)

	return serial_nos


@frappe.whitelist()
def make_maintenance_visit(source_name, target_doc=None, item_name=None, s_id=None):
	from frappe.model.mapper import get_mapped_doc

	def condition(doc):
		if s_id:
			return doc.name == s_id
		if item_name and doc.item_name != item_name:
			return False
		return doc.completion_status != "Fully Completed"

	def update_status_and_detail(source, target, parent):
		target.maintenance_type = "Scheduled"

	def update_serial(source, target, parent):
		if source.item_reference:
			if sbb := frappe.db.get_value(
				"Maintenance Schedule Item", source.item_reference, "serial_and_batch_bundle"
			):
				serial_nos = frappe.get_doc("Serial and Batch Bundle", sbb).get_serial_nos()

				if len(serial_nos) == 1:
					target.serial_no = serial_nos[0]
				else:
					target.serial_no = ""

	doclist = get_mapped_doc(
		"Maintenance Schedule",
		source_name,
		{
			"Maintenance Schedule": {
				"doctype": "Maintenance Visit",
				"field_map": {"name": "maintenance_schedule"},
				"validation": {"docstatus": ["=", 1]},
				"postprocess": update_status_and_detail,
			},
			"Maintenance Schedule Detail": {
				"doctype": "Maintenance Visit Purpose",
				"condition": condition,
				"field_map": {
					"sales_person": "service_person",
					"name": "maintenance_schedule_detail",
				},
				"postprocess": update_serial,
			},
		},
		target_doc,
	)

	return doclist