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_gmpdoc_doc import dim_gmpdoc_doc
from cloudengine.dwd.dim_gmpdoc_doc_history import dim_gmpdoc_doc_history
from cloudengine.dwd.dim_gmpchange_changemain import dim_gmpchange_changemain
from cloudengine.dwd.dim_gmpdoc_file_publish_version import dim_gmpdoc_file_publish_version
from cloudengine.dwd.dim_quatrust_user_department import dim_quatrust_user_department
from cloudengine.dwd.dim_quatrust_all_user_department import dim_quatrust_all_user_department
from cloudengine.dwd.dim_quatrust_user import dim_quatrust_user
from cloudengine.dwd.dim_quatrust_user_role import dim_quatrust_user_role
from cloudengine.dwd.dim_quatrust_user_soft import dim_quatrust_user_soft
from cloudengine.dwd.dim_operation_om import dim_operation_om
from cloudengine.dwd.dim_quatrust_operation_executor import dim_quatrust_operation_executor
from cloudengine.dwd.dim_quatrust_operation_module import dim_quatrust_operation_module
from cloudengine.dwd.dim_quatrust_operation_principal import dim_quatrust_operation_principal
from cloudengine.dwd.dim_gmpdoc_printing_reports import dim_gmpdoc_printing_reports


from cloudengine.dwd.dwd_gmpdoc_doc_task_detail import dwd_gmpdoc_doc_task_detail
from cloudengine.dwd.dwd_gmpchange_changemain_task_detail import dwd_gmpchange_changemain_task_detail
from cloudengine.dwd.dwd_gmpdoc_doc_execute_task_detail import dwd_gmpdoc_doc_execute_task_detail
from cloudengine.dwd.dwd_gmpdoc_parent_child_nonjoinder import dwd_gmpdoc_parent_child_nonjoinder
from cloudengine.dwd.dwd_gmpdoc_parent_child_relationship import dwd_gmpdoc_parent_child_relationship
from cloudengine.dwd.dwd_vdr_doc_hierarchical_folder import dwd_vdr_doc_hierarchical_folder
from cloudengine.dwd.dwd_vdr_hierarchical_folder import dwd_vdr_hierarchical_folder

from cloudengine.ads.ads_gmpdoc_doc_task_detail import ads_gmpdoc_doc_task_detail
from cloudengine.ads.ads_gmpchange_changemain_task_detail import ads_gmpchange_changemain_task_detail
from cloudengine.ads.ads_gmpdoc_exexute_task_detail import ads_gmpdoc_exexute_task_detail
from cloudengine.ads.ads_gmpdoc_dim_doc import ads_gmpdoc_dim_doc
from cloudengine.ads.ads_gmpdoc_dim_doc_history import ads_gmpdoc_dim_doc_history
from cloudengine.ads.ads_gmpdoc_doc_withdraw_action import ads_gmpdoc_doc_withdraw_action
from cloudengine.ads.ads_gmpdoc_doc_withdraw_in_approval import ads_gmpdoc_doc_withdraw_in_approval
from cloudengine.ads.ads_quatrust_user_department import ads_quatrust_user_department
from cloudengine.ads.ads_quatrust_user import ads_quatrust_user
from cloudengine.ads.ads_quatrust_user_role import ads_quatrust_user_role
from cloudengine.ads.ads_quatrust_user_soft import ads_quatrust_user_soft
from cloudengine.ads.ads_quatrust_all_user_department import ads_quatrust_all_user_department
from cloudengine.ads.ads_operation_om import ads_operation_om
from cloudengine.ads.ads_quatrust_operation_executor import ads_quatrust_operation_executor
from cloudengine.ads.ads_quatrust_operation_module import ads_quatrust_operation_module
from cloudengine.ads.ads_quatrust_operation_principal import ads_quatrust_operation_principal
from cloudengine.ads.ads_gmpdoc_printing_reports import ads_gmpdoc_printing_reports
from cloudengine.ads.ads_gmpdoc_dim_doc_nonjoinder import ads_gmpdoc_dim_doc_nonjoinder

from cloudengine.ads.ads_vdr_doc_hierarchical_folder import ads_vdr_doc_hierarchical_folder
from cloudengine.ads.ads_vdr_hierarchical_folder import ads_vdr_hierarchical_folder

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"

gmp_borad_flow_interval = config[CONF_SECTION]['gmp_board_flow']
gmp_board_flow_dag = DAG('gmp_board_flow', default_args=default_args,
                         schedule_interval=gmp_borad_flow_interval)


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

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

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

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

def gmpdoc_ods_load():
    extract.extract_group("gmpdoc")

task_ods_gmpdoc = PythonOperator(
    task_id='ods_gmpdoc',
    python_callable=gmpdoc_ods_load,
    dag=gmp_board_flow_dag)

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

