"""
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
import os
import time
from datetime import timedelta

from airflow import DAG
from airflow.operators.subdag_operator import SubDagOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator, ShortCircuitOperator

import projects.youtube_reports_api.config as config
import projects.youtube_reports_api.utils as utils
from projects.youtube_reports_api.subdag import subdag, backfill_subdag

content_owner = config.ORCH

default_args = {
    'owner': 'lyin',
    'depends_on_past': False,
    'start_date': config.start_date,
    'email': ['datarequests@theorchard.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': config.retry_delay,
    'content_owner': content_owner,
    'content_owner_name': config.CMS_DICT.get(content_owner),
    'pool': config.pool
}

# IE dev_yt_reports_orchardMusic
dag_id = config.dag_id+'_'+default_args['content_owner_name']

dag = DAG(
    dag_id, 
    default_args= default_args, 
    schedule_interval= config.cron_schedule,
    dagrun_timeout= timedelta(hours=2))

dag.default_args['bootstrap'] = utils.kick_off(default_args['content_owner'])

# placeholder for the beginning of the DAG
head = DummyOperator(
    task_id= default_args['content_owner_name']+'_head',
    default_args= default_args,
    dag= dag)

# placeholder for the end of the DAG
tail = DummyOperator(
    task_id= default_args['content_owner_name']+'_tail',
    default_args= default_args,
    dag= dag)

# first "real" task, just cleans up the directory for new files.
reset = PythonOperator(
    task_id= 'reset',
    python_callable= utils.clear_stage,
    op_kwargs= {'content_owner': default_args['content_owner']},
    execution_timeout= timedelta(minutes=2),
    provide_context= False,
    retries=1,
    dag= dag)

# DAG Assignement via Bitshift Composition.
head >> reset

for report_context in dag.default_args['bootstrap']:
    """
    After the reset task, we iterate through each report
    (whose metadata is available in the bootstrap stored
    as a default argument) and dynamically generate
    subdags to downloaded and upload each report.

    Aside from the SubDagOperator, the subdag itself
    is treated similarly to the python_callable arg
    in the PythonOperator.

    The subdag can thus have args we ask for!
    """
    job_name = utils.get_job_name(report_context['job_name'])

    # this subdag is where downloading and uploading etc happens!
    subtask = SubDagOperator(
        subdag=subdag(
            dag_id,
            child_dag_name= job_name,
            default_args= dag.default_args,
            schedule_interval= config.cron_schedule,
            report_context= report_context),
        task_id= job_name,
        retries= 2,
        retry_delay= timedelta(seconds=10), 
        dag= dag)

    # DAG Assignement via Bitshift Composition.
    reset >> subtask >> tail

    # backfill reports on a 7 and 30 day window...
    if job_name in config.BACKFILL:
        backfill_task_7 = SubDagOperator(
            subdag=backfill_subdag(
                dag_id,
                child_dag_name= job_name + '_backfill_7',
                default_args= dag.default_args,
                schedule_interval= config.cron_schedule,
                report_context= report_context,
                time_delta=4),
            task_id= job_name + '_backfill_7',
            retries= 2,
            retry_delay= timedelta(seconds=10), 
            dag= dag)

        backfill_task_30 = SubDagOperator(
            subdag=backfill_subdag(
                dag_id,
                child_dag_name= job_name + '_backfill_30',
                default_args= dag.default_args,
                schedule_interval= config.cron_schedule,
                report_context= report_context,
                time_delta=27),
            task_id= job_name + '_backfill_30',
            retries= 2,
            retry_delay= timedelta(seconds=10), 
            dag= dag)

        tail >> backfill_task_7 >> backfill_task_30