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 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 = SimpleHttpOperator(
    #     task_id="getDrowToken",
    #     http_conn_id="getDrowToken",
    #     endpoint="https://uat2.drow.cloud/api/auth/authenticate", 
    #     method="POST",
    #     data={
    #     "username": "icwp2@drow.cloud",
    #     "password": "dGVzdDAxQHRlc3QuY29t"
    #     },
    #     xcom_push=True,
    # )

    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")
    response = requests.get(
    url=f"{dRoW_api_end_url}/api/module/document-export/airflow/workflow/64dae7661b96600c895fd6d9?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 ("B3 Agreed Due Date for Completion" in summaryData and "B3 Agreed Due Date for Completion" in summaryData and not (summaryData["B3 Agreed Due Date for Completion"]!='') and (not (summaryData["B4 Date Completed"]!='')) and (summaryData["B3 Agreed Due Date for Completion"].astype('datetime64[ns]') < summaryData["B4 Date Completed"].astype('datetime64[ns]')).bool()):
                        total_late_retification += 1
                df2['total_late_retification'] = total_late_retification
            else:
                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':
                                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':
                                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_dc202106', con=conn, if_exists='replace', index= False)


# */2 * * * * Execute every two minute 
with DAG(
        dag_id="dc202106_safety_walk",
        schedule_interval="0 15 * * *",
        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,
    )

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

    # 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
# create_table >> 
getDrowToken >> getMongoDB
    
            # if df2['F2. 監理代表 Checked by SupD on Date'].isnull().bool():
            #     df2['report_complete_or_incomplete'] = 'incomplete'
            # else:
            #     df2['report_complete_or_incomplete'] = 'complete'
            # if not df2['F2. 監理代表 Checked by SupD on Date'].isnull().bool() and not df2['D4 Submission Date'].isnull().bool() and (df2["F2. 監理代表 Checked by SupD on Date"].astype('datetime64[ns]') < df2["D4 Submission Date"].astype('datetime64[ns]')).bool():
            #     df2['nc_report'] = True
            # else:
            #     df2['nc_report'] = False
            # if (not df2['F2. 監理代表 Checked by SupD on Date'].isnull().bool() and not df2['A1a 巡查日期 Inspection Date'].isnull().bool()):
            #     df2['complete_time_in_days'] = (((df2['F2. 監理代表 Checked by SupD on Date'].astype('datetime64[ns]') - 
            #     df2['A1a 巡查日期 Inspection Date'].astype('datetime64[ns]'))/ np.timedelta64(1, 'h'))/24).round(2)
            #     if df2['complete_time_in_days'].isnull().bool() or df2['complete_time_in_days'].lt(0).bool():
            #         df2['complete_time_in_days'] = 0
            # else:
            #     df2['complete_time_in_days'] = 0
            # if(not df2["C02 - Survey Checked on Date Time"].isnull().bool() and not df2["A10 - Request Submission Date Time"].isnull().bool() and ((df2["C02 - Survey Checked on Date Time"].astype('datetime64[ns]') < df2["A10 - Request Submission Date Time"].astype('datetime64[ns]')).bool())):
            #     # print((df2["C02 - Survey Checked on Date Time"].astype('datetime64[ns]') < df2["A10 - Request Submission Date Time"].astype('datetime64[ns]')))
            # else:
            #     # print((df2["C02 - Survey Checked on Date Time"].astype('datetime64[ns]') < df2["A10 - Request Submission Date Time"].astype('datetime64[ns]')))

            # if (not df2["C02 - Survey Checked on Date Time"].isnull().bool() and (not df2["A10 - Request Submission Date Time"].isnull().bool()) and (df2["C02 - Survey Checked 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["C02 - Survey Checked on Date Time"].isnull().bool() and not df2["A10 - Request Submission Date Time"].isnull().bool() and (((df2["C02 - Survey Checked on Date Time"].astype('datetime64[ns]') - df2["A10 - Request Submission Date Time"].astype('datetime64[ns]'))) < pd.Timedelta(24, unit='h')).bool()):
            #     df2['urgent_report'] = True
            # else:
            #     df2['urgent_report'] = False

            # if (not df2['E01 - Received on behalf of Contractor on Date Time'].isnull().bool() and not df2['A10 - Request Submission Date Time'].isnull().bool()):
            #     df2['elapsed_time'] = (((df2['E01 - Received on behalf of Contractor on Date Time'].astype('datetime64[ns]') - 
            #     df2['A10 - Request Submission Date Time'].astype('datetime64[ns]'))/ np.timedelta64(1, 'h'))/24).round(2)
            #     if df2['elapsed_time'].isnull().bool() or df2['elapsed_time'].lt(0).bool():
            #         df2['elapsed_time'] = 0
            # else:
            #     df2['elapsed_time'] = 0
            
            # if (not df2["C02 - Survey Checked on Date Time"].isnull().bool() and not df2["D01 - Countersigned on Date Time"].isnull().bool() and (((df2["D01 - Countersigned on Date Time"].astype('datetime64[ns]') - df2["C02 - Survey Checked on Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
            #     df2['overdue_report'] = True
            # else:
            #     df2['overdue_report'] = False
            
            # if (not df2["C02 - Survey Checked on Date Time"].isnull().bool() and not df2["C12 Time Pass to Senior or Contractor"].isnull().bool() and (((df2["C12 Time Pass to Senior or Contractor"].astype('datetime64[ns]') - df2["C02 - Survey Checked on Date Time"].astype('datetime64[ns]')))>= pd.Timedelta(24, unit='h')).bool()):
            #     df2['delayed_approval_report'] = True
            # else:
            #     df2['delayed_approval_report'] = False

            # if ((df2['A01a - Request No. Revision']=="-A").bool()):
            #     df2['fail_in_first_inspection'] = True
            # else:
            #     df2['fail_in_first_inspection'] = False
            
            # if (not df2["C12 Time Pass to Senior or Contractor"].isnull().bool() and not df2["C02 - Survey Checked on Date Time"].isnull().bool() and not df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()):
            #     df2['complete_incomplete_outstanding_report'] = 'complete'
            # elif ((df2["C12 Time Pass to Senior or Contractor"].isnull().bool() or df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()) and not df2["C02 - Survey Checked on Date Time"].isnull().bool()):
            #     df2['complete_incomplete_outstanding_report'] = 'in-complete'
            # elif ((not df2["C12 Time Pass to Senior or Contractor"].isnull().bool() or df2["E01 - Received on behalf of Contractor on Date Time"].isnull().bool()) and df2["C02 - Survey Checked on Date Time"].isnull().bool()):
            #     df2['complete_incomplete_outstanding_report'] = 'outstanding'
            # else:
            #     df2['complete_incomplete_outstanding_report'] = 'outstanding'