refactor: application of SLA and its metrics

This commit is contained in:
Saqib Ansari 2021-11-26 20:16:21 +05:30
parent 210f593a49
commit a8c75b6862
2 changed files with 137 additions and 205 deletions

View File

@ -341,8 +341,7 @@ scheduler_events = {
"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"

View File

@ -10,7 +10,6 @@ from frappe.core.utils import get_parent_doc
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 set_documents_with_active_service_level_agreement():
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 @@ def get_support_days(service_level):
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 @@ def change_service_level_agreement_and_priority(self):
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 @@ def reset_service_level_agreement(doc, reason, user):
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 @@ def update_agreement_status_on_custom_status(doc):
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 @@ def get_time_in_timedelta(time):
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)