Airflow - DAGs

Dependency	Reason
Dagrun Running	Task instance's dagrun was not in the 'running' state but in the state 'failed'.
Trigger Rule	Task's trigger rule 'all_success' requires all upstream tasks to have succeeded, but found 1 non-success(es). upstream_tasks_state={'total': 1, 'successes': 0, 'skipped': 0, 'failed': 1, 'upstream_failed': 0, 'done': 1}, upstream_task_ids={'getDrowToken'}
Task Instance State	Task is in the 'upstream_failed' state which is not a valid state for execution. The task must be cleared in order to be run.
Attribute: python_callable

def pipelineProcess(**context):
    token = context.get("ti").xcom_pull(key="token")

    # Contract Data
    contract_data = getSheetData(token, "66c5b378b9d9c90706ebdde7")
    # "Section and key date data"
    section_and_key_data = getSheetData(token, "66c5b378b9d9c90706ebddeb")
    
    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()

    df = pd.DataFrame()
    _df = pd.DataFrame()
    with conn as conn:
        if len(contract_data):
            Mappings = {
                'starting date': 'starting date',
                'ori comp date': 'ori comp date'
            }

            for x in contract_data:
                df_nested_list = json_normalize(x)
                df2 = df_nested_list
                df = df.append(df2)
            df.rename(columns=Mappings, inplace=True)
            df['starting date']=df['starting date'].apply(pd.to_datetime)
            df['ori comp date']=df['ori comp date'].apply(pd.to_datetime)
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(df.empty):
                df.to_sql('cv202303_nec_section_of_work', con=conn, if_exists='replace', index= False)
        
        if len(section_and_key_data):
            Mappings = {
                'Starting Date': 'Starting Date',
                'Original Completion Dates': 'Original completion dates'
            }

            for x in section_and_key_data:
                df_nested_list = json_normalize(x)
                df2=df_nested_list
                _df = _df.append(df2)
            _df.rename(columns=Mappings, inplace=True)
            _df['Starting Date']=_df['Starting Date'].apply(pd.to_datetime)
            _df['Original completion dates']=_df['Original completion dates'].apply(pd.to_datetime)
            _df.columns = _df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(_df.empty):
                _df.to_sql('cv202303_nec_section_of_work_key_date', con=conn, if_exists='replace', index= False)
    

    # PWDD and Target Cost with Actual Monthly Total
    pwdd_data = getSheetData(token, "66c5b378b9d9c90706ebdde8")
    df = pd.DataFrame.from_dict(pwdd_data)
    df['IP No.']=df['IP No.'].astype(str)
    numerics = df.select_dtypes(include="number").columns
    df=df.apply(pd.to_numeric, errors='ignore')
    df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))

    df = df.dropna(subset=['IP No.']) 
    df = df.dropna(subset=['Month - Year']) 
    df['Month - Year']=df['Month - Year'].apply(pd.to_datetime)
    df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
    df = df.drop_duplicates(subset=['IP_No'], keep='first')
    
    db = create_engine(conn_string)
    conn = db.connect()
    with conn as conn:
        if not(df.empty):
            df.to_sql('cv202303_finance_data', con=conn, if_exists='replace')
    conn.close()


    # Payment
    forecast_data = getSheetData(token, "66c6b003ccbba0948f6ba812")

    if len(forecast_data):
        df = pd.DataFrame.from_dict(forecast_data)
        numerics = df.select_dtypes(include="number").columns
        df=df.apply(pd.to_numeric, errors='ignore')
        df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))
        df['Month - Year']=df['Month - Year'].apply(pd.to_datetime)
        df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
        db = create_engine(conn_string)
        conn = db.connect()
        with conn as conn:
            if not(df.empty):
                df.to_sql('cv202303_finance_status_data', con=conn, if_exists='replace')
        conn.close()



    # # EOT DATA
    # data = getSheetData(token, "64a68a0d9207b30c741cd79a")
    # df = pd.DataFrame.from_dict(data)
    # numerics = df.select_dtypes(include="number").columns
    # df=df.apply(pd.to_numeric, errors='ignore')
    # df[numerics]=df[numerics].apply(lambda x: np.round(x, decimals=5))
    # df['Month - Year']=df['Month - Year'].apply(pd.to_datetime)
    # df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
    # db = create_engine(conn_string)
    # conn = db.connect()
    # with conn as conn:
    #     df.to_sql('cv202303_eot_data', con=conn, if_exists='replace')
    # conn.close()



    # Programme Data
    programme_data = getSheetData(token, "66c5b378b9d9c90706ebddea")
    if programme_data:
        df = pd.DataFrame.from_dict(programme_data)
        df['Submission Date']=df['Submission Date'].apply(pd.to_datetime)
        df['Acceptance Date']=df['Acceptance Date'].apply(pd.to_datetime)
        df['Programme Approval Elapsed Time (days)']=(df['Acceptance Date'] - df['Submission Date']).dt.days
        df=df.drop('Attachment', axis=1)
        df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
        db = create_engine(conn_string)
        conn = db.connect()
        with conn as conn:
            df.to_sql('cv202303_nec_programme_data', con=conn, if_exists='replace')
        conn.close()



    # # key date Planned Completion Date (PCD)
    # pcd_data = getSheetData(token, "66c5b378b9d9c90706ebddec")
    # df = pd.DataFrame.from_dict(pcd_data)
    # df['Planned Completion Date(PCD)']=df['Planned Completion Date(PCD)'].apply(pd.to_datetime)
    # df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent')
    # db = create_engine(conn_string)
    # conn = db.connect()
    # with conn as conn:
    #     df.to_sql('cv202303_key_date_data', con=conn, if_exists='replace')
    # conn.close()
     

    # CR_Event
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddcb")
    Mapping= {
            "Incident No.": "incident_No",
            "Event Subject": "event_Subject",
            "Initiated by": "initiated_by",
            "Initiate Date": "initiate_Date",
            "Status": "status",
            "Remarks": "remarks",
    }

    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    df = pd.DataFrame()
    with conn as conn:
        for x in _Data:
            try:
                if len(x['data'].keys()) == 0:
                    continue
                # Skip if incident_No is empty
                if not x['data']['Incident No.']:
                    continue
                df_nested_list = json_normalize(x['data'])
                df2 = df_nested_list.reindex(columns=Mapping.keys())
                df = df.append(df2)
            except Exception as e:
                print(e)
                continue

        df.rename(columns=Mapping, inplace=True)
        if 'initiate_date' in df:
            df['initiate_date'] = df['initiate_date'].apply(pd.to_datetime)
            df['initiate_date'] = df['initiate_date'] - pd.Timedelta(hours=8)
        else:
            df['initiate_date'] = None

        df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
        if not(df.empty):
            df.to_sql('cv202303_nec_cr_event', con=conn, if_exists='replace', index= False)
    conn.close()     

    # CR_C1 - Contractor's Early Warning Notification
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddcc")
    
    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect() 
        df = pd.DataFrame()
        Mappings = {
            "Date": "Doc Date",
        }

        with conn as conn:
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            df.rename(columns=Mappings, inplace=True)
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')

            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)
            
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            df['EW_Category_Clause__151'] = list(map(lambda x: json.dumps(x), df['EW_Category__Clause_151']))
            df['NEC_Clause'] = list(map(lambda x: json.dumps(x), df['NEC_Clause']))
            df = df.applymap(convert_to_json)
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_c1', con=conn, if_exists='replace', index= False)
            
        conn.close()

    # CR_C2 - Contractor's Notification of Compensation Event
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddcd")

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()

        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')

            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

            df.drop(['Attachments'], axis=1, inplace=True)
            df = df.applymap(convert_to_json)
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_c2', con=conn, if_exists='replace', index= False)

        conn.close()

    
    # CR_C3 - QS Submission of Quotation
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddce")
    Mappings = {
        'CSQ No.': 'CE No.'
    }

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue

            df.rename(columns=Mappings, inplace=True)
            df['Original NEC Form No'] = df['CE No.']
            df['NEC Form No'] = df.apply(lambda row: 'QS-' + str(row['CE No.']) if pd.notnull(row['CE No.']) else row['CE No.'],axis=1)
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')
            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

            df = df.applymap(convert_to_json)
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_c3', con=conn, if_exists='replace', index= False)
            
        conn.close()

    
    # CR_C4 - Communication
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddcf")
    
    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        Mappings = {
            "Date": "Doc Date"
        }

        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0 or len(x['data']) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            df.rename(columns=Mappings, inplace=True)
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')

            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)        

            df = df.applymap(convert_to_json)
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_c4', con=conn, if_exists='replace', index= False)
            
        conn.close()
    
    # CR_E1 - Employer's Early Warning Notification
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddd0")

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')

            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

            df = df.applymap(convert_to_json)
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            
            if not df.empty:
                df.to_sql('cv202303_nec_cr_e1', con=conn, if_exists='replace', index= False)
            
        conn.close()
    
    # CR_P1 - PM's Early Warning Notification
    _Data = getWorkflowData(token, "66d12fe5444a299f46e7ad17")

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        Mappings = {
            "Date": "Doc Date"
        }

        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')

            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

            if 'Attachments' in df.columns:
                df.drop(["Attachments"], axis=1, inplace=True)
            df = df.applymap(convert_to_json)
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_p1', con=conn, if_exists='replace', index= False)
            
        conn.close()
    
    # CR_P2 - Project Manager's Instruction
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddd2")

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')

            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

            df = df.applymap(convert_to_json)
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_p2', con=conn, if_exists='replace', index= False)
            
        conn.close()
    
    # CR_P3 - PM's Notification of Compensation Event
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddd3")

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        Mappings = {
            "NCE No.:": "CE No.",
            "Date": "Doc Date"
        }

        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            df.rename(columns=Mappings, inplace=True)
            if 'CE No.' in df.columns:
                df['Original NEC Form No'] = df['CE No.']
                df['NEC Form No'] = df.apply(lambda row: 'CE-' + str(row['CE No.']) if pd.notnull(row['CE No.']) else row['CE No.'],axis=1)
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')
            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            df = df.applymap(convert_to_json)
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_p3', con=conn, if_exists='replace', index= False)
            
        conn.close()
    
    # CR_P4 - PM's Reply to Quotation
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddd4")

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        Mappings = {
            "RQ No.:": "CE No."
        }

        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])

                    if len(x['data']['Change to Time'])>0:
                        df4=pd.DataFrame()
                        for change_to_time_table in x['data']['Change to Time']:
                            df3=df_nested_list.copy()
                            if 'Key Date' in change_to_time_table:
                                df3['Key Date'] = change_to_time_table['Key Date']
                            else:
                                df3['Key Date'] = ''
                            if 'Extension in days' in change_to_time_table:
                                df3['Extension in days'] = change_to_time_table['Extension in days']
                            else:
                                df3['Extension in days'] = 0
                            if 'Ori Completion Date' in change_to_time_table:
                                df3['Ori Completion Date'] = change_to_time_table['Ori Completion Date']
                                if not change_to_time_table['Ori Completion Date']==None:
                                    df3['Ori Completion Date'] = df3['Ori Completion Date'].apply(pd.to_datetime) - pd.Timedelta(hours=8)
                            if 'Revised Completion Date' in change_to_time_table:
                                df3['Revised Completion Date'] = change_to_time_table['Revised Completion Date']
                                if not change_to_time_table['Ori Completion Date']==None:
                                    df3['Revised Completion Date'] = df3['Revised Completion Date'].apply(pd.to_datetime) - pd.Timedelta(hours=8)
                            df4 = df4.append(df3)
                        df_nested_list = df_nested_list.iloc[0:0]
                        df_nested_list = df_nested_list.append(df4)

                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue

            if not(df.empty):
                df.rename(columns=Mappings, inplace=True)
                df['Original NEC Form No'] = df['CE No.']
                df['NEC Form No'] = df.apply(lambda row: 'RTQ-' + str(row['CE No.']) if pd.notnull(row['CE No.']) else row['CE No.'],axis=1)
                if 'Doc Date' in df.columns:
                    df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')
                if 'cc' in df.columns:
                    # Serialize list or dict to JSON string
                    df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                    # Replace JSON array brackets with PostgreSQL array braces
                    df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

                df = df.applymap(convert_to_json)        
                df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
                if 'Extension_in_days' in df.columns:
                    df['Extension_in_days'] = pd.to_numeric(df['Extension_in_days'], errors='coerce')
                df.to_sql('cv202303_nec_cr_p4', con=conn, if_exists='replace', index= False)
            
        conn.close()


    # CR_P5 - Project Manager's Notification
    _Data = getWorkflowData(token, "66c5b376b9d9c90706ebddd5")

    if len(_Data) > 0:
        conn_string = getdrowPSQLConnectionString()
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        with conn as conn: 
            for x in _Data:
                try:
                    if len(x['data'].keys()) == 0:
                        continue
                    df_nested_list = json_normalize(x['data'])
                    df = df.append(df_nested_list)
                    
                except Exception as e:
                    print(e)
                    continue
            
            if 'Doc Date' in df.columns:
                df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')
            if 'cc' in df.columns:
                # Serialize list or dict to JSON string
                df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
                # Replace JSON array brackets with PostgreSQL array braces
                df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

            df = df.applymap(convert_to_json)        
            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
            if not(df.empty):
                df.to_sql('cv202303_nec_cr_p5', con=conn, if_exists='replace', index= False)
            
        conn.close()
    
    # # CR_P5
    # _Data = getWorkflowData(token, "65c44bd2f05ca6e49db707db")
    # Mapping= {
    #         "CE No.": "NEC Form No",
    #         "Incident No.": "Incident No",
    #         "Doc Ver.": "Doc_Ver",
    #         "Doc Date": "Doc_Date",
    #         "Subject": "Subject",
    #         "NEC Clause": "NEC_Clause",
    #         "From": "From",
    #         "From Role": "From Role",
    #         "To": "To",
    #         "To Role": "To Role",
    #         "CE Amount": "CE_Amount",
    #         "CE Increase / Decrease": "CE_Increase_Decrease",
    # }
    
    # conn_string = getdrowPSQLConnectionString()
    # db = create_engine(conn_string)
    # conn = db.connect()
    # df = pd.DataFrame()
    # with conn as conn: 
    #     for x in _Data:
    #         try:
    #             if len(x['data'].keys()) == 0:
    #                 continue
    #             df_nested_list = json_normalize(x['data'])
    #             df2 = df_nested_list.reindex(columns=Mapping.keys())
    #             df2['CE No.'] = df2.apply(lambda row: 'QA-' + str(row['CE No.']) if pd.notnull(row['CE No.']) else row['CE No.'],axis=1)

    #             if len(x['data']['Change to Time'])>0:
    #                 df4=pd.DataFrame()
    #                 for change_to_time_table in x['data']['Change to Time']:
    #                     df3=df2.copy()
    #                     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 not change_to_time_table['Ori Completion Date']==None:
    #                             df3['Ori Completion Date'] = df3['Ori Completion Date'].apply(pd.to_datetime) - pd.Timedelta(hours=8)
    #                     if 'Revised Completion Date' in change_to_time_table:
    #                         df3['Revised Completion Date'] = change_to_time_table['Revised Completion Date']
    #                         if not change_to_time_table['Ori Completion Date']==None:
    #                             df3['Revised Completion Date'] = df3['Revised Completion Date'].apply(pd.to_datetime) - pd.Timedelta(hours=8)
    #                     df4 = df4.append(df3)
    #                 df2 = df2.iloc[0:0]
    #                 df2=df2.append(df4)
    #             df = df.append(df2)
    #         except Exception as e:
    #             print(e)
    #             continue
        
    #     df.rename(columns=Mapping, inplace=True)

    #     df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
    #     if not(df.empty):
    #         df.to_sql('cv202303_nec_cr_p5', con=conn, if_exists='replace', index= False)
        
    # conn.close()
    
    # CR_P6: Reply to Contractor's Notification of Compensation
    _Data = getWorkflowData(token, "66d1b5a2f16ce0e01331faec")

    conn_string = getdrowPSQLConnectionString()
    db = create_engine(conn_string)
    conn = db.connect()
    df = pd.DataFrame()
    with conn as conn: 
        for x in _Data:
            try:
                if len(x['data'].keys()) == 0:
                    continue
                df_nested_list = json_normalize(x['data'])
                df = df.append(df_nested_list)
                
            except Exception as e:
                print(e)
                continue
        
        if 'Doc Date' in df.columns:
            df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')
        if 'cc' in df.columns:
            # Serialize list or dict to JSON string
            df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
            # Replace JSON array brackets with PostgreSQL array braces
            df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

        df = df.applymap(convert_to_json)        
        df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
        if not(df.empty):
            df.to_sql('cv202303_nec_cr_p6', con=conn, if_exists='replace', index= False)
        
    conn.close()
    
    # # CR_P7
    # _Data = getWorkflowData(token, "65c44e95ebb3aa38d3e12d51")

    # conn_string = getdrowPSQLConnectionString()
    # db = create_engine(conn_string)
    # conn = db.connect()
    # df = pd.DataFrame()
    # with conn as conn: 
    #     for x in _Data:
    #         try:
    #             # print(x['data'])
    #             if len(x['data'].keys()) == 0:
    #                 continue
    #             df_nested_list = json_normalize(x['data'])
    #             df = df.append(df_nested_list)
                
    #         except Exception as e:
    #             print(e)
    #             continue

    #     if 'Doc Date' in df.columns:
    #         df['Doc Date'] = pd.to_datetime(df['Doc Date'], errors='coerce')
    #     if 'cc' in df.columns:
    #         # Serialize list or dict to JSON string
    #         df['cc'] = df['cc'].apply(lambda x: json.dumps(x) if isinstance(x, (list, dict)) else x)
    #         # Replace JSON array brackets with PostgreSQL array braces
    #         df['cc'] = df['cc'].apply(lambda x: x.replace('"', '').replace('[', '{').replace(']', '}') if isinstance(x, str) else x)

    #     df = df.applymap(convert_to_json)        
    #     df.columns = df.columns.str.replace(' ', '_').str.replace('.', '').str.replace('(', '_').str.replace(')', '').str.replace('%', 'percent').str.replace('/', '_')
    #     if not(df.empty):
    #         df.to_sql('cv202303_nec_cr_p7', con=conn, if_exists='replace', index= False)
        
    # conn.close()

    # Risk Registry
    resData = getWorkflowData(token, "66c5b376b9d9c90706ebddbe")
    Mappings = {
        "Date notified for EW": "Date of Early Warning",
        "Closing Date": "Date of close of EW",
        "Status (Live / Closed)": "Status",
        "Level of Risk": "Risk Probability",
    }

    if len(resData) > 0:
        db = create_engine(conn_string)
        conn = db.connect()
        df = pd.DataFrame()
        with conn as conn:
            for x in resData:
                df_nested_list = json_normalize(x['data'])
                df2 = df_nested_list

                df = df.append(df2)
            
            df.rename(columns=Mappings, inplace=True)
            
            df['Date of Early Warning'] = df['Date of Early Warning'].apply(pd.to_datetime)
            df['Date of Early Warning'] = df['Date of Early Warning'] - pd.Timedelta(hours=8)
            df['Date of close of EW'] = df['Date of close of EW'].apply(pd.to_datetime)
            df['Date of close of EW'] = df['Date of close of EW'] - pd.Timedelta(hours=8)

            df.columns = df.columns.str.replace(' ', '_').str.replace('.', '_').str.replace('(', '_').str.replace(')', '').str.replace('/', '_').str.replace('%', 'percent')
            df['Risk_Level__Low__Medium__High'] = df.apply(lambda row: 'closed' if row['Status'] == 'Closed'
            else 'TBC' if pd.isna(row['Risk_Probability']) or row['Risk_Probability'] == '' 
            else row['Risk_Probability'], axis=1)

            if not(df.empty):
                df.to_sql('cv202303_nec_risk_register', con=conn, if_exists='replace', index= False)
        conn.close()
