# coding: utf-8

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2017, 12, 06),
    'email': ['kocsism@mapsolutions.hu'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 5,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}

dag = DAG('Site_Statistics_Aggregator', default_args=default_args,schedule_interval="0 2 * * *")

t1 = BashOperator(
    task_id='Tajolo_daily_property_stat',
    bash_command='python /var/local/site-statistics-loader/tajolo-site-stats/tajolo_property_stat_down_from_ga_db.py',
    dag=dag)

t2 = BashOperator(
    task_id='OT_daily_property_stat',
    bash_command='python /var/local/site-statistics-loader/ot-site-stats/ot_property_dily_stat_down_from_db.py',
    dag=dag)

t3 = BashOperator(
    task_id='daily_property_stat_union',
    bash_command='PGPASSWORD=oe5Su4Aaghae psql worker  -h 127.0.0.1 -d dwh -f /var/local/site-statistics-loader/site-stats-group/daily_property_pageview_group.sql',
    dag=dag)

t4 = BashOperator(
    task_id='property_analytics_update',
    bash_command='PGPASSWORD=oe5Su4Aaghae psql worker  -h 127.0.0.1 -d dwh -f /var/local/site-statistics-loader/site-stats-group/property_pageview_update.sql',
    dag=dag)

t5 = BashOperator(
    task_id='property_watch_stat',
    bash_command='python /var/local/site-statistics-loader/site-services/property_watch_stat.py',
    dag=dag)


t6 = BashOperator(
    task_id='OT_marketing_data',
    bash_command='python /var/local/site-statistics-loader/ot-site-stats/ot_marketing_data_daily.py',
    dag=dag)

t7 = BashOperator(
    task_id='IT_marketing_data',
    bash_command='python /var/local/site-statistics-loader/tajolo-site-stats/tajolo_marketing_data_daily.py',
    dag=dag)

t8 = BashOperator(
    task_id='konverzio_terv_update',
    bash_command='python /var/local/site-statistics-loader/site-stats-group/marketing_terv_upd.py',
    dag=dag)

t3.set_upstream([t1,t2])
t4.set_upstream(t3)
t8.set_upstream([t6,t7])