refactor: application of SLA and its metrics
diff --git a/erpnext/hooks.py b/erpnext/hooks.py
index 2a277ee..186dbb5 100644
--- a/erpnext/hooks.py
+++ b/erpnext/hooks.py
@@ -341,8 +341,7 @@
"erpnext.erpnext_integrations.doctype.plaid_settings.plaid_settings.automatic_synchronization",
"erpnext.projects.doctype.project.project.hourly_reminder",
"erpnext.projects.doctype.project.project.collect_project_status",
- "erpnext.hr.doctype.shift_type.shift_type.process_auto_attendance_for_all_shifts",
- "erpnext.support.doctype.service_level_agreement.service_level_agreement.set_service_level_agreement_variance"
+ "erpnext.hr.doctype.shift_type.shift_type.process_auto_attendance_for_all_shifts"
],
"hourly_long": [
"erpnext.stock.doctype.repost_item_valuation.repost_item_valuation.repost_entries"
diff --git a/erpnext/support/doctype/service_level_agreement/service_level_agreement.py b/erpnext/support/doctype/service_level_agreement/service_level_agreement.py
index a25608b..8646402 100644
--- a/erpnext/support/doctype/service_level_agreement/service_level_agreement.py
+++ b/erpnext/support/doctype/service_level_agreement/service_level_agreement.py
@@ -10,7 +10,6 @@
from frappe.model.document import Document
from frappe.utils import (
add_to_date,
- cint,
get_datetime,
get_datetime_str,
get_link_to_form,
@@ -350,86 +349,113 @@
def apply(doc, method=None):
# Applies SLA to document on validate
- if frappe.flags.in_patch or frappe.flags.in_migrate or frappe.flags.in_install or frappe.flags.in_setup_wizard or \
- doc.doctype not in get_documents_with_active_service_level_agreement():
+ if (
+ frappe.flags.in_patch
+ or frappe.flags.in_migrate
+ or frappe.flags.in_install
+ or frappe.flags.in_setup_wizard
+ or doc.doctype not in get_documents_with_active_service_level_agreement()
+ ):
return
service_level_agreement = get_active_service_level_agreement_for(doc)
- print(service_level_agreement)
-
if not service_level_agreement:
return
- set_sla_properties(doc, service_level_agreement)
+ process_sla(doc, service_level_agreement)
-def set_sla_properties(doc, service_level_agreement):
- if frappe.db.exists(doc.doctype, doc.name):
- from_db = frappe.get_doc(doc.doctype, doc.name)
- else:
- from_db = frappe._dict({})
-
- meta = frappe.get_meta(doc.doctype)
-
- if meta.has_field("customer") and service_level_agreement.customer and doc.get("customer") and \
- not service_level_agreement.customer == doc.get("customer"):
- frappe.throw(_("Service Level Agreement {0} is specific to Customer {1}").format(service_level_agreement.name,
- service_level_agreement.customer))
-
- doc.service_level_agreement = service_level_agreement.name
- doc.priority = doc.get("priority") or service_level_agreement.default_priority
- priority = get_priority(doc)
+def process_sla(doc, service_level_agreement):
if not doc.creation:
doc.creation = now_datetime(doc.get("owner"))
-
- if meta.has_field("service_level_agreement_creation"):
+ if doc.meta.has_field("service_level_agreement_creation"):
doc.service_level_agreement_creation = now_datetime(doc.get("owner"))
+ doc.service_level_agreement = service_level_agreement.name
+ doc.priority = doc.get("priority") or service_level_agreement.default_priority
+
+ prev_status = frappe.db.get_value(doc.doctype, doc.name, 'status')
+ handle_status_change(doc, prev_status, service_level_agreement.apply_sla_for_resolution)
+ update_response_and_resolution_metrics(doc, service_level_agreement.apply_sla_for_resolution)
+ update_agreement_status(doc, service_level_agreement.apply_sla_for_resolution)
+
+
+def update_response_and_resolution_metrics(doc, apply_sla_for_resolution):
+ priority = get_response_and_resolution_duration(doc)
start_date_time = get_datetime(doc.get("service_level_agreement_creation") or doc.creation)
-
- set_response_by_and_variance(doc, meta, start_date_time, priority)
- if service_level_agreement.apply_sla_for_resolution:
- set_resolution_by_and_variance(doc, meta, start_date_time, priority)
-
- update_status(doc, from_db, meta)
+ set_response_by_and_variance(doc, start_date_time, priority)
+ if apply_sla_for_resolution:
+ set_resolution_by_and_variance(doc, start_date_time, priority)
-def update_status(doc, from_db, meta):
- if meta.has_field("status"):
- if meta.has_field("first_responded_on") and doc.status != "Open" and \
- from_db.status == "Open" and not doc.first_responded_on:
+def get_fulfillment_statuses(service_level_agreement):
+ return [entry.status for entry in frappe.db.get_all("SLA Fulfilled On Status", filters={
+ "parent": service_level_agreement
+ }, fields=["status"])]
+
+
+def get_hold_statuses(service_level_agreement):
+ return [entry.status for entry in frappe.db.get_all("Pause SLA On Status", filters={
+ "parent": service_level_agreement
+ }, fields=["status"])]
+
+
+def handle_status_change(doc, prev_status, apply_sla_for_resolution):
+
+ if doc.status != "Open" and prev_status == "Open":
+ # status changed from Open to something else
+ if doc.meta.has_field("first_responded_on") and not doc.first_responded_on:
+ # status changed to something other than Open
doc.first_responded_on = frappe.flags.current_time or now_datetime(doc.get("owner"))
- if meta.has_field("service_level_agreement") and doc.service_level_agreement:
- # mark sla status as fulfilled based on the configuration
- fulfillment_statuses = [entry.status for entry in frappe.db.get_all("SLA Fulfilled On Status", filters={
- "parent": doc.service_level_agreement
- }, fields=["status"])]
+ if doc.status == "Open" and prev_status != "Open":
+ # status changed from something else to Open
+ reset_resolution_metrics(doc)
- if doc.status in fulfillment_statuses and from_db.status not in fulfillment_statuses:
- apply_sla_for_resolution = frappe.db.get_value("Service Level Agreement", doc.service_level_agreement,
- "apply_sla_for_resolution")
+ handle_fulfillment_status(doc, prev_status, apply_sla_for_resolution)
+ handle_hold_status(doc, prev_status)
- if apply_sla_for_resolution and meta.has_field("resolution_date"):
- doc.resolution_date = frappe.flags.current_time or now_datetime(doc.get("owner"))
- if meta.has_field("agreement_status") and from_db.agreement_status == "Ongoing":
- set_service_level_agreement_variance(doc.doctype, doc.name)
- update_agreement_status(doc, meta)
+def handle_fulfillment_status(doc, prev_status, apply_sla_for_resolution):
+ fulfillment_statuses = get_fulfillment_statuses(doc.service_level_agreement)
+ if (
+ doc.status in fulfillment_statuses
+ and prev_status not in fulfillment_statuses
+ and apply_sla_for_resolution
+ ):
+ # status changed to any fulfillment_statuses
+ if doc.meta.has_field("resolution_date"):
+ doc.resolution_date = frappe.flags.current_time or now_datetime(doc.get("owner"))
+ if doc.meta.has_field("resolution_time"):
+ doc.resolution_time = time_diff_in_seconds(doc.resolution_date, doc.creation)
+ set_user_resolution_time(doc)
- if apply_sla_for_resolution:
- set_resolution_time(doc, meta)
- set_user_resolution_time(doc, meta)
- if doc.status == "Open" and from_db.status != "Open":
- # if no date, it should be set as None and not a blank string "", as per mysql strict config
- # enable SLA and variance on Reopen
- reset_metrics(doc, meta)
- set_service_level_agreement_variance(doc.doctype, doc.name)
+def handle_hold_status(doc, prev_status):
+ hold_statuses = get_hold_statuses(doc.service_level_agreement)
+ if doc.status in hold_statuses:
+ # reset if status is a hold status, regardless of previous status
+ reset_expected_response_and_resolution(doc)
+ if prev_status not in hold_statuses:
+ # set on_hold_since status changed from any non-hold status
+ # for eg. doc.status changed from Open to Replied
+ if doc.meta.has_field("on_hold_since"):
+ doc.on_hold_since = frappe.flags.current_time or now_datetime(doc.get("owner"))
- handle_hold_time(doc, meta, from_db.status)
+ if doc.status not in hold_statuses and prev_status in hold_statuses:
+ # status changed to any non-hold status
+ # for eg. doc.status changed from Replied to Closed
+ if doc.meta.has_field("on_hold_since") and doc.on_hold_since:
+ cumulate_hold_time(doc)
+ doc.on_hold_since = None
+
+
+def cumulate_hold_time(doc):
+ now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
+ on_hold_duration = time_diff_in_seconds(now_time, doc.on_hold_since)
+ doc.total_hold_time = (doc.total_hold_time or 0) + on_hold_duration
def get_expected_time_for(parameter, service_level, start_date_time):
@@ -500,41 +526,9 @@
return support_days
-def set_service_level_agreement_variance(doctype, doc=None):
-
- filters = {"status": "Open", "agreement_status": "Ongoing"}
-
- if doc:
- filters = {"name": doc}
-
- for entry in frappe.get_all(doctype, filters=filters):
- current_doc = frappe.get_doc(doctype, entry.name)
- current_time = frappe.flags.current_time or now_datetime(current_doc.get("owner"))
- apply_sla_for_resolution = frappe.db.get_value("Service Level Agreement", current_doc.service_level_agreement,
- "apply_sla_for_resolution")
-
- if not current_doc.first_responded_on: # first_responded_on set when first reply is sent to customer
- variance = round(time_diff_in_seconds(current_doc.response_by, current_time), 2)
- else:
- variance = round(time_diff_in_seconds(current_doc.response_by, current_doc.first_responded_on), 2)
-
- frappe.db.set_value(current_doc.doctype, current_doc.name, "response_by_variance", variance, update_modified=False)
- if variance < 0:
- frappe.db.set_value(current_doc.doctype, current_doc.name, "agreement_status", "Failed", update_modified=False)
-
- if apply_sla_for_resolution and not current_doc.get("resolution_date"): # resolution_date set when issue has been closed
- variance = round(time_diff_in_seconds(current_doc.resolution_by, current_time), 2)
- elif apply_sla_for_resolution and current_doc.get("resolution_date"):
- variance = round(time_diff_in_seconds(current_doc.resolution_by, current_doc.get("resolution_date")), 2)
-
- frappe.db.set_value(current_doc.doctype, current_doc.name, "resolution_by_variance", variance, update_modified=False)
- if variance < 0:
- frappe.db.set_value(current_doc.doctype, current_doc.name, "agreement_status", "Failed", update_modified=False)
-
-
-def set_user_resolution_time(doc, meta):
+def set_user_resolution_time(doc):
# total time taken by a user to close the issue apart from wait_time
- if not meta.has_field("user_resolution_time"):
+ if not doc.meta.has_field("user_resolution_time"):
return
communications = frappe.get_all("Communication", filters={
@@ -567,7 +561,7 @@
frappe.msgprint(_("Service Level Agreement has been changed to {0}.").format(self.service_level_agreement))
-def get_priority(doc):
+def get_response_and_resolution_duration(doc):
service_level_agreement = frappe.get_doc("Service Level Agreement", doc.service_level_agreement)
priority = service_level_agreement.get_service_level_agreement_priority(doc.priority)
priority.update({
@@ -596,115 +590,81 @@
doc.save()
-def reset_metrics(doc, meta):
- if meta.has_field("resolution_date"):
+def reset_resolution_metrics(doc):
+ if doc.meta.has_field("resolution_date"):
doc.resolution_date = None
- if not meta.has_field("resolution_time"):
+ if doc.meta.has_field("resolution_time"):
doc.resolution_time = None
- if not meta.has_field("user_resolution_time"):
+ if doc.meta.has_field("user_resolution_time"):
doc.user_resolution_time = None
- if meta.has_field("agreement_status"):
+ if doc.meta.has_field("agreement_status"):
doc.agreement_status = "Ongoing"
-def set_resolution_time(doc, meta):
- # total time taken from issue creation to closing
- if not meta.has_field("resolution_time"):
- return
-
- doc.resolution_time = time_diff_in_seconds(doc.resolution_date, doc.creation)
-
-
# called via hooks on communication update
def update_hold_time(doc, status):
+ if doc.communication_type == "Comment" or doc.sent_or_received != "Received":
+ return
+
parent = get_parent_doc(doc)
if not parent:
return
- if doc.communication_type == "Comment":
+ if not parent.meta.has_field('service_level_agreement'):
return
- status_field = parent.meta.get_field("status")
- if status_field:
- options = (status_field.options or "").splitlines()
+ apply_sla_for_resolution = frappe.db.get_value('Service Level Agreement', parent.service_level_agreement, 'apply_sla_for_resolution')
- # if status has a "Replied" option, then handle hold time
- if ("Replied" in options) and doc.sent_or_received == "Received":
- meta = frappe.get_meta(parent.doctype)
- handle_hold_time(parent, meta, 'Replied')
+ handle_status_change(parent, 'Replied', apply_sla_for_resolution)
+ update_response_and_resolution_metrics(parent, apply_sla_for_resolution)
+ update_agreement_status(parent, apply_sla_for_resolution)
+
+ parent.save()
-def handle_hold_time(doc, meta, status):
- if meta.has_field("service_level_agreement") and doc.service_level_agreement:
- # set response and resolution variance as None as the issue is on Hold for status as Replied
- hold_statuses = [entry.status for entry in frappe.db.get_all("Pause SLA On Status", filters={
- "parent": doc.service_level_agreement
- }, fields=["status"])]
+def reset_expected_response_and_resolution(doc):
+ update_values = {}
- if not hold_statuses:
- return
-
- if meta.has_field("status") and doc.status in hold_statuses and status not in hold_statuses:
- apply_hold_status(doc, meta)
-
- # calculate hold time when status is changed from any hold status to any non-hold status
- if meta.has_field("status") and doc.status not in hold_statuses and status in hold_statuses:
- reset_hold_status_and_update_hold_time(doc, meta)
-
-
-def apply_hold_status(doc, meta):
- update_values = {'on_hold_since': frappe.flags.current_time or now_datetime(doc.get("owner"))}
-
- if meta.has_field("first_responded_on") and not doc.first_responded_on:
+ if doc.meta.has_field("first_responded_on") and not doc.first_responded_on:
update_values['response_by'] = None
update_values['response_by_variance'] = 0
- update_values['resolution_by'] = None
- update_values['resolution_by_variance'] = 0
+ if doc.meta.has_field("resolution_by") and not doc.resolution_date:
+ update_values['resolution_by'] = None
+ update_values['resolution_by_variance'] = 0
doc.db_set(update_values)
-def reset_hold_status_and_update_hold_time(doc, meta):
- hold_time = doc.total_hold_time if meta.has_field("total_hold_time") and doc.total_hold_time else 0
- now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
- last_hold_time = 0
- update_values = {}
+def set_response_by_and_variance(doc, start_date_time, priority):
+ if doc.meta.has_field("response_by"):
+ doc.response_by = get_expected_time_for(parameter="response", service_level=priority, start_date_time=start_date_time)
+ if doc.meta.has_field("total_hold_time") and doc.total_hold_time:
+ doc.response_by = add_to_date(doc.response_by, seconds=round(doc.total_hold_time))
- if meta.has_field("on_hold_since") and doc.on_hold_since:
- # last_hold_time will be added to the sla variables
- last_hold_time = time_diff_in_seconds(now_time, doc.on_hold_since)
- update_values['total_hold_time'] = hold_time + last_hold_time
+ if doc.meta.has_field("response_by_variance") and not doc.get('first_responded_on'):
+ now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
+ doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, now_time), 2)
- # re-calculate SLA variables after issue changes from any hold status to any non-hold status
- start_date_time = get_datetime(doc.get("service_level_agreement_creation") or doc.creation)
- priority = get_priority(doc)
- now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
+ if doc.meta.has_field("response_by_variance") and doc.get('first_responded_on'):
+ doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, doc.get('first_responded_on')), 2)
- # add hold time to response by variance
- if meta.has_field("first_responded_on") and not doc.first_responded_on:
- response_by = get_expected_time_for(parameter="response", service_level=priority, start_date_time=start_date_time)
- response_by = add_to_date(response_by, seconds=round(last_hold_time))
- response_by_variance = round(time_diff_in_seconds(response_by, now_time))
- update_values['response_by'] = response_by
- update_values['response_by_variance'] = response_by_variance + last_hold_time
+def set_resolution_by_and_variance(doc, start_date_time, priority):
+ if doc.meta.has_field("resolution_by"):
+ doc.resolution_by = get_expected_time_for(parameter="resolution", service_level=priority, start_date_time=start_date_time)
+ if doc.meta.has_field("total_hold_time") and doc.total_hold_time:
+ doc.resolution_by = add_to_date(doc.resolution_by, seconds=round(doc.total_hold_time))
- # add hold time to resolution by variance
- if frappe.db.get_value("Service Level Agreement", doc.service_level_agreement, "apply_sla_for_resolution"):
- resolution_by = get_expected_time_for(parameter="resolution", service_level=priority, start_date_time=start_date_time)
- resolution_by = add_to_date(resolution_by, seconds=round(last_hold_time))
- resolution_by_variance = round(time_diff_in_seconds(resolution_by, now_time))
+ if doc.meta.has_field("resolution_by_variance") and not doc.get("resolution_date"):
+ now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
+ doc.resolution_by_variance = round(time_diff_in_seconds(doc.resolution_by, now_time), 2)
- update_values['resolution_by'] = resolution_by
- update_values['resolution_by_variance'] = resolution_by_variance + last_hold_time
-
- update_values['on_hold_since'] = None
-
- doc.db_set(update_values)
+ if doc.meta.has_field("resolution_by_variance") and doc.get('resolution_date'):
+ doc.resolution_by_variance = round(time_diff_in_seconds(doc.resolution_by, doc.get('resolution_date')), 2)
def get_service_level_agreement_fields():
@@ -808,45 +768,37 @@
meta = frappe.get_meta(doc.doctype)
now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
- if meta.has_field("first_responded_on") and not doc.first_responded_on:
+ if doc.meta.has_field("first_responded_on") and not doc.first_responded_on:
# first_responded_on set when first reply is sent to customer
doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, now_time), 2)
- if meta.has_field("first_responded_on") and doc.first_responded_on:
+ if doc.meta.has_field("first_responded_on") and doc.first_responded_on:
# first_responded_on set when first reply is sent to customer
doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, doc.first_responded_on), 2)
- if meta.has_field("resolution_date") and not doc.resolution_date:
+ if doc.meta.has_field("resolution_date") and not doc.resolution_date:
# resolution_date set when issue has been closed
doc.resolution_by_variance = round(time_diff_in_seconds(doc.resolution_by, now_time), 2)
- if meta.has_field("resolution_date") and doc.resolution_date:
+ if doc.meta.has_field("resolution_date") and doc.resolution_date:
# resolution_date set when issue has been closed
doc.resolution_by_variance = round(time_diff_in_seconds(doc.resolution_by, doc.resolution_date), 2)
- if meta.has_field("agreement_status"):
+ if doc.meta.has_field("agreement_status"):
doc.agreement_status = "Fulfilled" if doc.response_by_variance > 0 and doc.resolution_by_variance > 0 else "Failed"
-def update_agreement_status(doc, meta):
- if meta.has_field("service_level_agreement") and meta.has_field("agreement_status") and \
- doc.service_level_agreement and doc.agreement_status == "Ongoing":
-
- apply_sla_for_resolution = frappe.db.get_value("Service Level Agreement", doc.service_level_agreement,
- "apply_sla_for_resolution")
-
+def update_agreement_status(doc, apply_sla_for_resolution):
+ if (doc.meta.has_field("agreement_status")):
# if SLA is applied for resolution check for response and resolution, else only response
if apply_sla_for_resolution:
- if meta.has_field("response_by_variance") and meta.has_field("resolution_by_variance"):
- if cint(frappe.db.get_value(doc.doctype, doc.name, "response_by_variance")) < 0 or \
- cint(frappe.db.get_value(doc.doctype, doc.name, "resolution_by_variance")) < 0:
-
+ if doc.meta.has_field("response_by_variance") and doc.meta.has_field("resolution_by_variance"):
+ if doc.response_by_variance < 0 or doc.resolution_by_variance < 0:
doc.agreement_status = "Failed"
else:
doc.agreement_status = "Fulfilled"
else:
- if meta.has_field("response_by_variance") and \
- cint(frappe.db.get_value(doc.doctype, doc.name, "response_by_variance")) < 0:
+ if doc.meta.has_field("response_by_variance") and doc.response_by_variance < 0:
doc.agreement_status = "Failed"
else:
doc.agreement_status = "Fulfilled"
@@ -862,25 +814,6 @@
return datetime.timedelta(hours=time.hour, minutes=time.minute, seconds=time.second)
-def set_response_by_and_variance(doc, meta, start_date_time, priority):
- if meta.has_field("response_by"):
- doc.response_by = get_expected_time_for(parameter="response", service_level=priority, start_date_time=start_date_time)
-
- if meta.has_field("response_by_variance") and not doc.get('first_responded_on'):
- now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
- doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, now_time), 2)
- elif meta.has_field("response_by_variance") and doc.get('first_responded_on'):
- doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, doc.get('first_responded_on')), 2)
-
-def set_resolution_by_and_variance(doc, meta, start_date_time, priority):
- if meta.has_field("resolution_by"):
- doc.resolution_by = get_expected_time_for(parameter="resolution", service_level=priority, start_date_time=start_date_time)
-
- if meta.has_field("resolution_by_variance") and not doc.get("resolution_date"):
- now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
- doc.resolution_by_variance = round(time_diff_in_seconds(doc.resolution_by, now_time), 2)
-
-
def now_datetime(user):
dt = convert_utc_to_user_timezone(datetime.utcnow(), user)
return dt.replace(tzinfo=None)