1_t24_interface.py

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

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'].replace('[LOCAL_PATH]' ,f"""{Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/""").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

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_neraca = BashOperator(
        task_id="sftp_neraca",
        bash_command=f"""sshpass -p {Variable.get("SFTP_T24_PASSWORD")} sftp -o StrictHostKeyChecking=no -r -P 2222 {Variable.get("SFTP_T24_USER")}@{Variable.get("SFTP_T24_HOST")}:bnk.interface/REPORT.BP/NERACA/{yesterday_nodash}/ID0010001/* {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/""",
    )

    sftp_nominatif = BashOperator(
        task_id="sftp_nominatif",
        bash_command=f"""sshpass -p {Variable.get("SFTP_T24_PASSWORD")} sftp -o StrictHostKeyChecking=no -r -P 2222 {Variable.get("SFTP_T24_USER")}@{Variable.get("SFTP_T24_HOST")}:bnk.interface/NOMINATIF*/{yesterday_nodash}/ID0010001* {Variable.get("LOCAL_PATH")}t24/{yesterday_nodash}/""",
    )

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

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

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



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

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


    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};
         """,
    )

    history_finish = PostgresOperator(
        sql=f"""
               UPDATE ds_conf."Datasource_history"
                   SET status = 'DONE', finish_time = '{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}'::timestamp with time zone
               WHERE source = '{POSTGRES_SCHEMA}' and status = 'ONPROCESS';
    """,
        task_id="history_finish",
        postgres_conn_id=POSTGRES_CONN_ID,
    )



    begin >> sftp_neraca >> sftp_nominatif >> sftp_ppap >> 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 >> pentaho >> zip_today >> delete_before >> history_finish