1_t24_interface_boii.py

import json, os
import csv
from datetime import date
import calendar
import time

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator, ShortCircuitOperator
from airflow.operators.bash import BashOperator

from datetime import datetime, timedelta
from airflow.providers.sftp.operators.sftp import SFTPOperator
from airflow.models import Variable
from airflow.providers.postgres.operators.postgres import PostgresOperator

from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.utils.trigger_rule import TriggerRule

from airflow.models.xcom import XCom
from airflow.decorators import task
from airflow.hooks.postgres_hook import PostgresHook
from airflow import XComArg



# yesterday_nodash = (datetime.now() - timedelta(1)).strftime('%Y%m%d')
yesterday_nodash = (datetime.now() - timedelta(1)).strftime('%Y%m%d') if Variable.get(
    "DATE_OF_DATA") == 'today' else Variable.get("DATE_OF_DATA")
yesterday_strip = datetime.strptime(yesterday_nodash, '%Y%m%d').strftime('%Y-%m-%d')
yesterday_lusa = (datetime.strptime(yesterday_nodash, '%Y%m%d') - timedelta(1)).strftime('%Y%m%d')

POSTGRES_CONN_ID = Variable.get("DS_DB")
POSTGRES_ENV = Variable.get("ENV_T24")
POSTGRES_SCHEMA = 'ds_t24' + Variable.get("ENV_T24")
DS_CONN_ID = 'ds_t24'
DS_FOLDER = 't24_interface'
DS_DB = 't24_interface'
DS_SCHEMA = 't24_interface'
DS_CREATE_TABLE = ''


def _start():
    print("Start :: Extractor ")

def ds_list_extractor():
    sql_stmt = f"""select * from ds_conf.ds_extractor_list_extractor('t24_interface', '{yesterday_strip}');"""
    pg_hook = PostgresHook(
        postgres_conn_id=POSTGRES_CONN_ID,
    )
    pg_conn = pg_hook.get_conn()
    cursor = pg_conn.cursor()
    cursor.execute(sql_stmt)
    files = cursor.fetchall()
    return files

def ds_push_syntax(ti):
    iris = ti.xcom_pull(task_ids=['ds_list_extractor'])
    if not iris:
        raise Exception('No data.')
    return [{"op_kwargs": {
        # "copy_sql": f"""COPY {POSTGRES_SCHEMA}.{table['table_name']} from STDOUT delimiter '{table['delimiter']}' CSV HEADER quote E'\b'""",
        "copy_sql": f"""COPY {POSTGRES_SCHEMA}.{table['table_name']} from STDOUT delimiter '{table['delimiter']}' CSV HEADER quote '"'""",
        "file_id": f"""{yesterday_nodash}/{table['file_id']}"""}}
        for
        table in json.loads(iris[0][0][0])]

def pg_ddl_syntax(ti):
    iris = ti.xcom_pull(task_ids=['ds_list_extractor'])
    arr = []
    for table in json.loads(iris[0][0][0]):
        with open(f"""{Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}""",encoding = "ISO-8859-1") as csvFile:
            reader = csv.reader(csvFile, delimiter=f"""{table['delimiter']}""")
            field_names_list = reader.__next__()
            arr.append({"sql": f"""drop table if exists {POSTGRES_SCHEMA}.{table['table_name']} cascade; create table {POSTGRES_SCHEMA}.{table['table_name']} ({' text, '.join([w.replace(' ', '_').replace('.', '_').replace('/', '_').replace('(', '_').replace(')', '_').replace('+', '_').replace('___', '_').replace('%', '').lower().replace((table['delimiter']+'limit'+table['delimiter']), (table['delimiter']+'limit_reff'+table['delimiter'])) for w in field_names_list])} text);"""})
    return arr

def csv_clean_syntax(ti):
    iris = ti.xcom_pull(task_ids=['ds_list_extractor'])
    arr = []
    for table in json.loads(iris[0][0][0]):
        with open(f"""{Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}""", encoding = "ISO-8859-1") as csvFile:
            reader = csv.reader(csvFile, delimiter=f"""{table['delimiter']}""")
            field_names_list = reader.__next__()
            arr.append({"bash_command": f"""echo 'OK' """ if table['sed_command'] == 'nil' else (f"""{Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']} > {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}_bk && mv {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}_bk {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']} && """.join(table['sed_command'].split('|;|;|')) + f""" {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']} > {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}_bk && mv {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}_bk {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}""")})
    return arr


def ds_push_csv(ti, copy_sql, file_id):
    pg_hook = PostgresHook.get_hook(POSTGRES_CONN_ID)
    pg_hook.copy_expert(copy_sql, filename=f"""/opt/airflow/dags/DFE/t24/{file_id}""")

def sql_clean_syntax(ti):
    iris = ti.xcom_pull(task_ids=['ds_list_extractor'])
    arr = []
    for table in json.loads(iris[0][0][0]):
        with open(f"""{Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/{table['file_id']}""", encoding = "ISO-8859-1") as csvFile:
            reader = csv.reader(csvFile, delimiter=f"""{table['delimiter']}""")
            field_names_list = reader.__next__()
            arr.append({"sql": f"""select 'OK' """ if table['sql_command'] == 'nil' else f"""{";".join(table['sql_command'].split('|;|;|')).replace('T24_SOURCE', POSTGRES_SCHEMA)}"""})
    return arr

