try:

    from datetime import timedelta
    from airflow import DAG
    
    from airflow.operators.python_operator import PythonOperator
    from airflow.operators.http_operator import SimpleHttpOperator
    from datetime import datetime
    from pandas.io.json import json_normalize
    from airflow.operators.postgres_operator import PostgresOperator
    from pandas.tseries.offsets import CustomBusinessDay

    import pandas as pd
    import json
    import requests
    import numpy as np

    import psycopg2
    from sqlalchemy import create_engine
    #print("All Dag moudules are sucessfully imported")

except Exception as e:
    print("Error {} ".format(e))

dRoW_api_end_url = "https://drow.cloud"

def getDrowToken(**context):

    response = requests.post(
    url=f"{dRoW_api_end_url}/api/auth/authenticate",
    data={
    "username": "icwp2@drow.cloud",
    "password": "dGVzdDAxQHRlc3QuY29t"
    }
    ).json()
    context["ti"].xcom_push(key="token", value=response['token'])
    # return 'DLLM{}'.format(response)

def getdrowPSQLConnectionString():
    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"

    conn_string = ('postgres://' +
                           dbUserName + ':' + 
                           dbUserPassword +
                           '@' + host + ':' + port +
                           '/' + database)
    print(conn_string)
    return conn_string

def getMongoDB1c(**context):
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/module/document-export/airflow/workflow/64d8a50b15942b0c9dbe1652?export_type=0",
    headers={
    "x-access-token": f"Bearer {token}",
    "ICWPxAccessKey": "6WSD21ICWP_[1AG:4UdI){n=b~"
    }
    )
    print('got_data')
    RISC_Data = json.loads(response.text)
    # Request Submission A3 - Data Time for Inspection minus 24hrs.
    Mapping= {
    "B1 - Received Date Time" : "b01_request_received_date_time",
    'C1 - Inspect on Date Time' : 'c02_inspect_or_survey_on_date_time',
    "A3 - Data Time for Inspection" : "a05_proposed_inspection_or_survey_date_time",
    "E2 - Received by Contractor Date Time*" : "e01_received_on_behalf_of_contractor_on_date_time",
    'Rev': "a01a_request_no_revision",
    "A1 - Request No.": "a01_request_no",
    "C9 - Pass to Contractor's obligations DateTime" : 'c12_time_pass_to_senior_or_contractor',
    "D - Countersigned on Date Time*" : 'd01_countersigned_on_date_time',
    'C2 - Permission given?' : "c03_approval_given",
    "Work Category" : "a01b_work_category",
    "A4 - Location of Works": "a03_location_of_work",
    "C10 - Pass on Date Time": "c10_pass_on_date_time"
    # 'C12 Time Pass to Senior or Contractor' : 'Sign time by manager'
    }
    
    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    print('db connected')
    with conn as conn:
        df = pd.DataFrame()
        for x in RISC_Data:
            #print(x)
            df_nested_list = json_normalize(x['data'])
            df2 = df_nested_list.reindex(columns=Mapping.keys())
            print('Data', x['data'])
            df2["A1 - Request No."] = x['data']['A1a - Trade'] + x['data']['A1b - Type'] + x['data']['A1c - Form No.']
            df2["request_no"] = df2["A1 - Request No."].astype(str) + df2["Rev"]
            if df2['Work Category'].isnull().any() or (df2['Work Category'] == "").any():
                df2['Work Category']='Other'
            if len(x['ApproveLogSummary']) > 0:
                # request_date = df2["A3 - Data Time for Inspection"]
                request_data = [data for data in x['ApproveLogSummary'] if data.get('statusName')=="B1 : SIOW/IOW Assign Inspection"]
                if len(request_data) > 0:
                    _request_date = request_data[0]['from']
                else:
                    _request_date = None
            else:
                _request_date = None
            df2["a10_request_submission_date_time"] = _request_date
            if not df2["a10_request_submission_date_time"].isnull().bool():
                df2["a10_request_submission_date_time"]=df2["a10_request_submission_date_time"].astype('datetime64[ns]') 
                # + pd.Timedelta(8, unit='h')
            if (not df2["A3 - Data Time for Inspection"].isnull().bool()):
                df2["A3 - Data Time for Inspection"] = df2["A3 - Data Time for Inspection"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2["C1 - Inspect on Date Time"] = df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C10 - Pass on Date Time"].isnull().bool()):
                df2["C10 - Pass on Date Time"] = df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')

            # if(not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and ((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool())):
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))
            # else:
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))

            if (not df2["C1 - Inspect on Date Time"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool()):
                df2['nc_report'] = True
            else:
                df2['nc_report'] = False

            if (not df2["A3 - Data Time for Inspection"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (((df2["A3 - Data Time for Inspection"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) < pd.Timedelta(24, unit='h')).bool() and (((df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) > pd.Timedelta(0, unit='h')).bool()):
                df2['urgent_report'] = True
            else:
                df2['urgent_report'] = False

            if (not df2['E2 - Received by Contractor Date Time*'].isnull().bool() and not df2['a10_request_submission_date_time'].isnull().bool()):
                # Define custom business days, excluding weekends (Saturday and Sunday)
                custom_business_day = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri')
                if _request_date!=None and x['data']['E2 - Received by Contractor Date Time*']!=None:
                    start_date = np.datetime64(_request_date, 'ns').astype('datetime64[D]')
                    end_date = np.datetime64(x['data']['E2 - Received by Contractor Date Time*'], 'ns').astype('datetime64[D]')
                    # Calculate the elapsed time in business days
                    df2['elapsed_time'] = (end_date - start_date)/np.timedelta64(1,"D")
                    # df2['elapsed_time'] = np.busday_count(start_date, end_date, weekmask=custom_business_day.weekmask)
                    # df2['elapsed_time'] = np.busday_count(_request_date,
                    #                       x['E2 - Received by Contractor Date Time*'],
                    #                       weekmask=custom_business_day.weekmask)
                    # Round the elapsed time to two decimal places
                    df2['elapsed_time'] = df2['elapsed_time'].round(2)
                else:
                    df2['elapsed_time'] = 0
            else:
                df2['elapsed_time'] = 0
            
            if (not df2["C1 - Inspect on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and (((df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]') - df2["C1 - Inspect on Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['overdue_report'] = True
            else:
                df2['overdue_report'] = False
            
            if (not df2["C10 - Pass on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and ((( df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['delayed_approval_report'] = True
            else:
                df2['delayed_approval_report'] = False

            if (((df2['Rev']=="").bool() or df2['Rev'].isnull().bool()) and (df2['C2 - Permission given?']=="No").bool()):
                df2['fail_in_first_inspection'] = True
            else:
                df2['fail_in_first_inspection'] = False
            
            if (not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and not df2["C1 - Inspect on Date Time"].isnull().bool() and not df2["E2 - Received by Contractor Date Time*"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'complete'
            elif ((df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and not df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'in-complete'
            elif ((not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'outstanding'
            else:
                df2['complete_incomplete_outstanding_report'] = 'outstanding'            

            #print('process 2')
            df2.rename(columns=Mapping, inplace=True)
            df = df.append(df2)     

        # Sort dataframe by Request No. and Rev No. in descending order
        df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # Create a new column 'is_latest' and set the initial value to False
        df['is_latest'] = False

        # Assign a unique identifier to None values
        none_identifier = np.nan

        # Replace None values with the unique identifier
        df['a01_request_no'].fillna(none_identifier, inplace=True)

        # Group by Request No. and update the 'is_latest' column for the latest version
        df['is_latest'] = df.groupby('a01_request_no')['a01a_request_no_revision'].transform(lambda x: x.fillna('') == x.fillna('').max())

        # Reset the unique identifier to None
        df['a01_request_no'].replace(none_identifier, None, inplace=True)

        # df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # # Group by a01_request_no and select the first row of each group
        # latest_versions = df.groupby('a01_request_no').first().reset_index()

        # # Create a new column 'is_latest' and set the value to True for the latest versions
        # latest_versions['is_latest'] = True

        # # Merge the 'is_latest' column back to the original dataframe
        # df = pd.merge(df, latest_versions[['a01_request_no', 'is_latest']], on='a01_request_no', how='left')

        # # Fill the missing values in 'is_latest' column with False
        # df['is_latest'].fillna(False, inplace=True)

        df.to_sql('risc_6wsd21', con=conn, if_exists='append', index= False)
        conn.close()

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/62b58ca50871de0c91eed809?export_type=0",
    headers={
    "x-access-token": f"Bearer {token}",
    "ICWPxAccessKey": "6WSD21ICWP_[1AG:4UdI){n=b~"
    }
    )
    print('got_data')
    RISC_Data = json.loads(response.text)
    # Request Submission A3 - Data Time for Inspection minus 24hrs.
    Mapping= {
    "B1 - Received Date Time" : "b01_request_received_date_time",
    'C1 - Inspect on Date Time' : 'c02_inspect_or_survey_on_date_time',
    "A3 - Data Time for Inspection" : "a05_proposed_inspection_or_survey_date_time",
    "E2 - Received by Contractor Date Time*" : "e01_received_on_behalf_of_contractor_on_date_time",
    'Rev': "a01a_request_no_revision",
    "A1 - Request No.": "a01_request_no",
    "C9 - Pass to Contractor's obligations DateTime" : 'c12_time_pass_to_senior_or_contractor',
    "D - Countersigned on Date Time*" : 'd01_countersigned_on_date_time',
    'C2 - Permission given?' : "c03_approval_given",
    "Work Category" : "a01b_work_category",
    "A4 - Location of Works": "a03_location_of_work",
    "C10 - Pass on Date Time": "c10_pass_on_date_time"
    # 'C12 Time Pass to Senior or Contractor' : 'Sign time by manager'
    }
    
    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    print('db connected')
    with conn as conn:
        df = pd.DataFrame()
        for x in RISC_Data:
            #print(x)
            if (x['data']["A1 - Request No."] == None) or (x['data']["A1 - Request No."] == "") or (x['data']["A1 - Request No."] == "null0")or (x['data']["A1 - Request No."][-5:].isnumeric() and int(x['data']["A1 - Request No."][-5:]) < 358):
                print(x['data']['A1 - Request No.'])
                continue
            df_nested_list = json_normalize(x['data'])
            df2 = df_nested_list.reindex(columns=Mapping.keys())
            if df2["A1 - Request No."].isnull().any() or (df2["A1 - Request No."] == "").any():
                continue
            df2["request_no"] = df2["A1 - Request No."].astype(str) + df2["Rev"]
            if df2['Work Category'].isnull().any() or (df2['Work Category'] == "").any():
                df2['Work Category']='Other'
            if len(x['ApproveLogSummary']) > 0:
                # request_date = df2["A3 - Data Time for Inspection"]
                request_data = [data for data in x['ApproveLogSummary'] if data.get('statusName')=="B1 : SIOW/IOW Assign Inspection"]
                if len(request_data) > 0:
                    _request_date = request_data[0]['from']
                else:
                    _request_date = None
            else:
                _request_date = None
            df2["a10_request_submission_date_time"] = _request_date
            if not df2["a10_request_submission_date_time"].isnull().bool():
                df2["a10_request_submission_date_time"]=df2["a10_request_submission_date_time"].astype('datetime64[ns]') 
                # + pd.Timedelta(8, unit='h')
            if (not df2["A3 - Data Time for Inspection"].isnull().bool()):
                df2["A3 - Data Time for Inspection"] = df2["A3 - Data Time for Inspection"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2["C1 - Inspect on Date Time"] = df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C10 - Pass on Date Time"].isnull().bool()):
                df2["C10 - Pass on Date Time"] = df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')

            # if(not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and ((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool())):
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))
            # else:
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))

            if (not df2["C1 - Inspect on Date Time"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool()):
                df2['nc_report'] = True
            else:
                df2['nc_report'] = False

            if (not df2["A3 - Data Time for Inspection"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (((df2["A3 - Data Time for Inspection"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) < pd.Timedelta(24, unit='h')).bool() and (((df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) > pd.Timedelta(0, unit='h')).bool()):
                df2['urgent_report'] = True
            else:
                df2['urgent_report'] = False

            if (not df2['E2 - Received by Contractor Date Time*'].isnull().bool() and not df2['a10_request_submission_date_time'].isnull().bool()):
                # Define custom business days, excluding weekends (Saturday and Sunday)
                custom_business_day = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri')
                if _request_date!=None and x['data']['E2 - Received by Contractor Date Time*']!=None:
                    start_date = np.datetime64(_request_date, 'ns').astype('datetime64[D]')
                    end_date = np.datetime64(x['data']['E2 - Received by Contractor Date Time*'], 'ns').astype('datetime64[D]')
                    # Calculate the elapsed time in business days
                    df2['elapsed_time'] = (end_date - start_date)/np.timedelta64(1,"D")
                    # df2['elapsed_time'] = np.busday_count(start_date, end_date, weekmask=custom_business_day.weekmask)
                    # df2['elapsed_time'] = np.busday_count(_request_date,
                    #                       x['E2 - Received by Contractor Date Time*'],
                    #                       weekmask=custom_business_day.weekmask)
                    # Round the elapsed time to two decimal places
                    df2['elapsed_time'] = df2['elapsed_time'].round(2)
                else:
                    df2['elapsed_time'] = 0
            else:
                df2['elapsed_time'] = 0
            
            if (not df2["C1 - Inspect on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and (((df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]') - df2["C1 - Inspect on Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['overdue_report'] = True
            else:
                df2['overdue_report'] = False
            
            if (not df2["C10 - Pass on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and ((( df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['delayed_approval_report'] = True
            else:
                df2['delayed_approval_report'] = False

            if (((df2['Rev']=="").bool() or df2['Rev'].isnull().bool()) and (df2['C2 - Permission given?']=="No").bool()):
                df2['fail_in_first_inspection'] = True
            else:
                df2['fail_in_first_inspection'] = False
            
            if (not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and not df2["C1 - Inspect on Date Time"].isnull().bool() and not df2["E2 - Received by Contractor Date Time*"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'complete'
            elif ((df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and not df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'in-complete'
            elif ((not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'outstanding'
            else:
                df2['complete_incomplete_outstanding_report'] = 'outstanding'            

            #print('process 2')
            df2.rename(columns=Mapping, inplace=True)
            df = df.append(df2)     

        # Sort dataframe by Request No. and Rev No. in descending order
        df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # Create a new column 'is_latest' and set the initial value to False
        df['is_latest'] = False

        # Assign a unique identifier to None values
        none_identifier = np.nan

        # Replace None values with the unique identifier
        df['a01_request_no'].fillna(none_identifier, inplace=True)

        # Group by Request No. and update the 'is_latest' column for the latest version
        df['is_latest'] = df.groupby('a01_request_no')['a01a_request_no_revision'].transform(lambda x: x.fillna('') == x.fillna('').max())

        # Reset the unique identifier to None
        df['a01_request_no'].replace(none_identifier, None, inplace=True)

        # df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # # Group by a01_request_no and select the first row of each group
        # latest_versions = df.groupby('a01_request_no').first().reset_index()

        # # Create a new column 'is_latest' and set the value to True for the latest versions
        # latest_versions['is_latest'] = True

        # # Merge the 'is_latest' column back to the original dataframe
        # df = pd.merge(df, latest_versions[['a01_request_no', 'is_latest']], on='a01_request_no', how='left')

        # # Fill the missing values in 'is_latest' column with False
        # df['is_latest'].fillna(False, inplace=True)

        df.to_sql('risc_6wsd21', con=conn, if_exists='append', index= False)
        conn.close()
        
def getMongoDB1d(**context):
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/module/document-export/airflow/workflow/64d8f16d92f9db0cb2fc2f18?export_type=0",
    headers={
    "x-access-token": f"Bearer {token}",
    "ICWPxAccessKey": "6WSD21ICWP_[1AG:4UdI){n=b~"
    }
    )
    print('got_data')
    RISC_Data = json.loads(response.text)
    # Request Submission A3 - Data Time for Inspection minus 24hrs.
    Mapping= {
    "B1 - Received Date Time" : "b01_request_received_date_time",
    'C1 - Inspect on Date Time' : 'c02_inspect_or_survey_on_date_time',
    "A3 - Data Time for Inspection" : "a05_proposed_inspection_or_survey_date_time",
    "E2 - Received by Contractor Date Time*" : "e01_received_on_behalf_of_contractor_on_date_time",
    'Rev': "a01a_request_no_revision",
    "A1 - Request No.": "a01_request_no",
    "C9 - Pass to Contractor's obligations DateTime" : 'c12_time_pass_to_senior_or_contractor',
    "D - Countersigned on Date Time*" : 'd01_countersigned_on_date_time',
    'C2 - Permission given?' : "c03_approval_given",
    "Work Category" : "a01b_work_category",
    "A4 - Location of Works": "a03_location_of_work",
    "C10 - Pass on Date Time": "c10_pass_on_date_time"
    # 'C12 Time Pass to Senior or Contractor' : 'Sign time by manager'
    }
    
    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    print('db connected')
    with conn as conn:
        df = pd.DataFrame()
        for x in RISC_Data:
            #print(x)
            df_nested_list = json_normalize(x['data'])
            df2 = df_nested_list.reindex(columns=Mapping.keys())
            if df2["A1 - Request No."].isnull().any() or (df2["A1 - Request No."] == "").any():
                continue
            df2["request_no"] = df2["A1 - Request No."].astype(str) + df2["Rev"]
            if df2['Work Category'].isnull().any() or (df2['Work Category'] == "").any():
                df2['Work Category']='Other'
            if len(x['ApproveLogSummary']) > 0:
                # request_date = df2["A3 - Data Time for Inspection"]
                request_data = [data for data in x['ApproveLogSummary'] if data.get('statusName')=="B1 : SIOW/IOW Assign Inspection"]
                if len(request_data) > 0:
                    _request_date = request_data[0]['from']
                else:
                    _request_date = None
            else:
                _request_date = None
            df2["a10_request_submission_date_time"] = _request_date
            if not df2["a10_request_submission_date_time"].isnull().bool():
                df2["a10_request_submission_date_time"]=df2["a10_request_submission_date_time"].astype('datetime64[ns]') 
                # + pd.Timedelta(8, unit='h')
            if (not df2["A3 - Data Time for Inspection"].isnull().bool()):
                df2["A3 - Data Time for Inspection"] = df2["A3 - Data Time for Inspection"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2["C1 - Inspect on Date Time"] = df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C10 - Pass on Date Time"].isnull().bool()):
                df2["C10 - Pass on Date Time"] = df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')

            # if(not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and ((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool())):
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))
            # else:
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))

            if (not df2["C1 - Inspect on Date Time"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool()):
                df2['nc_report'] = True
            else:
                df2['nc_report'] = False

            if (not df2["A3 - Data Time for Inspection"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (((df2["A3 - Data Time for Inspection"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) < pd.Timedelta(24, unit='h')).bool() and (((df2["C1 - Inspect on Date Time"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) > pd.Timedelta(0, unit='h')).bool()):
                df2['urgent_report'] = True
            else:
                df2['urgent_report'] = False

            if (not df2['E2 - Received by Contractor Date Time*'].isnull().bool() and not df2['a10_request_submission_date_time'].isnull().bool()):
                # Define custom business days, excluding weekends (Saturday and Sunday)
                custom_business_day = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri')
                if _request_date!=None and x['data']['E2 - Received by Contractor Date Time*']!=None:
                    start_date = np.datetime64(_request_date, 'ns').astype('datetime64[D]')
                    end_date = np.datetime64(x['data']['E2 - Received by Contractor Date Time*'], 'ns').astype('datetime64[D]')
                    # Calculate the elapsed time in business days
                    df2['elapsed_time'] = (end_date - start_date)/np.timedelta64(1,"D")
                    # df2['elapsed_time'] = np.busday_count(start_date, end_date, weekmask=custom_business_day.weekmask)
                    # df2['elapsed_time'] = np.busday_count(_request_date,
                    #                       x['E2 - Received by Contractor Date Time*'],
                    #                       weekmask=custom_business_day.weekmask)
                    # Round the elapsed time to two decimal places
                    df2['elapsed_time'] = df2['elapsed_time'].round(2)
                else:
                    df2['elapsed_time'] = 0
            else:
                df2['elapsed_time'] = 0
            
            if (not df2["C1 - Inspect on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and (((df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]') - df2["C1 - Inspect on Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['overdue_report'] = True
            else:
                df2['overdue_report'] = False
            
            if (not df2["C10 - Pass on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and ((( df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['delayed_approval_report'] = True
            else:
                df2['delayed_approval_report'] = False

            if (((df2['Rev']=="").bool() or df2['Rev'].isnull().bool()) and (df2['C2 - Permission given?']=="No").bool()):
                df2['fail_in_first_inspection'] = True
            else:
                df2['fail_in_first_inspection'] = False
            
            if (not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and not df2["C1 - Inspect on Date Time"].isnull().bool() and not df2["E2 - Received by Contractor Date Time*"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'complete'
            elif ((df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and not df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'in-complete'
            elif ((not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and df2["C1 - Inspect on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'outstanding'
            else:
                df2['complete_incomplete_outstanding_report'] = 'outstanding'            

            #print('process 2')
            df2.rename(columns=Mapping, inplace=True)
            df = df.append(df2)     

        # Sort dataframe by Request No. and Rev No. in descending order
        df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # Create a new column 'is_latest' and set the initial value to False
        df['is_latest'] = False

        # Assign a unique identifier to None values
        none_identifier = np.nan

        # Replace None values with the unique identifier
        df['a01_request_no'].fillna(none_identifier, inplace=True)

        # Group by Request No. and update the 'is_latest' column for the latest version
        df['is_latest'] = df.groupby('a01_request_no')['a01a_request_no_revision'].transform(lambda x: x.fillna('') == x.fillna('').max())

        # Reset the unique identifier to None
        df['a01_request_no'].replace(none_identifier, None, inplace=True)

        # df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # # Group by a01_request_no and select the first row of each group
        # latest_versions = df.groupby('a01_request_no').first().reset_index()

        # # Create a new column 'is_latest' and set the value to True for the latest versions
        # latest_versions['is_latest'] = True

        # # Merge the 'is_latest' column back to the original dataframe
        # df = pd.merge(df, latest_versions[['a01_request_no', 'is_latest']], on='a01_request_no', how='left')

        # # Fill the missing values in 'is_latest' column with False
        # df['is_latest'].fillna(False, inplace=True)

        df.to_sql('risc_6wsd21', con=conn, if_exists='append', index= False)
        conn.close()

def getMongoDBSurveyCheck(**context):
    token = context.get("ti").xcom_pull(key="token")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/module/document-export/airflow/workflow/62b5946a2fbfff0c83eeb25d?export_type=0",
    headers={
    "x-access-token": f"Bearer {token}",
    "ICWPxAccessKey": "6WSD21ICWP_[1AG:4UdI){n=b~"
    }
    )
    print('got_data')
    RISC_Data = json.loads(response.text)
    # Request Submission A3 - Data Time for Inspection minus 24hrs.
    Mapping= {
    "B1 - Received Date Time" : "b01_request_received_date_time",
    'C1 - Survey Check on Date Time' : 'c02_inspect_or_survey_on_date_time',
    "A3 - Date Time for Survey Check" : "a05_proposed_inspection_or_survey_date_time",
    "E2 - Received by Contractor Date Time*" : "e01_received_on_behalf_of_contractor_on_date_time",
    'Rev': "a01a_request_no_revision",
    "A1 - Request No.": "a01_request_no",
    "C9 - Pass to Contractor's obligations DateTime" : 'c12_time_pass_to_senior_or_contractor',
    "D1 - Countersigned on Date Time*" : 'd01_countersigned_on_date_time',
    'C2 - Permission given?' : "c03_approval_given",
    "Work Category" : "a01b_work_category",
    "A4 - Location of Works": "a03_location_of_work",
    "C10 - Pass on Date Time": "c10_pass_on_date_time"
    # 'C12 Time Pass to Senior or Contractor' : 'Sign time by manager'
    }
    
    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    print('db connected')
    with conn as conn:
        df = pd.DataFrame()
        for x in RISC_Data:
            #print(x)
            df_nested_list = json_normalize(x['data'])
            df2 = df_nested_list.reindex(columns=Mapping.keys())
            if df2["A1 - Request No."].isnull().any() or (df2["A1 - Request No."] == "").any():
                continue
            df2["request_no"] = df2["A1 - Request No."].astype(str) + df2["Rev"]
            # if(df2['Work Category'].isnull().bool()):
            df2['a01b_work_category']='Survey'
            if len(x['ApproveLogSummary']) > 0:
                # request_date = df2['A3 - Date Time for Survey Check']
                request_data = [data for data in x['ApproveLogSummary'] if data.get('statusName')=="B1 : SSOE/SOE Assign Survey Check"]
                if len(request_data) > 0:
                    request_date = datetime.strptime(request_data[0]['from'], '%Y-%m-%dT%H:%M:%S.%f%z')
                else:
                    request_date = None
            else:
                request_date = None
            df2["a10_request_submission_date_time"] = request_date
            if not df2["a10_request_submission_date_time"].isnull().bool():
                df2["a10_request_submission_date_time"]=df2["a10_request_submission_date_time"].astype('datetime64[ns]') 
                # + pd.Timedelta(8, unit='h')
            if (not df2["C1 - Survey Check on Date Time"].isnull().bool()):
                df2["C1 - Survey Check on Date Time"] = df2["C1 - Survey Check on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["A3 - Date Time for Survey Check"].isnull().bool()):
                df2["A3 - Date Time for Survey Check"] = df2["A3 - Date Time for Survey Check"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            if (not df2["C10 - Pass on Date Time"].isnull().bool()):
                df2["C10 - Pass on Date Time"] = df2["C10 - Pass on Date Time"].astype('datetime64[ns]') - pd.Timedelta(8, unit='h')
            
            # if(not df2["C02 - Inspect on Date Time"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and ((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool())):
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))
            # else:
            #     print((df2["C02 - Inspect on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')))

            if (not df2["C1 - Survey Check on Date Time"].isnull().bool() and (not df2["a10_request_submission_date_time"].isnull().bool()) and (df2["C1 - Survey Check on Date Time"].astype('datetime64[ns]') < df2["a10_request_submission_date_time"].astype('datetime64[ns]')).bool()):
                df2['nc_report'] = True
            else:
                df2['nc_report'] = False

            if (not df2["A3 - Date Time for Survey Check"].isnull().bool() and not df2["a10_request_submission_date_time"].isnull().bool() and (((df2["A3 - Date Time for Survey Check"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) < pd.Timedelta(24, unit='h')).bool() and (((df2["A3 - Date Time for Survey Check"].astype('datetime64[ns]') - df2["a10_request_submission_date_time"].astype('datetime64[ns]'))) > pd.Timedelta(0, unit='h')).bool()):
                df2['urgent_report'] = True
            else:
                df2['urgent_report'] = False

            if (not df2['E2 - Received by Contractor Date Time*'].isnull().bool() and not df2['a10_request_submission_date_time'].isnull().bool()):
                custom_business_day = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri')
                if request_date!=None and x['data']['E2 - Received by Contractor Date Time*']!=None:
                    start_date = np.datetime64(request_date, 'ns').astype('datetime64[D]')
                    end_date = np.datetime64(x['data']['E2 - Received by Contractor Date Time*'], 'ns').astype('datetime64[D]')
                    # Calculate the elapsed time in business days
                    df2['elapsed_time'] = (end_date - start_date)/np.timedelta64(1,"D")
                    # df2['elapsed_time'] = np.busday_count(start_date, end_date, weekmask=custom_business_day.weekmask)
                    # df2['elapsed_time'] = np.busday_count(_request_date,
                    #                       x['E2 - Received by Contractor Date Time*'],
                    #                       weekmask=custom_business_day.weekmask)
                    # Round the elapsed time to two decimal places
                    df2['elapsed_time'] = df2['elapsed_time'].round(2)
            else:
                df2['elapsed_time'] = 0
            
            if (not df2["C1 - Survey Check on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and (((df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]') - df2["C1 - Survey Check on Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
                df2['overdue_report'] = True
            else:
                df2['overdue_report'] = False
            
            if (not df2["C10 - Pass on Date Time"].isnull().bool() and not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and ((( df2["C10 - Pass on Date Time"].astype('datetime64[ns]')) - df2["C9 - Pass to Contractor's obligations DateTime"].astype('datetime64[ns]'))>= pd.Timedelta(24, unit='h')).bool()):
                df2['delayed_approval_report'] = True
            else:
                df2['delayed_approval_report'] = False

            if (((df2['Rev']=="").bool() or df2['Rev'].isnull().bool()) and (df2['C2 - Permission given?']=="No").bool()):
                df2['fail_in_first_inspection'] = True
            else:
                df2['fail_in_first_inspection'] = False
            
            if (not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() and not df2["C1 - Survey Check on Date Time"].isnull().bool() and not df2["E2 - Received by Contractor Date Time*"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'complete'
            elif ((df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and not df2["C1 - Survey Check on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'in-complete'
            elif ((not df2["C9 - Pass to Contractor's obligations DateTime"].isnull().bool() or df2["E2 - Received by Contractor Date Time*"].isnull().bool()) and df2["C1 - Survey Check on Date Time"].isnull().bool()):
                df2['complete_incomplete_outstanding_report'] = 'outstanding'
            else:
                df2['complete_incomplete_outstanding_report'] = 'outstanding'            

            df2.rename(columns=Mapping, inplace=True)
            df = df.append(df2)

        # Sort dataframe by Request No. and Rev No. in descending order
        df.sort_values(['a01_request_no', 'a01a_request_no_revision'], ascending=[True, False], inplace=True)

        # Create a new column 'is_latest' and set the initial value to False
        df['is_latest'] = False

        # Assign a unique identifier to None values
        none_identifier = np.nan

        # Replace None values with the unique identifier
        df['a01_request_no'].fillna(none_identifier, inplace=True)

        # Group by Request No. and update the 'is_latest' column for the latest version
        df['is_latest'] = df.groupby('a01_request_no')['a01a_request_no_revision'].transform(lambda x: x.fillna('') == x.fillna('').max())

        # Reset the unique identifier to None
        df['a01_request_no'].replace(none_identifier, None, inplace=True)

        df.to_sql('risc_6wsd21', con=conn, if_exists='append', index= False)
        conn.close()


delete_table_sql_query = """
DROP TABLE IF EXISTS risc_6wsd21
"""

# id SERIAL
create_table_sql_query = """ 
    CREATE TABLE IF NOT EXISTS risc_6wsd21 (
    a10_request_submission_date_time TIMESTAMP, 
    b01_request_received_date_time TIMESTAMP,
    c02_inspect_or_survey_on_date_time TIMESTAMP,
    a05_proposed_inspection_or_survey_date_time TIMESTAMP,
    e01_received_on_behalf_of_contractor_on_date_time TIMESTAMP,
    a01a_request_no_revision VARCHAR (150),
    a01_request_no VARCHAR (150),
    c12_time_pass_to_senior_or_contractor TIMESTAMP,
    d01_countersigned_on_date_time TIMESTAMP,
    c03_approval_given VARCHAR (150),
    request_no VARCHAR (150),
    a01b_work_category VARCHAR (150),
    nc_report BOOLEAN,
    urgent_report BOOLEAN,
    elapsed_time NUMERIC(10,2),
    overdue_report BOOLEAN,
    delayed_approval_report BOOLEAN,
    fail_in_first_inspection BOOLEAN,
    complete_incomplete_outstanding_report VARCHAR(250),
    a03_location_of_work VARCHAR(250),
    is_latest BOOLEAN,
    c10_pass_on_date_time TIMESTAMP
    );
    """

# */2 * * * * Execute every two minute 
with DAG(
        dag_id="6wsd21_risc_icwp_inspection",
        schedule_interval="0 7 * * *",
        default_args={
            "owner": "airflow",
            "retries": 1,
            "retry_delay": timedelta(minutes=5),
            "start_date": datetime(2023, 1, 17)
        },
        catchup=False) as f:
    
    getMongoDB = PythonOperator(
        task_id="getMongoDB",
        python_callable=getMongoDB,
        op_kwargs={"name": "Dylan"},
        provide_context=True,
    )

    getMongoDB1c = PythonOperator(
        task_id="getMongoDB1c",
        python_callable=getMongoDB1c,
        op_kwargs={"name": "Dylan"},
        provide_context=True,
    )
    
    getMongoDB1d = PythonOperator(
        task_id="getMongoDB1d",
        python_callable=getMongoDB1d,
        op_kwargs={"name": "Dylan"},
        provide_context=True,
    )
    getMongoDBSurveyCheck = PythonOperator(
        task_id="getMongoDBSurveyCheck",
        python_callable=getMongoDBSurveyCheck,
        op_kwargs={"name": "Dylan"},
        provide_context=True,
    )

    # reformData = PythonOperator(
    #     task_id="reformData",
    #     python_callable=reformData,
    #     provide_context=True,
    #     # op_kwargs={"name": "Dylan"}
    # )

    getDrowToken = PythonOperator(
        task_id="getDrowToken",
        python_callable=getDrowToken,
        provide_context=True,
        # op_kwargs={"name": "Dylan"}
    )

    # insertData = PythonOperator(
    #     task_id="insetDateToPG",
    #     python_callable=insertData,
    #     provide_context=True,
    #     # op_kwargs={"name": "Dylan"}
    # )

    create_table = PostgresOperator(
        sql = create_table_sql_query,
        task_id = "create_table_task",
        postgres_conn_id = "postgres_rds",
    )

    delete_table = PostgresOperator(
        sql = delete_table_sql_query,
        task_id = "delete_table_task",
        postgres_conn_id = "postgres_rds",
    )

    # insert_data = PostgresOperator(
    #     sql = insert_data_sql_query,
    #     task_id = "insertData_sql_query_task",
    #     postgres_conn_id = "postgres_rds",
    # )



# getDrowToken >> getMongoDB >> reformData >> create_table
# create_table >> getDrowToken >> getMongoDB >> reformData >> insertData
# getDrowToken >> getMongoDB >> reformData >> insertData
# delete_table >> create_table >> getDrowToken >> getMongoDB
delete_table >> create_table >> getDrowToken >> getMongoDB >> getMongoDB1c >> getMongoDB1d >> getMongoDBSurveyCheck