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_gxptrain_lesson import dim_gxptrain_lesson
from cloudengine.dwd.dim_gxptrain_real_mock_plan_lesson import dim_gxptrain_real_mock_plan_lesson
from cloudengine.dwd.dim_gxptrain_training_plan import dim_gxptrain_training_plan
from cloudengine.dwd.dim_quatrust_user_department import dim_quatrust_user_department
from cloudengine.dwd.dim_quatrust_user_department_lesson import dim_quatrust_user_department_lesson

from cloudengine.dwd.dwd_gxptrain_lesson_learn import dwd_gxptrain_lesson_learn
from cloudengine.dwd.dwd_gxptrain_brand_course import dwd_gxptrain_brand_course
from cloudengine.dwd.dwd_gxptrain_label import dwd_gxptrain_label
from cloudengine.dwd.dwd_gxptrain_exam_participants import dwd_gxptrain_exam_participants

from cloudengine.dws.dws_gxptrain_exam_participants_amount import dws_gxptrain_exam_participants_amount
from cloudengine.dws.dws_gxptrain_lesson_amount import dws_gxptrain_lesson_amount
from cloudengine.dws.dws_gxptrain_lesson_learn_amount import dws_gxptrain_lesson_learn_amount
from cloudengine.dws.dws_gxptrain_plan_amount import dws_gxptrain_plan_amount

from cloudengine.ads.ads_gxptrain_lesson_learn import ads_gxptrain_lesson_learn
from cloudengine.ads.ads_mock_gxptrain_lesson_learn import ads_mock_gxptrain_lesson_learn
from cloudengine.ads.ads_quatrust_user_department import ads_quatrust_user_department
from cloudengine.ads.ads_quatrust_user_department_lesson import ads_quatrust_user_department_lesson
from cloudengine.ads.ads_gxptrain_training_plan import ads_gxptrain_training_plan
from cloudengine.ads.ads_gxptrain_label import ads_gxptrain_label
from cloudengine.ads.ads_gxptrain_exam_participants_amount import ads_gxptrain_exam_participants_amount
from cloudengine.ads.ads_gxptrain_lesson_amount import ads_gxptrain_lesson_amount
from cloudengine.ads.ads_gxptrain_lesson_learn_amount import ads_gxptrain_lesson_learn_amount
from cloudengine.ads.ads_gxptrain_plan_amount import ads_gxptrain_plan_amount

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"

gxp_training_flow_interval = config[CONF_SECTION]['gxp_training_flow']
gxp_training_flow_dag = DAG('gxp_training_flow', default_args=default_args,
                         schedule_interval=gxp_training_flow_interval)


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

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

def gxptrain_ods_load():
    extract.extract_group("gxptrain")

task_ods_gxptrain = PythonOperator(
    task_id='ods_gxptrain',
    python_callable=gxptrain_ods_load,
    dag=gxp_training_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=gxp_training_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=gxp_training_flow_dag)

def traincentermanage_ods_load():
    extract.extract_group("traincentermanage")

task_ods_traincentermanage = PythonOperator(
    task_id='ods_traincentermanage',
    python_callable=traincentermanage_ods_load,
    dag=gxp_training_flow_dag)

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

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

task_dim_gxptrain_lesson = PythonOperator(
    task_id='dim_gxptrain_lesson',
    python_callable=dim_gxptrain_lesson.process,
    dag=gxp_training_flow_dag)

task_dim_gxptrain_real_mock_plan_lesson = PythonOperator(
    task_id='dim_gxptrain_real_mock_plan_lesson',
    python_callable=dim_gxptrain_real_mock_plan_lesson.process,
    dag=gxp_training_flow_dag)

task_dim_gxptrain_training_plan = PythonOperator(
    task_id='dim_gxptrain_training_plan',
    python_callable=dim_gxptrain_training_plan.process,
    dag=gxp_training_flow_dag)

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

task_dim_quatrust_user_department_lesson = PythonOperator(
    task_id='dim_quatrust_user_department_lesson',
    python_callable=dim_quatrust_user_department_lesson.process,
    dag=gxp_training_flow_dag)

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

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

task_dwd_gxptrain_lesson_learn = PythonOperator(
    task_id='dwd_gxptrain_lesson_learn',
    python_callable=dwd_gxptrain_lesson_learn.process,
    dag=gxp_training_flow_dag)

task_dwd_gxptrain_brand_course = PythonOperator(
    task_id='dwd_gxptrain_brand_course',
    python_callable=dwd_gxptrain_brand_course.process,
    dag=gxp_training_flow_dag)

task_dwd_gxptrain_label = PythonOperator(
    task_id='dwd_gxptrain_label',
    python_callable=dwd_gxptrain_label.process,
    dag=gxp_training_flow_dag)

task_dwd_gxptrain_exam_participants = PythonOperator(
    task_id='dwd_gxptrain_exam_participants',
    python_callable=dwd_gxptrain_exam_participants.process,
    dag=gxp_training_flow_dag)