Task Instance Attributes

Attribute	Value
dag_id	cv202303_nec
duration	None
end_date	2024-11-15 16:07:40.768564+00:00
execution_date	2024-11-15T11:00:00+00:00
executor_config	{}
generate_command	<function TaskInstance.generate_command at 0x7f152f9bf320>
hostname
is_premature	False
job_id	None
key	('cv202303_nec', 'pipelineProcess', <Pendulum [2024-11-15T11:00:00+00:00]>, 1)
log	<Logger airflow.task (INFO)>
log_filepath	/usr/local/airflow/logs/cv202303_nec/pipelineProcess/2024-11-15T11:00:00+00:00.log
log_url	http://localhost:8080/admin/airflow/log?execution_date=2024-11-15T11%3A00%3A00%2B00%3A00&task_id=pipelineProcess&dag_id=cv202303_nec
logger	<Logger airflow.task (INFO)>
mark_success_url	http://localhost:8080/success?task_id=pipelineProcess&dag_id=cv202303_nec&execution_date=2024-11-15T11%3A00%3A00%2B00%3A00&upstream=false&downstream=false
max_tries	1
metadata	MetaData(bind=None)
next_try_number	1
operator	None
pid	None
pool	default_pool
prev_attempted_tries	0
previous_execution_date_success	2024-11-14 16:00:00+00:00
previous_start_date_success	2024-11-15 00:01:20.757772+00:00
previous_ti	<TaskInstance: cv202303_nec.pipelineProcess 2024-11-15 08:00:00+00:00 [upstream_failed]>
previous_ti_success	<TaskInstance: cv202303_nec.pipelineProcess 2024-11-14 16:00:00+00:00 [success]>
priority_weight	1
queue	default
queued_dttm	None
raw	False
run_as_user	None
start_date	2024-11-15 16:07:40.768542+00:00
state	upstream_failed
task	<Task(PythonOperator): pipelineProcess>
task_id	pipelineProcess
test_mode	False
try_number	1
unixname	airflow
Task Attributes

