import csv import pandas as pd import xlsxwriter import datetime import json class CSVTransform: _wip_total = 0 _on_hold_total = 0 _weekly_resolved = 0 _daily_total = 0 _weekly_total = 0 _yearly_total = 0 _priority_daily_counter = { 'P1 - Critical': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P2 - High': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P3 - Medium': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P4 - Low': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'Total': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 } } _priority_weekly_counter = { 'P1 - Critical': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P2 - High': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P3 - Medium': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P4 - Low': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'Total': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 } } _priority_rolling_week_counter = { 'P1 - Critical': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P2 - High': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P3 - Medium': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P4 - Low': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'Total': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 } } _priority_monthly_counter = { 'P1 - Critical': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P2 - High': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P3 - Medium': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P4 - Low': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'Total': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 } } _priority_yearly_counter = { 'P1 - Critical': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P2 - High': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P3 - Medium': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'P4 - Low': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 }, 'Total': { 'Closed':0, 'On Hold':0, 'Work in Progress':0, 'Resolved':0, 'Cancelled':0, 'Total':0 } } _se_counter = { 'Total': {} } _location_counter = { 'Total': {} } _new_activation_counter = { 'Total': {} } _new_activation_priority_counter = {} _new_activation_ci_counter = {} _new_activations = {} _ci_dict = {} _ci_counter = { 'Total': { 'KPS': 0, 'POS_Android':0, 'POS_iOS': 0, 'Misc': 0, } } _daily_incidents = [] _weekly_incidents = [] _8_to_14_incidents = [] _monthly_incidents = [] _sla_tracker = { 'P1 - Critical': { 'time': datetime.timedelta(), 'Total': 0 }, 'P2 - High': { 'time': datetime.timedelta(), 'Total': 0 }, 'P3 - Medium': { 'time': datetime.timedelta(), 'Total': 0 }, 'P4 - Low': { 'time': datetime.timedelta(), 'Total': 0 } } _today = datetime.date.today() _last_week = _today - datetime.timedelta(days=7) _8_days = _today - datetime.timedelta(days=8) _14_days = _today - datetime.timedelta(days=14) _this_month = _today.replace(day=1) -datetime.timedelta(days=1) def __init__(self, filename): self._set_new_activation_variables() self._set_ci_dictionary_variables() self._ingest_csv(filename) def _set_new_activation_variables(self): with open('./configs/new_activations.json') as new_sites: temp_dict = json.load(new_sites) for key in temp_dict.keys(): ## Used to skip locations who haven't been activated yet. ## Could also add a lower bound if we want to auto remove ## locations activated after a certain amount of time. if temp_dict[key]['date'] > str(self._today): continue self._new_activations[key] = temp_dict[key] self._new_activation_counter['Total'][key] = 0 self._new_activation_priority_counter[key] = { 'P1 - Critical': 0, 'P2 - High': 0, 'P3 - Medium': 0, 'P4 - Low': 0 } self._new_activation_ci_counter[key] = { 'KPS': 0, 'POS_Android':0, 'POS_iOS': 0, 'Misc': 0, } def _set_ci_dictionary_variables(self): with open('./configs/ci_dict.json') as json_file: self._ci_dict = json.load(json_file) def _ingest_csv(self, filename): with open(filename, newline='', encoding='cp1252') as csvfile: reader = csv.DictReader(csvfile) for row in reader: self._update_counters(row) self._extract_incidents(row) def _update_counters(self, row): created_datetime = row['sys_created_on'] created_date = datetime.datetime.strptime(created_datetime, '%Y-%m-%d %H:%M:%S %p').date() row_loc = row['location'] if created_date == self._today: self._update_daily_counters(row) if created_date >= self._last_week: self._update_weekly_counters(row) if created_date > self._this_month: self._update_monthly_counters(row) if created_date <= self._8_days and created_date >= self._14_days: self._update_last_week_counters(row) self._yearly_total += 1 self._update_state_counter(row['state']) self._update_assigned_counter(row['assigned_to']) self._update_location_counter(row_loc) self._update_new_activations(row) self._update_ci_counter(row) def _update_daily_counters(self, row): self._daily_total += 1 if self._priority_daily_counter['Total'][row['state']]: self._priority_daily_counter['Total'][row['state']] += 1 else: self._priority_daily_counter['Total'][row['state']] = 1 if self._priority_daily_counter[row['priority']][row['state']]: self._priority_daily_counter[row['priority']][row['state']] += 1 else: self._priority_daily_counter[row['priority']][row['state']] = 1 def _update_weekly_counters(self, row): self._weekly_total += 1 if row['state'] == 'Closed' or row['state'] == 'Resolved': self._weekly_resolved += 1 if self._priority_weekly_counter['Total'][row['state']]: self._priority_weekly_counter['Total'][row['state']] += 1 else: self._priority_weekly_counter['Total'][row['state']] = 1 if self._priority_weekly_counter[row['priority']][row['state']]: self._priority_weekly_counter[row['priority']][row['state']] += 1 else: self._priority_weekly_counter[row['priority']][row['state']] = 1 def _update_last_week_counters(self, row): if self._priority_rolling_week_counter['Total'][row['state']]: self._priority_rolling_week_counter['Total'][row['state']] += 1 else: self._priority_rolling_week_counter['Total'][row['state']] = 1 if self._priority_rolling_week_counter[row['priority']][row['state']]: self._priority_rolling_week_counter[row['priority']][row['state']] += 1 else: self._priority_rolling_week_counter[row['priority']][row['state']] = 1 def _update_monthly_counters(self, row): if self._priority_monthly_counter['Total'][row['state']]: self._priority_monthly_counter['Total'][row['state']] += 1 else: self._priority_monthly_counter['Total'][row['state']] = 1 if self._priority_monthly_counter[row['priority']][row['state']]: self._priority_monthly_counter[row['priority']][row['state']] += 1 else: self._priority_monthly_counter[row['priority']][row['state']] = 1 if row['resolved_at'] != '': self._update_resolved_sla(row) def _update_ci_counter(self, row): if row['ci'] in self._ci_dict['KPS']: self._ci_counter['Total']['KPS'] += 1 elif row['ci'] in self._ci_dict['POS_Android']: self._ci_counter['Total']['POS_Android'] += 1 elif row['ci'] in self._ci_dict['POS_iOS']: self._ci_counter['Total']['POS_iOS'] += 1 else: self._ci_counter['Total']['Misc'] += 1 def _update_new_activations(self, row): created_datetime = row['sys_created_on'] created_date = datetime.datetime.strptime(created_datetime, '%Y-%m-%d %H:%M:%S %p').date() row_loc = row['location'] if row['ci'] in self._ci_dict['KPS']: row_ci = 'KPS' elif row['ci'] in self._ci_dict['POS_Android']: row_ci = 'POS_Android' elif row['ci'] in self._ci_dict['POS_iOS']: row_ci = 'POS_iOS' else: return if row_loc in self._new_activations.keys(): if str(created_date) < self._new_activations[row_loc]['date']: return if row_ci not in self._new_activations[row_loc]['product']: return self._new_activation_counter['Total'][row_loc] += 1 self._new_activation_priority_counter[row_loc][row['priority']] += 1 self._new_activation_ci_counter[row_loc][row_ci] += 1 def _extract_incidents(self, row): created_datetime = row['sys_created_on'] created_date = datetime.datetime.strptime(created_datetime, '%Y-%m-%d %H:%M:%S %p').date() if created_date == self._today: self._daily_incidents.append(row) if created_date >= self._last_week: self._weekly_incidents.append(row) if created_date > self._this_month: self._monthly_incidents.append(row) if created_date <= self._8_days and created_date >= self._14_days: self._8_to_14_incidents.append(row) def _update_state_counter(self, state): if state == 'Work in Progress': self._wip_total += 1 if state == 'On Hold': self._on_hold_total += 1 def _update_assigned_counter(self, assignee): if assignee in self._se_counter['Total']: self._se_counter['Total'][assignee] += 1 else: self._se_counter['Total'][assignee] = 1 def _update_location_counter(self, row_loc): if row_loc in self._location_counter['Total']: self._location_counter['Total'][row_loc] += 1 else: self._location_counter['Total'][row_loc] = 1 def _update_resolved_sla(self, row): created = datetime.datetime.strptime(row['sys_created_on'], '%Y-%m-%d %H:%M:%S %p') resolved = datetime.datetime.strptime(row['resolved_at'], '%Y-%m-%d %H:%M:%S %p') self._sla_tracker[row['priority']]['time'] += (resolved - created) self._sla_tracker[row['priority']]['Total'] += 1 ###### Getters ##### def get_priority_daily_counters(self): return self._priority_daily_counter def get_priority_weekly_counters(self): return self._priority_weekly_counter def get_priority_rolling_week_counters(self): return self._priority_rolling_week_counter def get_priority_monthly_counters(self): return self._priority_monthly_counter def get_se_counter(self): return self._se_counter def get_location_counter(self): return self._location_counter def get_new_activations_counter(self): return self._new_activation_counter def get_na_priority_counter(self): return self._new_activation_priority_counter def get_na_ci_counter(self): return self._new_activation_ci_counter def get_ci_counter(self): return self._ci_counter def get_daily_incident(self): return self._daily_incidents def get_weekly_incident(self): return self._weekly_incidents def get_8_to_14_incident(self): return self._8_to_14_incidents def get_monthly_incident(self): return self._monthly_incidents def get_daily_total(self): return self._daily_total def get_weekly_total(self): return self._weekly_total def get_yearly_total(self): return self._yearly_total def get_weekly_resolved(self): return self._weekly_resolved def get_wip_total(self): return self._wip_total def get_on_hold_total(self): return self._on_hold_total def get_p1_sla_average(self): if self._sla_tracker['P1 - Critical']['Total'] == 0: return 'Not enough data.' return self._sla_tracker['P1 - Critical']['time'] / self._sla_tracker['P1 - Critical']['Total'] def get_p2_sla_average(self): if self._sla_tracker['P2 - High']['Total'] == 0: return 'Not enough data.' return self._sla_tracker['P2 - High']['time'] / self._sla_tracker['P2 - High']['Total'] def get_p3_sla_average(self): if self._sla_tracker['P3 - Medium']['Total'] == 0: return 'Not enough data.' return self._sla_tracker['P3 - Medium']['time'] / self._sla_tracker['P3 - Medium']['Total'] def get_p4_sla_average(self): if self._sla_tracker['P4 - Low']['Total'] == 0: return 'Not enough data.' return self._sla_tracker['P4 - Low']['time'] / self._sla_tracker['P4 - Low']['Total'] if __name__ == "__main__": CSVTransform('incident.csv')