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

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

    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 getMongoDB(**context):
    token = context.get("ti").xcom_pull(key="token")

    # 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"

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

    db = create_engine(conn_string)
    conn = db.connect()
    # print('db connected')
    df = pd.DataFrame()
    with conn as conn:
        # df_from_sql_nec_records=pd.read_sql("SELECT * FROM public.nec_c04;",
        #     conn)
        df_from_sql_nec_section_of_works = pd.read_sql("SELECT * FROM public.c5_nec_section_of_work;", conn)

        # Initialize an empty DataFrame
        df = pd.DataFrame()

        # Clean and convert the specific columns to float
        df['Cumulative_PWDD'] = df_from_sql_nec_section_of_works['Cumulative_PWDD'].str.replace(',', '').str.strip().astype(float)
        df['Forecast_of_the_final_Prices_for_the_Work_Done_to_Date__PWDD'] = df_from_sql_nec_section_of_works['Forecast_of_the_final_Prices_for_the_Work_Done_to_Date__PWDD'].str.replace(',', '').str.strip().astype(float)
        df['Forecast_Total_of_the_Prices']=df_from_sql_nec_section_of_works['Latest_Forecast_Total_of_the_Prices']
        df['Forecast_of_the_final_total_of_the_Prices'] = df_from_sql_nec_section_of_works['Forecast_of_the_final_total_of_the_Prices'].str.replace(',', '').str.strip().astype(float)
        
        
        # Calculate the percentage
        df['A1. Ratio of Price for Work Done to Date (PWDD) to forecast final PWDD'] = ((df['Cumulative_PWDD'] / df['Forecast_of_the_final_Prices_for_the_Work_Done_to_Date__PWDD']) * 100).round(2)
        df['A2. Ratio of forecast final Price for Work Done to Date (PWDD) to forecast final total of the Prices'] = ((df['Forecast_of_the_final_Prices_for_the_Work_Done_to_Date__PWDD'] / df['Forecast_of_the_final_total_of_the_Prices']) * 100).round(2)
        
        # Load data from SQL into DataFrame
        df_from_sql_nec_c5 = pd.read_sql("SELECT * FROM public.c5_nec_cas;", conn)
        # print(df_from_sql_nec_c5)

        # Convert 'Revised_Completion_Date' to datetime, errors='coerce' will handle None and invalid dates
        df_from_sql_nec_c5['Revised_Completion_Date'] = pd.to_datetime(df_from_sql_nec_c5['Revised_Completion_Date'], errors='coerce')

        # Find the latest 'Revised_Completion_Date'
        latest_row = df_from_sql_nec_c5.loc[df_from_sql_nec_c5['Revised_Completion_Date'].idxmax()]

        # Extract the latest 'Revised_Completion_Date'
        latest_date = latest_row['Revised_Completion_Date']

        # Assuming df_from_sql_nec_section_of_works is already defined and contains 'starting_date'
        df['contract_start_date'] = df_from_sql_nec_section_of_works['starting_date'].dt.tz_localize(None)

        # Assign the latest 'Revised_Completion_Date' to 'Longest Section / Key day' in df
        df['Longest Section / Key day'] = latest_date
        
        # Calculate today's date
        today = pd.to_datetime(datetime.today())

        # Calculate the time elapsed from contract start date to today
        df['Time Elapsed'] = (today - df['contract_start_date']).dt.days
        
        # Calculate the total contractual duration
        df['Contractual Duration'] = (df['Longest Section / Key day'].dt.tz_localize(None) - df['contract_start_date']).dt.days

        # Calculate the ratio of time elapsed to contractual duration as a percentage
        df['Ratio of Time Elapsed to Contractual Duration'] = ((df['Time Elapsed'] / df['Contractual Duration']) * 100).round(2)
        
        df_json = pd.DataFrame()
        # Convert date columns to datetime objects
        df_json['Key_Date'] = df_from_sql_nec_c5['Key_Date']
        df_json['Revised_Completion_Date'] = pd.to_datetime(df_from_sql_nec_c5['Revised_Completion_Date'])
        df_json['Ori_Completion_Date'] = pd.to_datetime(df_from_sql_nec_c5['Ori_Completion_Date'])
        
        # Calculate the latest "Revised_Completion_Date" for each section or key date
        latest_revised_completion = df_json.groupby(['Key_Date'])['Revised_Completion_Date'].max().reset_index()
        latest_revised_completion.rename(columns={'Revised_Completion_Date': 'Section or Key day Revised_Completion_Date'}, inplace=True)
        
        # Calculate the earliest "Ori_Completion_Date" for each section or key date
        earliest_ori_completion = df_json.groupby(['Key_Date'])['Ori_Completion_Date'].min().reset_index()
        earliest_ori_completion.rename(columns={'Ori_Completion_Date': 'Section or Key day Ori_Completion_Date'}, inplace=True)
        
        # Merge the latest and earliest dates into a single DataFrame
        merged_dates = pd.merge(latest_revised_completion, earliest_ori_completion, on=['Key_Date'])
        
        # Calculate the EOT (Extension Of Time) in days
        merged_dates['EOT'] = (merged_dates['Section or Key day Revised_Completion_Date'] - merged_dates['Section or Key day Ori_Completion_Date']).dt.days
        
        merged_dates = merged_dates[merged_dates['Key_Date'].notnull() & (merged_dates['Key_Date'] != '')]
        
        eot_dict = merged_dates.set_index('Key_Date')['EOT'].to_dict()
        
        df['B3. Extension of time of the contract'] = json.dumps(eot_dict)
        
        df_from_sql_PCD = pd.read_sql("SELECT * FROM public.c5_key_date_data;", conn)
        
        merged_PCD = pd.merge(df_from_sql_PCD, latest_revised_completion, left_on='key_Date', right_on='Key_Date', how='inner')
        
        merged_PCD = merged_PCD[['Key_Date', 'Section or Key day Revised_Completion_Date', 'Planned_Completion_Date_PCD']]
        
        merged_PCD['Section or Key day Revised_Completion_Date'] = merged_PCD['Section or Key day Revised_Completion_Date'].dt.strftime('%Y-%m-%d')
        merged_PCD['Planned_Completion_Date_PCD'] = pd.to_datetime(merged_PCD['Planned_Completion_Date_PCD']).dt.strftime('%Y-%m-%d')
        
        merged_PCD_json_string = merged_PCD.to_json(orient='records')
        
        df['B4/5. Completion date, planned completion, completion date'] = merged_PCD_json_string
        
        filtered_df_CEW = df_from_sql_nec_c5[
        (df_from_sql_nec_c5['NEC_Doc_Type'] == 'EW-') &
        (df_from_sql_nec_c5['From'].str.startswith('DCK JV')) &
        (df_from_sql_nec_c5['Doc_Ver'] == '0') | (df_from_sql_nec_c5['Doc_Ver'] == 0)
        ]

        # Get the total number of records that meet the conditions
        filtered_df_CEW_total_records = len(filtered_df_CEW)
        
        df['C1. Total Number of Early Warnings Initiated by Contractor'] = filtered_df_CEW_total_records
        
        filtered_df_PMEW = df_from_sql_nec_c5[(df_from_sql_nec_c5['NEC_Doc_Type'] == 'EW-') &
            (~df_from_sql_nec_c5['From'].str.startswith('DCK JV')) &
            (df_from_sql_nec_c5['Doc_Ver'] == '0') | (df_from_sql_nec_c5['Doc_Ver'] == 0)
            ]

        # Get the total number of records that meet the conditions
        filtered_df_PMEW_total_records = len(filtered_df_PMEW)
        
        df['C2. Total Number of Early Warnings Initiated by Project Manager'] = filtered_df_PMEW_total_records
        
        df['C3. Total Number of Early Warnings'] = filtered_df_CEW_total_records + filtered_df_PMEW_total_records
        
        filtered_df_Closed_EW = df_from_sql_nec_c5[(df_from_sql_nec_c5['NEC_Doc_Type'] == 'EW-') &
            ((df_from_sql_nec_c5['Doc_Ver'] == '0')|(df_from_sql_nec_c5['Doc_Ver'] == 0)) &
            (df_from_sql_nec_c5['record_status'] == 'Closed')
            ]
        
        if(filtered_df_CEW_total_records + filtered_df_PMEW_total_records)== 0 :
            df['C4. Ratio of Closed Early Warnings to Total Number of Early Warnings'] = 100
        else:
            df['C4. Ratio of Closed Early Warnings to Total Number of Early Warnings'] = round(((len(filtered_df_Closed_EW) / filtered_df_CEW_total_records + filtered_df_PMEW_total_records)*100),2)
        
        df_from_sql_risk_reg = pd.read_sql("SELECT * FROM public.c5_nec_risk_register;", conn)
        
        # Convert the relevant columns to datetime objects, handling errors
        df_from_sql_risk_reg['Date_of_Close_of_EW'] = pd.to_datetime(df_from_sql_risk_reg['Date_of_Close_of_EW'], errors='coerce')
        df_from_sql_risk_reg['Date_of_Early_Warning'] = pd.to_datetime(df_from_sql_risk_reg['Date_of_Early_Warning'], errors='coerce')

        # Filter out rows where either date is null
        filtered_rr_df = df_from_sql_risk_reg.dropna(subset=['Date_of_Close_of_EW', 'Date_of_Early_Warning'])

        # Calculate the difference in days between Date_of_Close_of_EW and Date_of_Early_Warning
        filtered_rr_df['Duration_Days'] = (filtered_rr_df['Date_of_Close_of_EW'] - filtered_rr_df['Date_of_Early_Warning']).dt.days

        # Calculate the average duration
        average_duration = filtered_rr_df['Duration_Days'].mean()
        
        df['C5. Average Duration to Resolve Early Warnings'] = average_duration
        
        filtered_df_PMI= df_from_sql_nec_c5[(df_from_sql_nec_c5['NEC_Doc_Type'] == 'PMI-') &
            (df_from_sql_nec_c5['Doc_Ver'] == '0') | (df_from_sql_nec_c5['Doc_Ver'] == 0)
            ]
        filtered_df_NCE= df_from_sql_nec_c5[(df_from_sql_nec_c5['NEC_Doc_Type'] == 'NCE-') &
            (df_from_sql_nec_c5['Doc_Ver'] == '0') | (df_from_sql_nec_c5['Doc_Ver'] == 0)
            ]
        filtered_df_PMN= df_from_sql_nec_c5[(df_from_sql_nec_c5['NEC_Doc_Type'] == 'PMN-') &
            (df_from_sql_nec_c5['Doc_Ver'] == '0') | (df_from_sql_nec_c5['Doc_Ver'] == 0)
            ]
        
        df['D1. Total Number of Project Manager’s Instructions']=len(filtered_df_PMI)
        df['D2. Total Number of Contractor’s Notified Compensation Events (NCE)']=len(filtered_df_NCE)
        df['D3. Total Number of Project Manager’s Notified Compensation Events']=len(filtered_df_PMN)
        
        # Initialize a counter for accepted records
        accepted_count = 0

        # Iterate through each filtered NCE record
        for _, row in filtered_df_NCE.iterrows():
            nec_event_no = row['NEC_Event_No']

            # Check if there's a 'QA-' record with the same NEC_Event_No
            if not df_from_sql_nec_c5[(df_from_sql_nec_c5['NEC_Event_No'] == nec_event_no) & (df_from_sql_nec_c5['NEC_Doc_Type'] == 'PMN-')].empty:
                accepted_count += 1
        df["D3a. Total Number of Contractor’s NCE which PM accepted and instructed for quotation or Total Number of Contractor's NCE with PM’s decision made"] = accepted_count
        
        df["D4. Total Number of Notified Compensation Events"]= len(filtered_df_NCE) + len(filtered_df_PMN)
        
        # Filter records where NEC_Clause starts with '60.1'
        filtered_ground_clause_df = df_from_sql_nec_c5[df_from_sql_nec_c5['NEC_Clause'].str.startswith('60.1')]

        # Function to extract the classification of ground from NEC_Clause
        def extract_classification(nec_clause):
            match = re.search(r'60\.1\((\d+)\)', nec_clause)
            if match:
                # print(int(match.group(1)))
                return int(match.group(1))
            return None

        # Apply the function to the filtered DataFrame
        filtered_ground_clause_df['Classification_of_Ground'] = filtered_ground_clause_df['NEC_Clause'].apply(extract_classification)
        

        # Drop rows where classification extraction failed (if any)
        filtered_ground_clause_df = filtered_ground_clause_df.dropna(subset=['Classification_of_Ground'])
        
        filtered_ground_clause_df = filtered_ground_clause_df[['Original_Doc_No', 'Classification_of_Ground']]
        
        filtered_ground_clause_df_json = filtered_ground_clause_df.to_json(orient='records')
        
        df['D5. Classification of Grounds for Implemented Compensation Events (NEC Clause 60.1)'] = filtered_ground_clause_df_json
        
        filtered_df_QA= df_from_sql_nec_c5[(df_from_sql_nec_c5['NEC_Doc_Type'] == 'QA-') &
            (df_from_sql_nec_c5['Doc_Ver'] == '0') |df_from_sql_nec_c5['Doc_Ver'] == 0
            ]
        if (len(filtered_df_NCE) + len(filtered_df_PMN)) ==0:
            df['D6. Ratio of Implemented Compensation Events to Notified Compensation Events']= 0
        else:
             df['D6. Ratio of Implemented Compensation Events to Notified Compensation Events']= round((len(filtered_df_QA) / (len(filtered_df_NCE) + len(filtered_df_PMN)))*100,2)
        # Group by NEC_Event_No and calculate the date difference
        def calculate_date_difference(group, ce_doc_type='NCE-', qa_doc_type='QA-'):
            nce_date = group.loc[group['NEC_Doc_Type'] == ce_doc_type, 'Doc_Date']
            qa_date = group.loc[group['NEC_Doc_Type'] == qa_doc_type, 'Doc_Date']

            if not nce_date.empty and not qa_date.empty:
                # Calculate the difference in days
                date_diff = (qa_date.iloc[0] - nce_date.iloc[0]).days
                return pd.Series({
                    'NEC_Event_No': group['NEC_Event_No'].iloc[0],
                    'CE_No': group['Original_Doc_No'].iloc[0],
                    'Date_Difference': date_diff
                })
            return None


        # # Apply the calculation to each group and filter out None results
        # NCE_QA_date_diff_df = df_from_sql_nec_c4.groupby('NEC_Event_No').apply(calculate_date_difference).dropna().reset_index(drop=True)
        # Apply the calculation to each group using a lambda function to pass parameters
        NCE_QA_date_diff_df = df_from_sql_nec_c5.groupby('NEC_Event_No').apply(lambda group: calculate_date_difference(group, 'NCE-', 'QA-')).dropna().reset_index(drop=True)
        
        df['D7. Average duration from Notification to Implementation of Compensation Events']=NCE_QA_date_diff_df['Date_Difference'].mean()
        
        PMN_CQS_date_diff_df = df_from_sql_nec_c5.groupby('NEC_Event_No').apply(lambda group: calculate_date_difference(group, 'PMN-', 'CSQ-')).dropna().reset_index(drop=True)
        
        df['D8. Average duration of Quotation Submission from Contractor']=PMN_CQS_date_diff_df['Date_Difference'].mean()
        
        CSQ_QA_date_diff_df = df_from_sql_nec_c5.groupby('NEC_Event_No').apply(lambda group: calculate_date_difference(group, 'CSQ-', 'QA-')).dropna().reset_index(drop=True)
        
        df['D9. Average duration of Quotation Assessment']=CSQ_QA_date_diff_df['Date_Difference'].mean()
        
        # print(NCE_QA_date_diff_df)
        
        
        if 'CE_Increase_Decrease' in filtered_df_QA.columns and 'CE_PMI_Amount' in filtered_df_QA.columns:
            # Calculate the Cost Implication
            def calculate_cost(row):
                if row['CE_Increase_Decrease'].lower() == 'increase':
                    return row['CE_PMI_Amount']
                elif row['CE_Increase_Decrease'].lower() == 'decrease':
                    return -row['CE_PMI_Amount']
                return 0

            filtered_df_QA['Cost_Implication'] = filtered_df_QA.apply(calculate_cost, axis=1)
            total_cost_implication = filtered_df_QA['Cost_Implication'].sum()
            filtered_df_QA_non_zero = filtered_df_QA[(filtered_df_QA['CE_PMI_Amount'] != 0)]
            df['D10. Cost Implication of Implemented Compensation Event'] = total_cost_implication
            df['D10a. Average cost implication of implemented compensation events'] =  total_cost_implication / len(filtered_df_QA_non_zero)
        
        filtered_df_QA_Time = filtered_df_QA[['Original_Doc_No', 'Key_Date', 'Revised_Completion_Date', 'Ori_Completion_Date']]
        
        # Drop rows where 'Revised_Completion_Date' or 'Ori_Completion_Date' is empty
        filtered_df_QA_Time = filtered_df_QA_Time.dropna(subset=['Revised_Completion_Date', 'Ori_Completion_Date'])

        # Convert the date columns to datetime objects
        filtered_df_QA_Time['Revised_Completion_Date'] = pd.to_datetime(filtered_df_QA_Time['Revised_Completion_Date']).dt.strftime('%Y-%m-%d')
        filtered_df_QA_Time['Ori_Completion_Date'] = pd.to_datetime(filtered_df_QA_Time['Ori_Completion_Date']).dt.strftime('%Y-%m-%d')
        
        df['D11. Time Implication of Implemented Compensation Events'] = filtered_df_QA_Time.to_json(orient='records')
        
        # filtered_df_QA_amount = filtered_df_QA[['Original_Doc_No', 'CE_PMI_Amount', 'CE_Increase_Decrease']]
        
        # ['D10. Cost Implication of Implemented Compensation Event'] = filtered_df_QA_amount.to_json(orient='records')
        
        # filtered_df_QA_non_zero = 
        
        # ['D10a. Average cost implication of implemented compensation events'] = 
        
        
        
        df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')

        # Write the DataFrame back to a SQL table
        df.to_sql('nec_c05_icwp', con=conn, if_exists='replace', index=False)
        # for x in Data:
        #     try:
        #         if len(x['data'].keys()) == 0:
        #             continue
        #         df_nested_list = json_normalize(x['data'])
        #         # print('process 1')
        #         # print(x['data'].keys())
        #         df2 = df_nested_list.reindex(columns=Mapping.keys())
        #         df2['record_status'] = x['Status']
        #         df2['NEC Doc Title']=x['data']['NEC Doc Type']+x['data']['NEC Event No.']
        #         df2['Doc Org Ver']= x['data']['Doc Ver.']
        #         y=0
        #         if x['data']['Doc Ver.'] == None:
        #             df2['Doc Ver.'] = y
        #         elif x['data']['Doc Ver.'].startswith('Rev. '):
        #             y = x['data']['Doc Ver.'].replace('Rev. ', '')
        #             y = int(y)
        #         else:
        #             y = x['data']['Doc Ver.'].replace('-', '').replace('r','')
        #             if y!='' and not y.isnumeric():
        #                 # print('ver',y)
        #                 y = int(ord(y)) - int(ord('A')) + 1
        #                 # print(y)
        #             elif y =='':
        #                 y = 0
        #             else :
        #                 y = int(y)
        #             df2['Doc Ver.'] = y
        #         if (not df2['NEC Doc Title'].empty and 'NEC Doc Title' in df.columns):
        #             # print (y)
        #             # print('NEC Doc Title' in df.columns)
        #             check_ver_df = df.loc[(df['NEC Doc Title'] == x['data']['NEC Doc Type']+x['data']['NEC Event No.'])]
        #             if check_ver_df.empty:
        #                 df2['is_latest'] = 'Yes'
        #             else :
        #                 check_ver_df2 = check_ver_df.loc[(check_ver_df['Doc Ver.'] > y)]
        #                 if not check_ver_df2.empty:
        #                     df2['is_latest'] = "No"
        #                 else:
        #                     df.loc[(df['NEC Doc Title'] == x['data']['NEC Doc Type']+x['data']['NEC Event No.']) & ( df['Doc Ver.']<y), 'is_latest'] = 'No'
        #                     df2['is_latest'] = 'Yes'
        #         else:
        #             df2['is_latest'] = 'Yes'
        #         df2['NEC Doc Title With Version']=x['data']['NEC Doc Type']+x['data']['NEC Event No.']+'-'+str(y)

        #         if (x['data'].get('NEC Doc Type') or '').strip().upper() == 'PMN-' and y==0:
        #             df2['From_Status'] = 'CE notified'
        #         elif (x['data'].get('NEC Doc Type') or '').strip().upper() == 'CSQ-' and y==0:
        #             df2['From_Status'] = 'Quotation / Revised quotation submitted by Contractor'
        #         elif (x['data'].get('NEC Doc Type') or '').strip().upper() == 'QA-' and y==0:
        #             df2['From_Status'] = 'CE implemented'
        #         else:
        #             df2['From_Status'] = None

        #         if len(x['data']['Change to Time'])>0 and x['data']['NEC Doc Type']!='EW-':
        #             df4=pd.DataFrame()
        #             for change_to_time_table in x['data']['Change to Time']:
        #                 df3=df2.copy()
        #                 i = i+1
        #                 if 'Key Date' in change_to_time_table:
        #                     df3['Key Date'] = change_to_time_table['Key Date']
        #                 if 'Extension in days' in change_to_time_table:
        #                     df3['Extension in days'] = change_to_time_table['Extension in days']
        #                 if 'Ori Completion Date' in change_to_time_table:
        #                     df3['Ori Completion Date'] = change_to_time_table['Ori Completion Date']
        #                 if 'Revised Completion Date' in change_to_time_table:
        #                     df3['Revised Completion Date'] = change_to_time_table['Revised Completion Date']
        #                 if i >0:
        #                     df2['From_Status'] = None
        #                 df4 = df4.append(df3)
        #             i = 0
        #             df2 = df2.iloc[0:0]
        #             df2=df2.append(df4)
        #             # print('process 2')
        #             # print('loading into DB')
        #         df = df.append(df2)
        #     except:
        #         continue
        # # df['is_latest'].fillna('No',inplace=True)
        # df.rename(columns=Mapping, inplace=True)
        # fields_to_adjust = ['Doc_Date', 'Ori Completion Date', 'Revised Completion Date']

        # for field in fields_to_adjust:
        #     if field in df.columns:
        #         df[field] = df[field].apply(pd.to_datetime)
        #         df[field] = df[field] - pd.Timedelta(hours=8)
        # # df['Doc_Date']=df['Doc_Date'].apply(pd.to_datetime)
        # # df['Doc_Date'] = df['Doc_Date'] - pd.Timedelta(hours=8)
        # # df['Ori Completion Date']=df['Ori Completion Date'].apply(pd.to_datetime)
        # # df['Ori Completion Date'] = df['Ori Completion Date'] - pd.Timedelta(hours=8)
        # # df['Revised Completion Date']=df['Revised Completion Date'].apply(pd.to_datetime)
        # # df['Revised Completion Date'] = df['Revised Completion Date'] - pd.Timedelta(hours=8)
        # df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
        # df.to_sql('nec_c04', con=conn, if_exists='replace', index= False)
    
    # print("success")

# */2 * * * * Execute every two minute 
with DAG(
        dag_id="1test__nec_c5_icwp",
        schedule_interval="0 0,4,8,11,16 * * *",
        default_args={
            "owner": "airflow",
            "retries": 1,
            "retry_delay": timedelta(minutes=5),
            "start_date": datetime(2022, 10, 24)
        },
        catchup=False) as f:
    
    getDataAndSendToPSQL = PythonOperator(
        task_id="getDataAndSendToPSQL",
        python_callable=getMongoDB,
        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"}
    )

getDrowToken >> getDataAndSendToPSQL