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 | def getMongoDB(**context):
token = context.get("ti").xcom_pull(key="token")
response = requests.get(
url=f"{dRoW_api_end_url}/api/module/document-export/airflow/workflow/64ba0f78cb634b04c44d9a48?export_type=0",
headers={
"x-access-token": f"Bearer {token}",
"ICWPxAccessKey": "nd@201907ICWP_[1AG:4UdI){n=b~"
}
)
# print('got_data')
RISC_Data = json.loads(response.text)
Mapping= {
# "Sup Rep Signed Date" : "sup_rep_signed_date",
# "contractor_rep_signed_date" : "contractor_rep_signed_date",
"A3. Date Time" : "a3_date_time",
"A1. No. of Walk": "a1_no_of_walk",
# "1. General_compelete": "general_complete",
# "1. General_incompelete": "general_incomplete",
# "2. Flammable Liquids / Gases_compelete": "flammable_liquids_gases_complete",
# "2. Flammable Liquids / Gases_incompelete": "flammable_liquids_gases_incomplete",
# "3. Hazardous Substances_compelete": "general_complete",
# "3. Hazardous Substances_incompelete": "general_incomplete",
}
saftey_cats=[
"1. General",
"2. Flammable Liquids / Gases",
"3. Hazardous Substances",
"4. Electricity",
"5. Fire Precaution",
"6. Working Area",
"7. Lifting Operation",
"8. Material Hoist",
"9. Confined Spaces",
"10. Noise",
"11. Gas Welding and Cutting Equipment",
"12. Electricityâarc Welding",
"13. Mechanical Plant and Equipment",
"14. Tunnel",
"15. Formwork",
"16. Hoarding",
"17. Working at Height",
"18. Abrasive Wheels",
"19. Excavations",
"20. Slings and other Lifting Gears",
"21. Compressed Air/ Pneumatic Air Tools",
"22. Protection of the Public",
"23. Prevention of Mosquito Breed",
"24. Work Over Water",
"25. Welfare Facilities",
# "26. Others / Remarks"
]
# print('start transform')
host = 'drowdatewarehouse.crlwwhgepgi7.ap-east-1.rds.amazonaws.com'
# User name of the database server
dbUserName = 'dRowAdmin'
# Password for the database user
dbUserPassword = 'drowsuper'
# Name of the database
database = 'drowDateWareHouse'
# Character set
charSet = "utf8mb4"
port = "5432"
# #cursor Type
# cusrsorType = pymysql.cursors.DictCursor
#create_engine('mysql+mysqldb://root:password@localhost:3306/mydbname', echo = False)
conn_string = ('postgres://' +
dbUserName + ':' +
dbUserPassword +
'@' + host + ':' + port +
'/' + database)
# df = context.get("ti").xcom_pull(key="InsertData")
# # print(df)
# conn_string = 'postgres://user:password@host/data1'
db = create_engine(conn_string)
conn = db.connect()
# print('db connected')
df = pd.DataFrame()
with conn as conn:
for x in RISC_Data:
df_nested_list = json_normalize(x['data'])
# print('process 1')
df2 = df_nested_list.reindex(columns=Mapping.keys())
if len(x['ApproveLogSummary']) > 0:
# request_date = pd.to_datetime(df2["C1 - Inspect on Date Time"]) - pd.Timedelta(days=1)
request_data = [data for data in x['ApproveLogSummary'] if data.get('statusName')=="B : RSS Check/Agree Report"]
if len(request_data) > 0 and 'from' in request_data[-1]:
df2['sup_rep_signed_date'] = request_data[len(request_data)-1]['from']
else:
df2['sup_rep_signed_date'] = None
if len(request_data) > 0 and 'to' in request_data[-1]:
df2['contractor_rep_signed_date'] = request_data[len(request_data)-1]['to']
else:
df2['contractor_rep_signed_date'] = None
else:
df2['sup_rep_signed_date'] = None
df2['contractor_rep_signed_date'] = None
if x['data']['A1. No. of Walk'] != None :
df2["report_name"] = x['data']['A1. No. of Walk']
else :
df2["report_name"] = None
if (len(x['data']['C Summary of Follow-up Actions']) > 0):
total_late_retification = 0
for summaryData in x['data']['C Summary of Follow-up Actions']:
if ("Agreed Due Date for Completion" in summaryData and "Agreed Due Date for Completion" in summaryData and not (summaryData["Agreed Due Date for Completion"]!='') and (not (summaryData["Date Completion"]!='')) and (summaryData["Agreed Due Date for Completion"].astype('datetime64[ns]') < summaryData["Date Completion"].astype('datetime64[ns]')).bool()):
total_late_retification += 1
df2['total_late_retification'] = total_late_retification
else:
df2['total_late_retification'] = 0
if (not df2['contractor_rep_signed_date'].isnull().bool() and not df2['A3. Date Time'].isnull().bool()):
df2['days_complete'] = (((df2['contractor_rep_signed_date'].astype('datetime64[ns]') -
df2['A3. Date Time'].astype('datetime64[ns]'))/ np.timedelta64(1, 'h'))/24).round(2)
if df2['days_complete'].isnull().bool() or df2['days_complete'].lt(0).bool():
df2['days_complete'] = 0
else:
df2['days_complete'] = None
df4=pd.DataFrame()
for saftey_cat in saftey_cats:
df3=df2.copy()
complete = 0
incomplete = 0
if not df2['sup_rep_signed_date'].isnull().bool():
if (len(x['data'][str(saftey_cat)[0:3].strip()+' Checklist']) > 0):
for record in x['data'][str(saftey_cat)[0:3].strip()+' Checklist']:
if record[str(saftey_cat)[0:3].strip()+' Result'] != 'N/A' and record[str(saftey_cat)[0:3].strip()+' Result'].strip() != "":
complete += 1
else:
if (len(x['data'][str(saftey_cat)[0:3].strip()+' Checklist']) > 0):
for record in x['data'][str(saftey_cat)[0:3].strip()+' Checklist']:
if record[str(saftey_cat)[0:3].strip()+' Result'] != 'N/A' and record[str(saftey_cat)[0:3].strip()+' Result'].strip() != "":
incomplete += 1
df3['saftey_cat'] = saftey_cat
df3['saftey_cat' + '_' + 'complete'] = complete
df3['saftey_cat' + '_' + 'incomplete'] = incomplete
df4 = df4.append(df3)
df2=df2.append(df4)
df = df.append(df2)
df.rename(columns=Mapping, inplace=True)
df['sup_rep_signed_date']=df['sup_rep_signed_date'].apply(pd.to_datetime)
df['contractor_rep_signed_date']=df['contractor_rep_signed_date'].apply(pd.to_datetime)
df['a3_date_time']=df['a3_date_time'].apply(pd.to_datetime)
df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
df.to_sql('safety_walk_5wsd21', con=conn, if_exists='replace', index= False)
|