task_ods_gmpchange = PythonOperator(
    task_id='ods_gmpchange',
    python_callable=gmpchange_ods_load,
    dag=gmp_board_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=gmp_board_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=gmp_board_flow_dag)

def operation_ods_load():
    extract.extract_group("operation")

task_ods_operation = PythonOperator(
    task_id='ods_operation',
    python_callable=operation_ods_load,
    dag=gmp_board_flow_dag)

def newdirdocserver_ods_load():
    extract.extract_group("newdirdocserver")

task_ods_newdirdocserver = PythonOperator(
    task_id='ods_newdirdocserver',
    python_callable=newdirdocserver_ods_load,
    dag=gmp_board_flow_dag)

def filemanage_ods_load():
    extract.extract_group("filemanage")

task_ods_filemanage = PythonOperator(
    task_id='ods_filemanage',
    python_callable=filemanage_ods_load,
    dag=gmp_board_flow_dag)

def centermanage_ods_load():
    extract.extract_group("centermanage")

task_ods_centermanage = PythonOperator(
    task_id='ods_centermanage',
    python_callable=centermanage_ods_load,
    dag=gmp_board_flow_dag)


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

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

# def dim_gmpdoc_doc():
#     dim_gmpdoc_doc.process()

task_dim_gmpdoc_doc = PythonOperator(
    task_id='dim_gmpdoc_doc',
    python_callable=dim_gmpdoc_doc.process,
    dag=gmp_board_flow_dag)

# def dim_gmpchange_changemain():
#     dim_gmpchange_changemain.process()

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

task_dim_gmpdoc_doc_history = PythonOperator(
    task_id='dim_gmpdoc_doc_history',
    python_callable=dim_gmpdoc_doc_history.process,
    dag=gmp_board_flow_dag)

