import os
import sys

# for airflow subfolder, when no module named
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from datetime import timedelta, datetime
from airflow.operators.python_operator import PythonOperator
from airflow.models import DAG

from cloudengine.utils.common import config, failure_callback
from cloudengine.ods import extract

from cloudengine.dwd.dim_gmpaudit_audit_main_plan import dim_gmpaudit_audit_main_plan
from cloudengine.dwd.dim_gmpcapa_capa import dim_gmpcapa_capa
from cloudengine.dwd.dim_gmpchange_changemain import dim_gmpchange_changemain
from cloudengine.dwd.dim_gmpdeviation_deviation import dim_gmpdeviation_deviation
from cloudengine.dwd.dim_gmpdiscrepancy_discrepancy import dim_gmpdiscrepancy_discrepancy
from cloudengine.dwd.dim_gmpvalidation_validation import dim_gmpvalidation_validation

from cloudengine.ads.ads_gmpaudit_audit_main_plan import ads_gmpaudit_audit_main_plan
from cloudengine.ads.ads_gmpcapa_capa import ads_gmpcapa_capa
from cloudengine.ads.ads_gmpchange_changemain import ads_gmpchange_changemain
from cloudengine.ads.ads_gmpdeviation_deviation import ads_gmpdeviation_deviation
from cloudengine.ads.ads_gmpdiscrepancy_discrepancy import ads_gmpdiscrepancy_discrepancy
from cloudengine.ads.ads_gmpvalidation_validation import ads_gmpvalidation_validation

from cloudengine.utils.tableau.refresh_data import tableau_data_extract

error_inform = config['email_list']['error_inform']
error_inform_list = error_inform.split(',')

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2020, 9, 2),
    'email': error_inform_list,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=2),
    'provide_context': False,
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'schedule_interval': '20 16 * * *',
    # 'end_date': datetime(2018, 4, 14),
}

default_args['on_failure_callback'] = failure_callback

CONF_SECTION = "airflow_schedule"

gmpdeviation_flow_interval = config[CONF_SECTION]['gmpdeviation_flow']
gmpdeviation_flow_dag = DAG('gmpdeviation_flow', default_args=default_args,
                    schedule_interval=gmpdeviation_flow_interval)

####### ods level
def level_load():
    pass

task_ods_level = PythonOperator(
    task_id='ods_level_load',
    python_callable=level_load,
    dag=gmpdeviation_flow_dag)

def quatrust_ods_load():
    extract.extract_group("quatrustdata")

task_ods_quatrust = PythonOperator(
    task_id='ods_quatrust',
    python_callable=quatrust_ods_load,
    dag=gmpdeviation_flow_dag)

def aotai_sys_ods_load():
    extract.extract_group("aotai_sys")

task_ods_aotai_sys = PythonOperator(
    task_id='ods_aotai_sys',
    python_callable=aotai_sys_ods_load,
    dag=gmpdeviation_flow_dag)

def aotai_ums_ods_load():
    extract.extract_group("aotai_ums")

task_ods_aotai_ums = PythonOperator(
    task_id='ods_aotai_ums',
    python_callable=aotai_ums_ods_load,
    dag=gmpdeviation_flow_dag)

def gmpvalid_ods_load():
    extract.extract_group("gmpvalidation")

task_ods_gmpvalid = PythonOperator(
    task_id='ods_gmpvalid',
    python_callable=gmpvalid_ods_load,
    dag=gmpdeviation_flow_dag)

def gmpaudit_ods_load():
    extract.extract_group("gmpauditmanage")

task_ods_gmpaudit = PythonOperator(
    task_id='ods_gmpaudit',
    python_callable=gmpaudit_ods_load,
    dag=gmpdeviation_flow_dag)

def gmpcapa_ods_load():
    extract.extract_group("gmpcapadata")

task_ods_gmpcapa = PythonOperator(
    task_id='ods_gmpcapa',
    python_callable=gmpcapa_ods_load,
    dag=gmpdeviation_flow_dag)

def gmpchange_ods_load():
    extract.extract_group("gmpchangemanagement")

