-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathutils2.py
More file actions
172 lines (135 loc) · 7.9 KB
/
Copy pathutils2.py
File metadata and controls
172 lines (135 loc) · 7.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import csv
from utils import ROLES_BY_LEVEL,DEPARTMENTS_BY_BUSINESS_IMPACT,DEPENDENCIES_BY_CRITICALITY
log_messages=[]
def read_and_process_data(input_path, expected_fields):
global log_messages
try:
with open(input_path,"r") as input_reader:
lines = csv.DictReader(input_reader)
current_fields = lines.fieldnames
missing_headers = [header for header in expected_fields if header not in current_fields]
extra_headers = [header for header in current_fields if header not in expected_fields]
if extra_headers:
log_messages.append(f"[WARNING] Extra header(s) encountered: {extra_headers}.")
if missing_headers == []:
final_data = []
for index,row in enumerate(lines,start=1):
try:
# values =[ value for key,value in row.items()]
missing_fields = [key for key,value in row.items() if value is None or value.strip() == ""]
if not(missing_fields):
found = False
for couple in list(DEPARTMENTS_BY_BUSINESS_IMPACT.items()):
if row["Department"] in couple[1]:
Department = couple[0]
found = True
break
if not found:
log_messages.append(f"[ERROR] ROW{index}: Unknown department '{row['Department']}'. Row Skipped.")
continue
found = False
for pair in list(ROLES_BY_LEVEL.items()):
if row["Job_Role"] in pair[1]:
Role = pair[0]
found = True
break
if not found:
log_messages.append(f"[ERROR] ROW{index}: Unknown role '{row['Job_Role']}'. Row Skipped.")
continue
found = False
for element in list(DEPENDENCIES_BY_CRITICALITY.items()):
if row["Dependency"] in element[1]:
Dependency = element[0]
found = True
break
if not found:
log_messages.append(f"[ERROR] ROW{index}: Unknown dependency '{row['Dependency']}'. Row Skipped.")
continue
try:
complete = float(row["Completed_Tasks"].split()[0])
assigned = float(row["Assigned_Tasks"].split()[0])
if assigned >0:
if abs(complete/assigned*100)>=95:
Task_Performance = 5
elif abs(complete/assigned*100)>85:
Task_Performance = 4
elif abs(complete/assigned*100)>75:
Task_Performance = 3
elif abs(complete/assigned*100)>65:
Task_Performance = 2
else:
Task_Performance = 1
else:
Task_Performance=1
if assigned < 0:
log_messages.append(f"[ERROR] ROW{index}: Negative assigned_tasks value.")
except (ValueError,IndexError) as e:
log_messages.append(f"[ERROR] ROW{index}: Invalid task values. Skipping ROW{index}.")
continue
try:
tenure = float(row["Tenure_in_Company"].split()[0])
if tenure>7:
Tenure = 5
elif tenure>5:
Tenure = 4
elif tenure>3:
Tenure = 3
elif tenure>1:
Tenure = 2
else:
Tenure = 1
except (ValueError,IndexError) as e:
log_messages.append(f"[ERROR] ROW{index}: Invalid tenure value. Skipping ROW{index}.")
continue
composite_score = (Task_Performance * 0.4) + (Role * 0.2) + (Department * 0.15) + (Dependency * 0.15) + (Tenure * 0.1)
if composite_score>4.5:
raise_percent=15*((composite_score/5))
elif composite_score>4:
raise_percent=12*((composite_score/4.49))
elif composite_score>3.5:
raise_percent=9*((composite_score/3.99))
elif composite_score>3:
raise_percent=7*((composite_score/3.49))
elif composite_score>2.5:
raise_percent=5*((composite_score/2.99))
else:
raise_percent=2*((composite_score/2.5))
final_data.append({"Serial_Number":row["Serial_Number"],
"Employee_Name":row["Employee_Name"],
"Employee_ID":row["Employee_ID"],
"Raise_percent":round(raise_percent, 2)})
else:
log_messages.append(f"[ERROR] ROW{index} Missing field(s): {missing_fields}. ROW{index} skipped.")
except Exception as e:
log_messages.append(f"[ERROR] ROW{index}: Unexpected error - {str(e)}. ROW{index} skipped.")
continue
return final_data
else:
log_messages.append(f"[ERROR] Missing required header(s): {missing_headers}.")
print("Fatal Error encountered: MISSING_COLUMNS")
return None
except FileNotFoundError:
log_messages.append(f"[ERROR] Input file '{input_path}' not found. Process aborted.")
def write_output(output_path, output_data):
if output_data==None:
log_messages.append(f"[ERROR] No data to write to '{output_path}'.")
return
try:
global log_messages
with open(output_path,"w",newline="") as output_fileobj:
fields = ["Serial_Number","Employee_Name","Employee_ID","Raise_percent"]
writer = csv.DictWriter(output_fileobj,fieldnames=fields)
writer.writeheader()
writer.writerows(output_data)
log_messages.append(f"[INFO] Output successfully written to '{output_path}'.")
print("Done")
except Exception as e:
log_messages.append(f"[ERROR] Failed to write output file: {str(e)}")
def write_logs(log_path, log_messages):
try:
with open(log_path,"w") as log_writer:
for message in log_messages:
log_writer.write(message+'\n')
log_writer.write("--------------------------------\nAll Functions Executed\n--------------------------------")
except Exception as e:
print(f"Filed to write Log file: {str(e)}")