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_training_lesson import dim_training_lesson
from cloudengine.dwd.dim_training_lesson_learn import dim_training_lesson_learn
from cloudengine.dwd.dim_gmpdoc_doc import dim_gmpdoc_doc
from cloudengine.dwd.dim_gmpchange_changemain import dim_gmpchange_changemain
from cloudengine.dwd.dim_training_training_plan import dim_training_training_plan
from cloudengine.dwd.dim_gmpaudit_audit_main_plan import dim_gmpaudit_audit_main_plan
from cloudengine.dwd.dim_gmpaudit_audit_process import dim_gmpaudit_audit_process
from cloudengine.dwd.dim_gmpcapa_capa import dim_gmpcapa_capa
from cloudengine.dwd.dim_gmpdeviation_deviation import dim_gmpdeviation_deviation
from cloudengine.dwd.dim_gmpdiscrepancy_discrepancy import dim_gmpdiscrepancy_discrepancy
from cloudengine.dwd.dim_operation_om import dim_operation_om
from cloudengine.dwd.dim_gmpvalidation_validation import dim_gmpvalidation_validation
from cloudengine.dwd.dim_quatrust_task import dim_quatrust_task
from cloudengine.dwd.dim_quatrust_user import dim_quatrust_user
from cloudengine.dwd.dim_signtrust_contracts import dim_signtrust_contracts
from cloudengine.dwd.dim_taskmanage_sig import dim_taskmanage_sig
from cloudengine.dwd.dim_taskmanage_information import dim_taskmanage_information
from cloudengine.dwd.dim_taskmanage_task import dim_taskmanage_task
from cloudengine.dwd.dim_quatrust_company import dim_quatrust_company

from cloudengine.dws.dws_quatrust_user_amount import dws_quatrust_user_amount
from cloudengine.dws.dws_quatrust_user_amount_week_chain import dws_quatrust_user_amount_week_chain

from cloudengine.ads.ads_quatrust_unfinished_task import ads_quatrust_unfinished_task
from cloudengine.ads.ads_quatrust_all_task import ads_quatrust_all_task
from cloudengine.ads.ads_quatrust_user_amount import ads_quatrust_user_amount
from cloudengine.ads.ads_quatrust_user_amount_week_chain import ads_quatrust_user_amount_week_chain
from cloudengine.ads.ads_quatrust_company import ads_quatrust_company



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"

gmptask_flow_interval = config[CONF_SECTION]['gmptask_flow']
gmptask_flow_dag = DAG('gmptask_flow', default_args=default_args,
                         schedule_interval=gmptask_flow_interval)


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


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


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


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


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


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


def gmptrain_ods_load():
    extract.extract_group("gmptrain")


task_ods_gmptrain = PythonOperator(
    task_id='ods_gmptrain',
    python_callable=gmptrain_ods_load,
    dag=gmptask_flow_dag)


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


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

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

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

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

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

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

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

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

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

def signtrust_ods_load():
    extract.extract_group("signtrust")

task_ods_signtrust = PythonOperator(
    task_id='ods_signtrust',
    python_callable=signtrust_ods_load,
    dag=gmptask_flow_dag)

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

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

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

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

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

task_ods_aotai_ums = PythonOperator(
    task_id='ods_aotai_ums',
    python_callable=ods_aotai_ums_ods_load,
    dag=gmptask_flow_dag)

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

task_ods_aotai_sys = PythonOperator(
    task_id='ods_aotai_sys',
    python_callable=ods_aotai_sys_ods_load,
    dag=gmptask_flow_dag)

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

task_ods_centermanage = PythonOperator(
    task_id='ods_centermanage',
    python_callable=ods_centermanage_ods_load,
    dag=gmptask_flow_dag)

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

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

task_dim_training_lesson = PythonOperator(
    task_id='dim_training_lesson',
    python_callable=dim_training_lesson.process,
    dag=gmptask_flow_dag)

task_dim_training_lesson_learn = PythonOperator(
    task_id='dim_training_lesson_learn',
    python_callable=dim_training_lesson_learn.process,
    dag=gmptask_flow_dag)

task_dim_training_training_plan = PythonOperator(
    task_id='dim_training_training_plan',
    python_callable=dim_training_training_plan.process,
    dag=gmptask_flow_dag)

task_dim_gmpvalidation_validation = PythonOperator(
    task_id='dim_gmpvalidation_validation',
    python_callable=dim_gmpvalidation_validation.process,
    dag=gmptask_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=gmptask_flow_dag)

task_dim_gmpaudit_audit_process = PythonOperator(
    task_id='dim_gmpaudit_audit_process',
    python_callable=dim_gmpaudit_audit_process.process,
    dag=gmptask_flow_dag)

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

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

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

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

task_dim_signtrust_contracts = PythonOperator(
    task_id='dim_signtrust_contracts',
    python_callable=dim_signtrust_contracts.process,
    dag=gmptask_flow_dag)

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

task_dim_quatrust_task = PythonOperator(
    task_id='dim_quatrust_task',
    python_callable=dim_quatrust_task.process,
    dag=gmptask_flow_dag)

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

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