task_ods_gmpchange = PythonOperator(
    task_id='ods_gmpchange',
    python_callable=gmpchange_ods_load,
    dag=gmpdeviation_flow_dag)

def gmpdevi_ods_load():
    extract.extract_group("gmpdeviation")

task_ods_gmpdevi = PythonOperator(
    task_id='ods_gmpdevi',
    python_callable=gmpdevi_ods_load,
    dag=gmpdeviation_flow_dag)

def gmpdisc_ods_load():
    extract.extract_group("gmpdiscrepancy")

task_ods_gmpdisc = PythonOperator(
    task_id='ods_gmpdisc',
    python_callable=gmpdisc_ods_load,
    dag=gmpdeviation_flow_dag)

def m_authorizedobjectmanagement_ods_load():
    extract.extract_group("authorizedobjectmanagement")

task_ods_authorizedobjectmanagement = PythonOperator(
    task_id='ods_authorizedobjectmanagement',
    python_callable=m_authorizedobjectmanagement_ods_load,
    dag=gmpdeviation_flow_dag)


####  dim level
def dim_level_load():
    pass

task_dim_level = PythonOperator(
    task_id='dim_level_load',
    python_callable=level_load,
    dag=gmpdeviation_flow_dag)

task_dim_gmpaudit_audit_main_plan = PythonOperator(
    task_id='dim_gmpaudit_audit_main_plan',
    python_callable=dim_gmpaudit_audit_main_plan.process,
    dag=gmpdeviation_flow_dag)

task_dim_gmpcapa_capa = PythonOperator(
    task_id='dim_gmpcapa_capa',
    python_callable=dim_gmpcapa_capa.process,
    dag=gmpdeviation_flow_dag)

task_dim_gmpchange_changemain = PythonOperator(
    task_id='dim_gmpchange_changemain',
    python_callable=dim_gmpchange_changemain.process,
    dag=gmpdeviation_flow_dag)

task_dim_gmpdeviation_deviation = PythonOperator(
    task_id='dim_gmpdeviation_deviation',
    python_callable=dim_gmpdeviation_deviation.process,
    dag=gmpdeviation_flow_dag)

task_dim_gmpdiscrepancy_discrepancy = PythonOperator(
    task_id='dim_gmpdiscrepancy_discrepancy',
    python_callable=dim_gmpdiscrepancy_discrepancy.process,
    dag=gmpdeviation_flow_dag)

task_dim_gmpvalidation_validation = PythonOperator(
    task_id='dim_gmpvalidation_validation',
    python_callable=dim_gmpvalidation_validation.process,
    dag=gmpdeviation_flow_dag)

####  dwd level
def dwd_level_load():
    pass

task_dwd_level = PythonOperator(
    task_id='dwd_level_load',
    python_callable=level_load,
    dag=gmpdeviation_flow_dag)

def dws_level_load():
    pass

####  dws level
task_dws_level = PythonOperator(
    task_id='dws_level_load',
    python_callable=level_load,
    dag=gmpdeviation_flow_dag)

####  ads level
def ads_level_load():
    pass

task_ads_level = PythonOperator(
    task_id='ads_level_load',
    python_callable=level_load,
    dag=gmpdeviation_flow_dag)

task_ads_gmpaudit_audit_main_plan = PythonOperator(
    task_id='ads_gmpaudit_audit_main_plan',
    python_callable=ads_gmpaudit_audit_main_plan.process,
    dag=gmpdeviation_flow_dag)

task_ads_gmpcapa_capa = PythonOperator(
    task_id='ads_gmpcapa_capa',
    python_callable=ads_gmpcapa_capa.process,
    dag=gmpdeviation_flow_dag)

task_ads_gmpchange_changemain = PythonOperator(
    task_id='ads_gmpchange_changemain',
    python_callable=ads_gmpchange_changemain.process,
    dag=gmpdeviation_flow_dag)

task_ads_gmpdeviation_deviation = PythonOperator(
    task_id='ads_gmpdeviation_deviation',
    python_callable=ads_gmpdeviation_deviation.process,
    dag=gmpdeviation_flow_dag)

