fix: set/apply sla and first response logic

crm/fcrm/doctype/crm_lead/crm_lead.json

@@ -220,9 +220,9 @@
    "options": "CRM Service Level Agreement"
   },
   {
-   "fieldname": "response_by",
+   "fieldname": "sla_creation",
    "fieldtype": "Datetime",
-   "label": "Response By"
+   "label": "SLA Creation"
   },
   {
    "fieldname": "column_break_ffnp",
@@ -234,16 +234,16 @@
    "label": "SLA Status",
    "options": "\nFirst Response Due\nFailed\nFulfilled"
   },
-  {
-   "fieldname": "sla_creation",
-   "fieldtype": "Datetime",
-   "label": "SLA Creation"
-  },
   {
    "fieldname": "response_details_section",
    "fieldtype": "Section Break",
    "label": "Response Details"
   },
+  {
+   "fieldname": "response_by",
+   "fieldtype": "Datetime",
+   "label": "Response By"
+  },
   {
    "fieldname": "column_break_pweh",
    "fieldtype": "Column Break"
@@ -262,7 +262,7 @@
  "image_field": "image",
  "index_web_pages_for_search": 1,
  "links": [],
- "modified": "2023-12-04 15:25:07.155734",
+ "modified": "2023-12-10 13:54:53.630114",
  "modified_by": "Administrator",
  "module": "FCRM",
  "name": "CRM Lead",

crm/fcrm/doctype/crm_lead/crm_lead.py

@@ -9,12 +9,18 @@ from frappe.utils import has_gravatar, validate_email_address
 
 
 class CRMLead(Document):
+	def before_validate(self):
+		self.set_sla()
+
 	def validate(self):
 		self.set_full_name()
 		self.set_lead_name()
 		self.set_title()
 		self.validate_email()
 
+	def before_save(self):
+		self.apply_sla()
+
 	def set_full_name(self):
 		if self.first_name:
 			self.lead_name = " ".join(
@@ -121,6 +127,22 @@ class CRMLead(Document):
 		deal.insert(ignore_permissions=True)
 		return deal.name
 
+	def set_sla(self):
+		"""
+		Find an SLA to apply to the lead.
+		"""
+		if sla := get_sla("CRM Lead"):
+			if not sla:
+				return
+			self.sla = sla.name
+
+	def apply_sla(self):
+		"""
+		Apply SLA if set.
+		"""
+		if sla := frappe.get_last_doc("CRM Service Level Agreement", {"name": self.sla}):
+			sla.apply(self)
+
 	@staticmethod
 	def sort_options():
 		return [
@@ -209,4 +231,10 @@ def convert_to_deal(lead):
 	contact = lead.create_contact(False)
 	deal = lead.create_deal(contact)
 	lead.save()
-	return deal
+	return deal
+
+def get_sla(doctype):
+	sla = frappe.db.exists("CRM Service Level Agreement", {"apply_on": doctype, "enabled": 1})
+	if not sla:
+		return None
+	return frappe.get_cached_doc("CRM Service Level Agreement", sla)

crm/fcrm/doctype/crm_service_day/crm_service_day.json

@@ -0,0 +1,59 @@
+{
+ "actions": [],
+ "allow_rename": 1,
+ "creation": "2023-12-04 16:07:20.400084",
+ "doctype": "DocType",
+ "editable_grid": 1,
+ "engine": "InnoDB",
+ "field_order": [
+  "workday",
+  "section_break_uegc",
+  "start_time",
+  "column_break_maie",
+  "end_time"
+ ],
+ "fields": [
+  {
+   "fieldname": "workday",
+   "fieldtype": "Select",
+   "in_list_view": 1,
+   "label": "Workday",
+   "options": "Monday\nTuesday\nWednesday\nThursday\nFriday\nSaturday\nSunday",
+   "reqd": 1
+  },
+  {
+   "fieldname": "section_break_uegc",
+   "fieldtype": "Section Break"
+  },
+  {
+   "fieldname": "start_time",
+   "fieldtype": "Time",
+   "in_list_view": 1,
+   "label": "Start Time",
+   "reqd": 1
+  },
+  {
+   "fieldname": "column_break_maie",
+   "fieldtype": "Column Break"
+  },
+  {
+   "fieldname": "end_time",
+   "fieldtype": "Time",
+   "in_list_view": 1,
+   "label": "End Time",
+   "reqd": 1
+  }
+ ],
+ "index_web_pages_for_search": 1,
+ "istable": 1,
+ "links": [],
+ "modified": "2023-12-04 16:09:22.928308",
+ "modified_by": "Administrator",
+ "module": "FCRM",
+ "name": "CRM Service Day",
+ "owner": "Administrator",
+ "permissions": [],
+ "sort_field": "modified",
+ "sort_order": "DESC",
+ "states": []
+}

crm/fcrm/doctype/crm_service_day/crm_service_day.py

@@ -0,0 +1,9 @@
+# Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and contributors
+# For license information, please see license.txt
+
+# import frappe
+from frappe.model.document import Document
+
+
+class CRMServiceDay(Document):
+	pass

crm/fcrm/doctype/crm_service_level_agreement/crm_service_level_agreement.json

@@ -13,7 +13,9 @@
   "column_break_uxua",
   "condition",
   "section_break_ufaf",
-  "priorities"
+  "priorities",
+  "section_break_rmgo",
+  "working_hours"
  ],
  "fields": [
   {
@@ -66,11 +68,22 @@
    "label": "Priorities",
    "options": "CRM Service Level Priority",
    "reqd": 1
+  },
+  {
+   "fieldname": "section_break_rmgo",
+   "fieldtype": "Section Break"
+  },
+  {
+   "fieldname": "working_hours",
+   "fieldtype": "Table",
+   "label": "Working Hours",
+   "options": "CRM Service Day",
+   "reqd": 1
   }
  ],
  "index_web_pages_for_search": 1,
  "links": [],
- "modified": "2023-12-04 13:49:09.547523",
+ "modified": "2023-12-04 16:13:24.638239",
  "modified_by": "Administrator",
  "module": "FCRM",
  "name": "CRM Service Level Agreement",

crm/fcrm/doctype/crm_service_level_agreement/crm_service_level_agreement.py

@@ -1,9 +1,159 @@
 # Copyright (c) 2023, Frappe Technologies Pvt. Ltd. and contributors
 # For license information, please see license.txt
+from typing import Literal
 
 # import frappe
 from frappe.model.document import Document
+from frappe.utils import (
+	add_to_date,
+	get_datetime,
+	get_weekdays,
+	getdate,
+	now_datetime,
+	time_diff_in_seconds,
+)
 
 
 class CRMServiceLevelAgreement(Document):
-	pass
+	def apply(self, doc: Document):
+		self.handle_new(doc)
+		self.handle_status(doc)
+		self.handle_targets(doc)
+		self.handle_sla_status(doc)
+
+	def handle_new(self, doc: Document):
+		if not doc.is_new():
+			return
+		creation = doc.sla_creation or now_datetime()
+		doc.sla_creation = creation
+
+	def handle_status(self, doc: Document):
+		if doc.is_new() or not doc.has_value_changed("status"):
+			return
+		self.set_first_response_time(doc)
+
+	def set_first_response_time(self, doc: Document):
+		start_at = doc.sla_creation
+		end_at = doc.first_responded_on
+		if not start_at or not end_at:
+			return
+		doc.first_response_time = self.calc_elapsed_time(start_at, end_at)
+
+	def handle_targets(self, doc: Document):
+		self.set_response_by(doc)
+
+	def set_response_by(self, doc: Document):
+		start = doc.sla_creation
+		doc.response_by = self.calc_time(start, doc.status, "first_response_time")
+
+	def handle_sla_status(self, doc: Document):
+		is_failed = self.is_first_response_failed(doc)
+		options = {
+			"Fulfilled": True,
+			"First Response Due": not doc.first_responded_on,
+			"Failed": is_failed,
+		}
+		for status in options:
+			if options[status]:
+				doc.sla_status = status
+
+	def is_first_response_failed(self, doc: Document):
+		if not doc.first_responded_on:
+			return get_datetime(doc.response_by) < now_datetime()
+		return get_datetime(doc.response_by) < get_datetime(doc.first_responded_on)
+
+	def calc_time(
+		self,
+		start_at: str,
+		priority: str,
+		target: Literal["first_response_time"],
+	):
+		res = get_datetime(start_at)
+		priority = self.get_priorities()[priority]
+		time_needed = priority.get(target, 0)
+		holidays = []
+		weekdays = get_weekdays()
+		workdays = self.get_workdays()
+		while time_needed:
+			today = res
+			today_day = getdate(today)
+			today_weekday = weekdays[today.weekday()]
+			is_workday = today_weekday in workdays
+			is_holiday = today_day in holidays
+			if is_holiday or not is_workday:
+				res = add_to_date(res, days=1, as_datetime=True)
+				continue
+			today_workday = workdays[today_weekday]
+			now_in_seconds = time_diff_in_seconds(today, today_day)
+			start_time = max(today_workday.start_time.total_seconds(), now_in_seconds)
+			till_start_time = max(start_time - now_in_seconds, 0)
+			end_time = max(today_workday.end_time.total_seconds(), now_in_seconds)
+			time_left = max(end_time - start_time, 0)
+			if not time_left:
+				res = getdate(add_to_date(res, days=1, as_datetime=True))
+				continue
+			time_taken = min(time_needed, time_left)
+			time_needed -= time_taken
+			time_required = till_start_time + time_taken
+			res = add_to_date(res, seconds=time_required, as_datetime=True)
+		return res
+
+
+	def calc_elapsed_time(self, start_at, end_at) -> float:
+		"""
+		Get took from start to end, excluding non-working hours
+
+		:param start_at: Date at which calculation starts
+		:param end_at: Date at which calculation ends
+		:return: Number of seconds
+		"""
+		start_at = getdate(start_at)
+		end_at = getdate(end_at)
+		time_took = 0
+		holidays = []
+		weekdays = get_weekdays()
+		workdays = self.get_workdays()
+		while getdate(start_at) <= getdate(end_at):
+			today = start_at
+			today_day = getdate(today)
+			today_weekday = weekdays[today.weekday()]
+			is_workday = today_weekday in workdays
+			is_holiday = today_day in holidays
+			if is_holiday or not is_workday:
+				start_at = getdate(add_to_date(start_at, days=1, as_datetime=True))
+				continue
+			today_workday = workdays[today_weekday]
+			is_today = getdate(start_at) == getdate(end_at)
+			if not is_today:
+				working_start = today_workday.start_time
+				working_end = today_workday.end_time
+				working_time = time_diff_in_seconds(working_start, working_end)
+				time_took += working_time
+				start_at = getdate(add_to_date(start_at, days=1, as_datetime=True))
+				continue
+			now_in_seconds = time_diff_in_seconds(today, today_day)
+			start_time = max(today_workday.start_time.total_seconds(), now_in_seconds)
+			end_at_seconds = time_diff_in_seconds(getdate(end_at), end_at)
+			end_time = max(today_workday.end_time.total_seconds(), end_at_seconds)
+			time_taken = end_time - start_time
+			time_took += time_taken
+			start_at = getdate(add_to_date(start_at, days=1, as_datetime=True))
+		return time_took
+
+	def get_priorities(self):
+		"""
+		Return priorities related info as a dict. With `priority` as key
+		"""
+		res = {}
+		for row in self.priorities:
+			res[row.priority] = row
+		return res
+
+	def get_workdays(self) -> dict[str, dict]:
+		"""
+		Return workdays related info as a dict. With `workday` as key
+		"""
+		res = {}
+		for row in self.working_hours:
+			res[row.workday] = row
+		return res