Files
Incident_Report_Generator/csv_transformer.py
2025-11-05 12:58:07 -05:00

554 lines
17 KiB
Python

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')