task_ads_gmpdiscrepancy_discrepancy = PythonOperator(
    task_id='ads_gmpdiscrepancy_discrepancy',
    python_callable=ads_gmpdiscrepancy_discrepancy.process,
    dag=gmpdeviation_flow_dag)

task_ads_gmpvalidation_validation = PythonOperator(
    task_id='ads_gmpvalidation_validation',
    python_callable=ads_gmpvalidation_validation.process,
    dag=gmpdeviation_flow_dag)

####  tableau_data_extract
def gmpaudit_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpaudit_flow")
    refresh_data.run_job()

task_gmpaudit_tableau_data_extract = PythonOperator(
    task_id='gmpaudit_tableau_data_extract_task',
    python_callable=gmpaudit_tableau_data_extract_task,
    dag=gmpdeviation_flow_dag)

def gmpcapa_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpcapa_flow")
    refresh_data.run_job()

task_gmpcapa_tableau_data_extract = PythonOperator(
    task_id='gmpcapa_tableau_data_extract_task',
    python_callable=gmpcapa_tableau_data_extract_task,
    dag=gmpdeviation_flow_dag)

def gmpchange_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpchange_flow")
    refresh_data.run_job()

task_gmpchange_tableau_data_extract = PythonOperator(
    task_id='gmpchange_tableau_data_extract_task',
    python_callable=gmpchange_tableau_data_extract_task,
    dag=gmpdeviation_flow_dag)
def gmpdeviation_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpdeviation_flow")
    refresh_data.run_job()

task_gmpdeviation_tableau_data_extract = PythonOperator(
    task_id='gmpdeviation_tableau_data_extract_task',
    python_callable=gmpdeviation_tableau_data_extract_task,
    dag=gmpdeviation_flow_dag)

def gmpdiscrepancy_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpdiscrepancy_flow")
    refresh_data.run_job()

task_gmpdiscrepancy_tableau_data_extract = PythonOperator(
    task_id='gmpdiscrepancy_tableau_data_extract_task',
    python_callable=gmpdiscrepancy_tableau_data_extract_task,
    dag=gmpdeviation_flow_dag)

def gmpvalidation_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpvalidation_flow")
    refresh_data.run_job()

task_gmpvalidation_tableau_data_extract = PythonOperator(
    task_id='gmpvalidation_tableau_data_extract_task',
    python_callable=gmpvalidation_tableau_data_extract_task,
    dag=gmpdeviation_flow_dag)

####  tableau level
def tableau_level_load():
    pass

task_tableau_level = PythonOperator(
    task_id='tableau_level_load',
    python_callable=level_load,
    dag=gmpdeviation_flow_dag)

task_ods_level >> [task_ods_quatrust, task_ods_aotai_sys, task_ods_aotai_ums, task_ods_authorizedobjectmanagement, task_ods_gmpvalid, task_ods_gmpaudit, task_ods_gmpcapa, task_ods_gmpchange, task_ods_gmpdevi, task_ods_gmpdisc] >> task_dim_level >> [
    task_dim_gmpaudit_audit_main_plan, task_dim_gmpcapa_capa, task_dim_gmpchange_changemain, task_dim_gmpdeviation_deviation, task_dim_gmpdiscrepancy_discrepancy, task_dim_gmpvalidation_validation] >> task_dwd_level >> task_dws_level >> [
    task_ads_gmpaudit_audit_main_plan, task_ads_gmpcapa_capa, task_ads_gmpchange_changemain, task_ads_gmpdeviation_deviation, task_ads_gmpdiscrepancy_discrepancy, task_ads_gmpvalidation_validation] >> task_ads_level >> [
    task_gmpaudit_tableau_data_extract, task_gmpcapa_tableau_data_extract, task_gmpchange_tableau_data_extract, task_gmpdeviation_tableau_data_extract, task_gmpdiscrepancy_tableau_data_extract, task_gmpvalidation_tableau_data_extract] >> task_tableau_level