Attribute	Value
dag	<DAG: cv202303_nec>
dag_id	cv202303_nec
depends_on_past	False
deps	{<TIDep(Not In Retry Period)>, <TIDep(Trigger Rule)>, <TIDep(Previous Dagrun State)>}
do_xcom_push	True
downstream_list	[]
downstream_task_ids	set()
email	None
email_on_failure	True
email_on_retry	True
end_date	None
execution_timeout	None
executor_config	{}
extra_links	[]
global_operator_extra_link_dict	{}
inlets	[]
lineage_data	None
log	<Logger airflow.task.operators (INFO)>
logger	<Logger airflow.task.operators (INFO)>
max_retry_delay	None
on_failure_callback	None
on_retry_callback	None
on_success_callback	None
op_args	[]
op_kwargs	{}
operator_extra_link_dict	{}
operator_extra_links	()
outlets	[]
owner	airflow
params	{}
pool	default_pool
priority_weight	1
priority_weight_total	1
provide_context	True
queue	default
resources	None
retries	1
retry_delay	0:05:00
retry_exponential_backoff	False
run_as_user	None
schedule_interval	0 0,4,8,11,16 * * *
shallow_copy_attrs	('python_callable', 'op_kwargs')
sla	None
start_date	2022-10-24T00:00:00+00:00
subdag	None
task_concurrency	None
task_id	pipelineProcess
task_type	PythonOperator
template_ext	[]
template_fields	('templates_dict', 'op_args', 'op_kwargs')
templates_dict	None
trigger_rule	all_success
ui_color	#ffefeb
ui_fgcolor	#000
upstream_list	[<Task(PythonOperator): getDrowToken>]
upstream_task_ids	{'getDrowToken'}
wait_for_downstream	False
weight_rule	downstream
DAG: cv202303_nec

schedule: 0 0,4,8,11,16 * * *

Task Instance: pipelineProcess

Task Instance Details

Dependencies Blocking Task From Getting Scheduled

Attribute: python_callable

Task Instance Attributes

Task Attributes