task_dim_taskmanage_sig = PythonOperator(
    task_id='dim_taskmanage_sig',
    python_callable=dim_taskmanage_sig.process,
    dag=gmptask_flow_dag)

task_dim_taskmanage_information = PythonOperator(
    task_id='dim_taskmanage_information',
    python_callable=dim_taskmanage_information.process,
    dag=gmptask_flow_dag)

task_dim_taskmanage_task = PythonOperator(
    task_id='dim_taskmanage_task',
    python_callable=dim_taskmanage_task.process,
    dag=gmptask_flow_dag)

task_dim_quatrust_company = PythonOperator(
    task_id='dim_quatrust_company',
    python_callable=dim_quatrust_company.process,
    dag=gmptask_flow_dag)

###  dwd level
def dwd_level_load():
    pass

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

def dws_level_load():
    pass

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

task_dws_quatrust_user_amount = PythonOperator(
    task_id='dws_quatrust_user_amount',
    python_callable=dws_quatrust_user_amount.process,
    dag=gmptask_flow_dag)

task_dws_quatrust_user_amount_week_chain = PythonOperator(
    task_id='dws_quatrust_user_amount_week_chain',
    python_callable=dws_quatrust_user_amount_week_chain.process,
    dag=gmptask_flow_dag)


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


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

task_ads_quatrust_unfinished_task = PythonOperator(
    task_id='ads_quatrust_unfinished_task',
    python_callable=ads_quatrust_unfinished_task.process,
    dag=gmptask_flow_dag)

task_ads_quatrust_all_task = PythonOperator(
    task_id='ads_quatrust_all_task',
    python_callable=ads_quatrust_all_task.process,
    dag=gmptask_flow_dag)

task_ads_quatrust_user_amount = PythonOperator(
    task_id='ads_quatrust_user_amount',
    python_callable=ads_quatrust_user_amount.process,
    dag=gmptask_flow_dag)


task_ads_quatrust_user_amount_week_chain = PythonOperator(
    task_id='ads_quatrust_user_amount_week_chain',
    python_callable=ads_quatrust_user_amount_week_chain.process,
    dag=gmptask_flow_dag)

task_ads_quatrust_company = PythonOperator(
    task_id='ads_quatrust_company',
    python_callable=ads_quatrust_company.process,
    dag=gmptask_flow_dag)

####  tableau_data_extract
def tableau_data_extract_task():
    refresh_data = tableau_data_extract("gmptask_flow")
    refresh_data.run_job()

task_tableau_data_extract = PythonOperator(
    task_id='tableau_data_extract_task',
    python_callable=tableau_data_extract_task,
    dag=gmptask_flow_dag)

def user_tableau_data_extract_task():
    refresh_data = tableau_data_extract("user_flow")
    refresh_data.run_job()

task_user_tableau_data_extract = PythonOperator(
    task_id='user_tableau_data_extract_task',
    python_callable=user_tableau_data_extract_task,
    dag=gmptask_flow_dag)

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

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

task_ods_level >> [task_ods_signtrust, task_ods_gmpdoc, task_ods_gmptrain, task_ods_centermanage] >> task_dim_level >> [task_dim_gmpdoc_doc, task_dim_taskmanage_task] >> task_dwd_level >> [
    task_dws_quatrust_user_amount, task_dws_quatrust_user_amount_week_chain] >> task_dws_level >> [
    task_ads_quatrust_unfinished_task, task_ads_quatrust_all_task, task_ads_quatrust_user_amount, task_ads_quatrust_user_amount_week_chain, task_ads_quatrust_company] >> task_ads_level >> [task_tableau_data_extract, task_user_tableau_data_extract] >> task_tableau_level
task_ods_level >> [task_ods_aotai_ums, task_ods_aotai_sys] >> task_dim_level
task_ods_level >> task_ods_quatrust >> task_dim_level
task_ods_level >> task_ods_gmpdisc >> task_ods_gmpaudit >> task_ods_gmpdevi >> task_ods_gmpchange >> task_dim_level
task_ods_level >> task_ods_gmpvalid >> task_ods_gmpcapa >> task_ods_operation >> task_dim_level
task_dim_level >> task_dim_training_lesson >> task_dim_training_lesson_learn >> task_dim_training_training_plan >> task_dwd_level
task_dim_level >> task_dim_gmpaudit_audit_main_plan >> task_dim_gmpaudit_audit_process >> task_dim_signtrust_contracts >> task_dwd_level
task_dim_level >> task_dim_quatrust_task >> task_dim_quatrust_user >> task_dim_quatrust_company >> task_dwd_level
task_dim_level >> task_dim_gmpcapa_capa >> task_dim_gmpdeviation_deviation >> task_dim_gmpdiscrepancy_discrepancy >> task_dim_gmpvalidation_validation >> task_dim_gmpchange_changemain >> task_dim_operation_om >> task_dwd_level
task_dim_level >> task_dim_taskmanage_sig >> task_dim_taskmanage_information >> task_dwd_level