def stop_task(**kwargs):
    return True

def check_backdate(**kwargs):
    if Variable.get("DATE_OF_DATA") == 'today':
        print(f"""Date: {Variable.get("DATE_OF_DATA")}""")
        print("OK!")
        return True
    else:
        print(f"""Date: {Variable.get("DATE_OF_DATA")}""")
        print("This is Backdate!")
        return False

def check_surrounding_done(**kwargs):
    paths_labels = [
        ("mgate", "MGate"),
        ("switching", "Switching"),
        ("ibb", "IBB"),
        ("ebanking", "Ebanking")
    ]
    run_next = False 
    base_path = f"""{Variable.get("LOCAL_PATH")}"""
    done_ext = f"""{yesterday_nodash}/done_ext.csv"""
    while run_next == False:
     i = 0
     for path, label in paths_labels:
        file_path = f"""{base_path}{path}/{done_ext}"""
        print(f"{label} {'SELESAI' if os.path.isfile(f'{base_path}{path}/{done_ext}') else 'BELUM SELESAI'} - Path : {file_path}")
        i += (0 if not os.path.isfile(f'{base_path}{path}/{done_ext}') else 1)
     print(i)
     if i == 4:
         return True
     else:
         print("===================================")
         time.sleep(60)

with DAG("APJ_1_t24_interface",
         start_date=datetime(2021, 1, 1),
         schedule_interval=None,
         catchup=False,
         concurrency=3) as dag:
    begin = PythonOperator(
        task_id=f"Begin",
        python_callable=_start,
        op_kwargs={

        }
    )

    sftp_xx = BashOperator(
        task_id="sftp_xx",
        bash_command=f"""sshpass -p {Variable.get("SFTP_T24_PASSWORD")} sftp -o StrictHostKeyChecking=no -r {Variable.get("SFTP_T24_USER")}@{Variable.get("SFTP_T24_HOST")}:/REPORT.BP/NOM/TREASURY/*{yesterday_nodash}* {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/""",
    )

    sftp_ppap_t24 = BashOperator(
        task_id="sftp_ppap_t24",
        bash_command=f"""sshpass -p {Variable.get("SFTP_T24_PASSWORD")} sftp -o StrictHostKeyChecking=no -r {Variable.get("SFTP_T24_USER")}@{Variable.get("SFTP_T24_HOST")}:/PPAP.NOMINATIF/*{yesterday_nodash}*.csv {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/""",
    )

    # sftp_neraca = BashOperator(
    #     task_id="sftp_xx",
    #     bash_command=f"""sshpass -p {Variable.get("SFTP_T24_PASSWORD")} sftp -o StrictHostKeyChecking=no -r {Variable.get("SFTP_T24_USER")}@{Variable.get("SFTP_T24_HOST")}:/REPORT.BP/NERACA/*{yesterday_nodash}* {Variable.get("LOCAL_PATH")}t24_neraca/{yesterday_nodash}/""",
    # )
    # sftp_neraca_tel = BashOperator(
    #     task_id="sftp_xx",
    #     bash_command=f"""sshpass -p {Variable.get("SFTP_T24_PASSWORD")} sftp -o StrictHostKeyChecking=no -r {Variable.get("SFTP_T24_USER")}@{Variable.get("SFTP_T24_HOST")}:/REPORT.BP/NERACA/*{yesterday_nodash}* {Variable.get("LOCAL_PATH")}t24_neraca/{yesterday_nodash}/""",
    # )

    ds_list_extractor = PythonOperator(
        task_id='ds_list_extractor',
        python_callable=ds_list_extractor,
        do_xcom_push=True
    )

    # ft_reve = BashOperator(
    #     task_id="ft_reve",
    #     bash_command=f"""head -n -2 {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.csv > {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.2.csv && mv {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.2.csv {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.csv""",
    # )

    csv_clean_syntax = PythonOperator(
        task_id='csv_clean_syntax',
        python_callable=csv_clean_syntax
    )

    clean_csv = BashOperator.partial(
        task_id="clean_csv",
        # bash_command=f"""head -n -2 {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.csv > {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.2.csv && mv {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.2.csv {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/FT.REVERSE.csv""",
    ).expand_kwargs(
        XComArg(csv_clean_syntax),
    )


    # pg_drop_schema = PostgresOperator(
    #     sql= f"""DROP TABLE IF EXISTS {POSTGRES_SCHEMA} CASCADE;""",
    #     task_id="pg_drop_schema",
    #     postgres_conn_id=POSTGRES_CONN_ID,
    # )
    #
    # pg_create_schema = PostgresOperator(
    #     sql= f"""create schema IF NOT EXISTS {POSTGRES_SCHEMA};""",
    #     task_id="pg_create_schema",
    #     postgres_conn_id=POSTGRES_CONN_ID,
    # )

    stop_task = ShortCircuitOperator(
        task_id="stop_task",
        provide_context=True,
        python_callable=stop_task,
        op_kwargs={},
    )


    pg_ddl_syntax = PythonOperator(
        task_id='pg_ddl_syntax',
        python_callable=pg_ddl_syntax
    )

    pg_create_table = PostgresOperator.partial(
        task_id="pg_create_table",
        postgres_conn_id=POSTGRES_CONN_ID,
    ).expand_kwargs(
        XComArg(pg_ddl_syntax),
    )

    # ds_truncate_syntax = PythonOperator(
    #     task_id='ds_truncate_syntax',
    #     python_callable=ds_truncate_syntax
    # )
    #
    # ds_truncate = PostgresOperator.partial(
    #     task_id="ds_truncate",
    #     postgres_conn_id=POSTGRES_CONN_ID,
    # ).expand_kwargs(
    #     XComArg(ds_truncate_syntax)
    # )
    #
    ds_push_syntax = PythonOperator(
        task_id='ds_syntax_push',
        python_callable=ds_push_syntax
    )

    ds_csv_to_table = PythonOperator.partial(
        task_id="ds_csv_to_table",
        python_callable=ds_push_csv,
        dag=dag
    ).expand_kwargs(
        XComArg(ds_push_syntax),
    )

    sql_clean_syntax = PythonOperator(
        task_id='sql_clean_syntax',
        python_callable=sql_clean_syntax
    )

    ds_clean_data = PostgresOperator.partial(
        # sql=f"""select ds_conf.ds_t24_create_table_history_wo_t24_dfe('{POSTGRES_SCHEMA}');""",
        task_id="ds_clean_data",
        postgres_conn_id=POSTGRES_CONN_ID,
    ).expand_kwargs(
        XComArg(sql_clean_syntax),
    )


    ds_create_table_history_nominatif = PostgresOperator(
        sql=f"""select ds_conf.ds_t24_create_table_history_wo_t24_dfe('{POSTGRES_SCHEMA}');""",
        task_id="ds_create_table_history_nominatif",
        postgres_conn_id=POSTGRES_CONN_ID,
    )


    ds_to_history = PostgresOperator(
        sql=f"""select ds_conf.ds_t24_copy_to_history('{yesterday_strip}', '{POSTGRES_SCHEMA}');""",
        task_id="ds_to_history",
        postgres_conn_id=POSTGRES_CONN_ID,
    )

    set_access_schemma = PostgresOperator(
        sql=f"""GRANT USAGE ON SCHEMA {POSTGRES_SCHEMA} TO readaccess;""",
        task_id="set_access_schemma",
        postgres_conn_id=POSTGRES_CONN_ID,
    )

    set_access_all_table = PostgresOperator(
        sql=f"""GRANT SELECT ON ALL TABLES IN SCHEMA {POSTGRES_SCHEMA} TO readaccess;""",
        task_id="set_access_all_table",
        postgres_conn_id=POSTGRES_CONN_ID,
    )

    check_backdate = ShortCircuitOperator(
        task_id="check_backdate",
        provide_context=True,
        python_callable=check_backdate,
        op_kwargs={}
    )

    check_surrounding_done = ShortCircuitOperator(
        task_id="check_surrounding_done",
        provide_context=True,
        python_callable=check_surrounding_done,
        op_kwargs={}
    )

    pentaho = BashOperator(
            task_id='pentaho',
            bash_command=f"""curl '{Variable.get("PENTAHO_HOST_PASSWORD")}/kettle/executeJob/?job=/home/apj/files/scripts/etl/D_REPORT.kjb'"""
            # bash_command=f"""curl '{Variable.get("PENTAHO_HOST")}/kettle/executeJob/?rep=test-repo&job=/home/oper/files/f/Untitled'"""
    )

    rep_nominatif_customer = PostgresOperator(
        sql=f"""select ds_conf.dm_customer_nominative('{yesterday_strip}');""",
        task_id="rep_nominatif_customer",
        postgres_conn_id=POSTGRES_CONN_ID,
    )


    zip_today = BashOperator(
        task_id="zip_today",
        bash_command=f"""
         zip -r {yesterday_nodash}.zip /opt/airflow/dags/DFE/t24/{yesterday_nodash};
         """,
    )

    delete_before = BashOperator(
        task_id="delete_before",
        bash_command=f"""
         rm -r /opt/airflow/dags/DFE/t24/{yesterday_lusa};
         """,
    )


    begin >> sftp_xx >> sftp_ppap_t24 >> ds_list_extractor >> csv_clean_syntax >> clean_csv  >> pg_ddl_syntax >> pg_create_table >> ds_push_syntax >> sql_clean_syntax >> ds_csv_to_table >> ds_clean_data  >> stop_task >> ds_create_table_history_nominatif >> ds_to_history >> set_access_schemma >> set_access_all_table >> check_backdate >> check_surrounding_done >> pentaho >> rep_nominatif_customer >> zip_today >> delete_before