task_dim_gmpdoc_file_publish_version = PythonOperator(
    task_id='dim_gmpdoc_file_publish_version',
    python_callable=dim_gmpdoc_file_publish_version.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_user_department = PythonOperator(
    task_id='dim_quatrust_user_department',
    python_callable=dim_quatrust_user_department.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_all_user_department = PythonOperator(
    task_id='dim_quatrust_all_user_department',
    python_callable=dim_quatrust_all_user_department.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_user = PythonOperator(
    task_id='dim_quatrust_user',
    python_callable=dim_quatrust_user.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_user_role = PythonOperator(
    task_id='dim_quatrust_user_role',
    python_callable=dim_quatrust_user_role.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_user_soft = PythonOperator(
    task_id='dim_quatrust_user_soft',
    python_callable=dim_quatrust_user_soft.process,
    dag=gmp_board_flow_dag)

task_dim_operation_om = PythonOperator(
    task_id='dim_operation_om',
    python_callable=dim_operation_om.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_operation_executor = PythonOperator(
    task_id='dim_quatrust_operation_executor',
    python_callable=dim_quatrust_operation_executor.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_operation_module = PythonOperator(
    task_id='dim_quatrust_operation_module',
    python_callable=dim_quatrust_operation_module.process,
    dag=gmp_board_flow_dag)

task_dim_quatrust_operation_principal = PythonOperator(
    task_id='dim_quatrust_operation_principal',
    python_callable=dim_quatrust_operation_principal.process,
    dag=gmp_board_flow_dag)

task_dim_gmpdoc_printing_reports = PythonOperator(
    task_id='dim_gmpdoc_printing_reports',
    python_callable=dim_gmpdoc_printing_reports.process,
    dag=gmp_board_flow_dag)

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

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

task_dwd_gmpdoc_doc_task_detail = PythonOperator(
    task_id='dwd_gmpdoc_doc_task_detail',
    python_callable=dwd_gmpdoc_doc_task_detail.process,
    dag=gmp_board_flow_dag)

task_dwd_gmpchange_changemain_task_detail = PythonOperator(
    task_id='dwd_gmpchange_changemain_task_detail',
    python_callable=dwd_gmpchange_changemain_task_detail.process,
    dag=gmp_board_flow_dag)

task_dwd_gmpdoc_doc_execute_task_detail = PythonOperator(
    task_id='dwd_gmpdoc_doc_execute_task_detail',
    python_callable=dwd_gmpdoc_doc_execute_task_detail.process,
    dag=gmp_board_flow_dag)

task_dwd_gmpdoc_parent_child_nonjoinder = PythonOperator(
    task_id='dwd_gmpdoc_parent_child_nonjoinder',
    python_callable=dwd_gmpdoc_parent_child_nonjoinder.process,
    dag=gmp_board_flow_dag)

task_dwd_gmpdoc_parent_child_relationship = PythonOperator(
    task_id='dwd_gmpdoc_parent_child_relationship',
    python_callable=dwd_gmpdoc_parent_child_relationship.process,
    dag=gmp_board_flow_dag)

task_dwd_vdr_doc_hierarchical_folder = PythonOperator(
    task_id='dwd_vdr_doc_hierarchical_folder',
    python_callable=dwd_vdr_doc_hierarchical_folder.process,
    dag=gmp_board_flow_dag)

task_dwd_vdr_hierarchical_folder = PythonOperator(
    task_id='dwd_vdr_hierarchical_folder',
    python_callable=dwd_vdr_hierarchical_folder.process,
    dag=gmp_board_flow_dag)

def dws_level_load():
    pass

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

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

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

task_ads_gmpdoc_doc_task_detail = PythonOperator(
    task_id='ads_gmpdoc_doc_task_detail',
    python_callable=ads_gmpdoc_doc_task_detail.process,
    dag=gmp_board_flow_dag)

task_ads_gmpchange_changemain_task_detail = PythonOperator(
    task_id='ads_gmpchange_changemain_task_detail',
    python_callable=ads_gmpchange_changemain_task_detail.process,
    dag=gmp_board_flow_dag)

task_ads_gmpdoc_exexute_task_detail = PythonOperator(
    task_id='ads_gmpdoc_exexute_task_detail',
    python_callable=ads_gmpdoc_exexute_task_detail.process,
    dag=gmp_board_flow_dag)

task_ads_gmpdoc_dim_doc = PythonOperator(
    task_id='ads_gmpdoc_dim_doc',
    python_callable=ads_gmpdoc_dim_doc.process,
    dag=gmp_board_flow_dag)

task_ads_gmpdoc_dim_doc_history = PythonOperator(
    task_id='ads_gmpdoc_dim_doc_history',
    python_callable=ads_gmpdoc_dim_doc_history.process,
    dag=gmp_board_flow_dag)

task_ads_gmpdoc_doc_withdraw_action = PythonOperator(
    task_id='ads_gmpdoc_doc_withdraw_action',
    python_callable=ads_gmpdoc_doc_withdraw_action.process,
    dag=gmp_board_flow_dag)

task_ads_gmpdoc_doc_withdraw_in_approval = PythonOperator(
    task_id='ads_gmpdoc_doc_withdraw_in_approval',
    python_callable=ads_gmpdoc_doc_withdraw_in_approval.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_user_department = PythonOperator(
    task_id='ads_quatrust_user_department',
    python_callable=ads_quatrust_user_department.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_all_user_department = PythonOperator(
    task_id='ads_quatrust_all_user_department',
    python_callable=ads_quatrust_all_user_department.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_user = PythonOperator(
    task_id='ads_quatrust_user',
    python_callable=ads_quatrust_user.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_user_role = PythonOperator(
    task_id='ads_quatrust_user_role',
    python_callable=ads_quatrust_user_role.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_user_soft = PythonOperator(
    task_id='ads_quatrust_user_soft',
    python_callable=ads_quatrust_user_soft.process,
    dag=gmp_board_flow_dag)

task_ads_operation_om = PythonOperator(
    task_id='ads_operation_om',
    python_callable=ads_operation_om.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_operation_executor = PythonOperator(
    task_id='ads_quatrust_operation_executor',
    python_callable=ads_quatrust_operation_executor.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_operation_module = PythonOperator(
    task_id='ads_quatrust_operation_module',
    python_callable=ads_quatrust_operation_module.process,
    dag=gmp_board_flow_dag)

task_ads_quatrust_operation_principal = PythonOperator(
    task_id='ads_quatrust_operation_principal',
    python_callable=ads_quatrust_operation_principal.process,
    dag=gmp_board_flow_dag)

task_ads_gmpdoc_printing_reports = PythonOperator(
    task_id='ads_gmpdoc_printing_reports',
    python_callable=ads_gmpdoc_printing_reports.process,
    dag=gmp_board_flow_dag)

task_ads_gmpdoc_dim_doc_nonjoinder = PythonOperator(
    task_id='ads_gmpdoc_dim_doc_nonjoinder',
    python_callable=ads_gmpdoc_dim_doc_nonjoinder.process,
    dag=gmp_board_flow_dag)

task_ads_vdr_doc_hierarchical_folder = PythonOperator(
    task_id='ads_vdr_doc_hierarchical_folder',
    python_callable=ads_vdr_doc_hierarchical_folder.process,
    dag=gmp_board_flow_dag)

task_ads_vdr_hierarchical_folder = PythonOperator(
    task_id='ads_vdr_hierarchical_folder',
    python_callable=ads_vdr_hierarchical_folder.process,
    dag=gmp_board_flow_dag)

####  tableau_data_extract
def gmpdoc_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpdoc_flow")
    refresh_data.run_job()

task_gmpdoc_tableau_data_extract = PythonOperator(
    task_id='gmpdoc_tableau_data_extract_task',
    python_callable=gmpdoc_tableau_data_extract_task,
    dag=gmp_board_flow_dag)

def user_condition_tableau_data_extract_task():
    refresh_data = tableau_data_extract("user_condition_flow")
    refresh_data.run_job()

task_user_condition_tableau_data_extract = PythonOperator(
    task_id='user_condition_tableau_data_extract_task',
    python_callable=user_condition_tableau_data_extract_task,
    dag=gmp_board_flow_dag)

def operation_tableau_data_extract_task():
    refresh_data = tableau_data_extract("operation_flow")
    refresh_data.run_job()

task_operation_tableau_data_extract = PythonOperator(
    task_id='operation_tableau_data_extract_task',
    python_callable=operation_tableau_data_extract_task,
    dag=gmp_board_flow_dag)

def gmpdocprintingreports_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmpdocprintingreports_flow")
    refresh_data.run_job()

task_gmpdocprintingreports_flow_tableau_data_extract = PythonOperator(
    task_id='gmpdocprintingreports_tableau_data_extract_task',
    python_callable=gmpdocprintingreports_tableau_data_extract_task,
    dag=gmp_board_flow_dag)

def vdr_tableau_data_extract_task():
    refresh_data = tableau_data_extract("vdr_flow")
    refresh_data.run_job()

task_vdr_tableau_data_extract_task = PythonOperator(
    task_id='vdr_tableau_data_extract_task',
    python_callable=vdr_tableau_data_extract_task,
    dag=gmp_board_flow_dag)


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

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



task_ods_level >> task_ods_operation >> task_ods_quatrust >> task_ods_gmpdoc >> task_dim_level >> [task_dim_gmpchange_changemain,
                                                                                    task_dim_gmpdoc_file_publish_version,
                                                                                    task_dim_quatrust_user, task_dim_quatrust_user_soft, task_dim_gmpdoc_printing_reports] >> task_dwd_level >> [
    task_dwd_gmpdoc_doc_task_detail, task_dwd_gmpchange_changemain_task_detail,
    task_dwd_gmpdoc_doc_execute_task_detail, task_dwd_gmpdoc_parent_child_nonjoinder, task_dwd_gmpdoc_parent_child_relationship, task_dwd_vdr_doc_hierarchical_folder, task_dwd_vdr_hierarchical_folder] >> task_dws_level >> [
    task_ads_gmpchange_changemain_task_detail,
    task_ads_operation_om, task_ads_gmpdoc_printing_reports] >> task_ads_level >> [
    task_gmpdoc_tableau_data_extract, task_user_condition_tableau_data_extract, task_operation_tableau_data_extract, task_gmpdocprintingreports_flow_tableau_data_extract, task_vdr_tableau_data_extract_task] >> task_tableau_level
task_ods_level >> task_ods_newdirdocserver >> task_ods_filemanage >> task_ods_centermanage >> task_dim_level
task_ods_level >> task_ods_gmpchange >> task_ods_aotai_ums >> task_ods_aotai_sys >> task_dim_level
task_dim_level >> task_dim_gmpdoc_doc >> task_dim_gmpdoc_doc_history >> task_dwd_level
task_dim_level >> task_dim_operation_om >> task_dim_quatrust_operation_executor >> task_dim_quatrust_operation_module >> task_dim_quatrust_operation_principal >> task_dwd_level
task_dim_level >> task_dim_quatrust_user_department >> task_dim_quatrust_all_user_department >> task_dim_quatrust_user_role >> task_dwd_level
task_dws_level >> task_ads_quatrust_user_department >> task_ads_quatrust_all_user_department >> task_ads_quatrust_user_role >> task_ads_level
task_dws_level >> task_ads_quatrust_user >> task_ads_quatrust_user_soft >> task_ads_level
task_dws_level >> task_ads_gmpdoc_doc_task_detail >> task_ads_gmpdoc_exexute_task_detail >> task_ads_gmpdoc_dim_doc >> task_ads_gmpdoc_dim_doc_nonjoinder >> task_ads_level
task_dws_level >> task_ads_gmpdoc_dim_doc_history >> task_ads_level
task_dws_level >> task_ads_gmpdoc_doc_withdraw_in_approval >> task_ads_gmpdoc_doc_withdraw_action >> task_ads_level
task_dws_level >> task_ads_quatrust_operation_executor >> task_ads_quatrust_operation_module >> task_ads_quatrust_operation_principal >> task_ads_level
task_dws_level >> task_ads_vdr_doc_hierarchical_folder >> task_ads_vdr_hierarchical_folder >> task_ads_level