def dws_level_load():
    pass

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

task_dws_gxptrain_exam_participants_amount = PythonOperator(
    task_id='dws_gxptrain_exam_participants_amount',
    python_callable=dws_gxptrain_exam_participants_amount.process,
    dag=gxp_training_flow_dag)

task_dws_gxptrain_lesson_amount = PythonOperator(
    task_id='dws_gxptrain_lesson_amount',
    python_callable=dws_gxptrain_lesson_amount.process,
    dag=gxp_training_flow_dag)

task_dws_gxptrain_lesson_learn_amount = PythonOperator(
    task_id='dws_gxptrain_lesson_learn_amount',
    python_callable=dws_gxptrain_lesson_learn_amount.process,
    dag=gxp_training_flow_dag)

task_dws_gxptrain_plan_amount = PythonOperator(
    task_id='dws_gxptrain_plan_amount',
    python_callable=dws_gxptrain_plan_amount.process,
    dag=gxp_training_flow_dag)

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

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

task_ads_gxptrain_lesson_learn = PythonOperator(
    task_id='ads_gxptrain_lesson_learn',
    python_callable=ads_gxptrain_lesson_learn.process,
    dag=gxp_training_flow_dag)

task_ads_mock_gxptrain_lesson_learn = PythonOperator(
    task_id='ads_mock_gxptrain_lesson_learn',
    python_callable=ads_mock_gxptrain_lesson_learn.process,
    dag=gxp_training_flow_dag)

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

task_ads_quatrust_user_department_lesson = PythonOperator(
    task_id='ads_quatrust_user_department_lesson',
    python_callable=ads_quatrust_user_department_lesson.process,
    dag=gxp_training_flow_dag)

task_ads_gxptrain_training_plan = PythonOperator(
    task_id='ads_gxptrain_training_plan',
    python_callable=ads_gxptrain_training_plan.process,
    dag=gxp_training_flow_dag)

task_ads_gxptrain_label = PythonOperator(
    task_id='ads_gxptrain_label',
    python_callable=ads_gxptrain_label.process,
    dag=gxp_training_flow_dag)

task_ads_gxptrain_exam_participants_amount = PythonOperator(
    task_id='ads_gxptrain_exam_participants_amount',
    python_callable=ads_gxptrain_exam_participants_amount.process,
    dag=gxp_training_flow_dag)

task_ads_gxptrain_lesson_amount = PythonOperator(
    task_id='ads_gxptrain_lesson_amount',
    python_callable=ads_gxptrain_lesson_amount.process,
    dag=gxp_training_flow_dag)

task_ads_gxptrain_lesson_learn_amount = PythonOperator(
    task_id='ads_gxptrain_lesson_learn_amount',
    python_callable=ads_gxptrain_lesson_learn_amount.process,
    dag=gxp_training_flow_dag)

task_ads_gxptrain_plan_amount = PythonOperator(
    task_id='ads_gxptrain_plan_amount',
    python_callable=ads_gxptrain_plan_amount.process,
    dag=gxp_training_flow_dag)


###  tableau_data_extract

def gxptrain_tableau_data_extract_task():
    refresh_data = tableau_data_extract("gxptrain_flow")
    refresh_data.run_job()

task_gxptrain_tableau_data_extract = PythonOperator(
    task_id='gxptrain_tableau_data_extract_task',
    python_callable=gxptrain_tableau_data_extract_task,
    dag=gxp_training_flow_dag)

###  tableau level
def tableau_level_load():
    pass

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

task_ods_level >> [task_ods_gxptrain, task_ods_aotai_sys, task_ods_aotai_ums, task_ods_traincentermanage] >> task_dim_level >> [task_dim_gxptrain_lesson,
                                                                                    task_dim_gxptrain_real_mock_plan_lesson,
                                                                                    task_dim_gxptrain_training_plan,
                                                                                    task_dim_quatrust_user_department_lesson,
                                                                                    task_dim_quatrust_user_department] >> task_dwd_level >> [task_dwd_gxptrain_lesson_learn, task_dwd_gxptrain_brand_course, task_dwd_gxptrain_label, task_dwd_gxptrain_exam_participants] >> task_dws_level >> [task_dws_gxptrain_exam_participants_amount,
task_dws_gxptrain_lesson_amount,
task_dws_gxptrain_lesson_learn_amount,
task_dws_gxptrain_plan_amount] >> task_ads_level >> [
    task_ads_gxptrain_lesson_learn,
    task_ads_mock_gxptrain_lesson_learn,
    task_ads_gxptrain_training_plan,
    task_ads_gxptrain_label,
    task_ads_quatrust_user_department_lesson,
    task_ads_quatrust_user_department,
    task_ads_gxptrain_exam_participants_amount,
task_ads_gxptrain_lesson_amount,
task_ads_gxptrain_lesson_learn_amount,
task_ads_gxptrain_plan_amount] >> task_tableau_level >> task_gxptrain_tableau_data_extract