"""DAG for prepping contract's mechanical deductions."""

from datetime import datetime

from airflow import DAG
from airflow.operators.python import PythonOperator

from lib import constants
from lib.utils.slack import slack_failure_callback
from tasks.accounting_period_mechanicals.notify_failure import notify_failure_task
from tasks.accounting_period_mechanicals.notify_started import notify_started_task
from tasks.accounting_period_mechanicals.notify_success import notify_success_task
from tasks.accounting_period_mechanicals.snowflake_calculate_mech_royalty_amounts \
    import snowflake_calculate_mechanical_royalty_amounts_task
from tasks.accounting_period_mechanicals.snowflake_snapshot_mech_txn_details \
    import snowflake_snapshot_mechanical_transaction_details_task
from tasks.accounting_period_mechanicals.snowflake_snapshot_mech_txns \
    import snowflake_snapshot_mechanical_transactions_task

dag = DAG(
    constants.DAG_MECHANICALS_EVENT_NAME,
    description='Prep mechanical deductions',
    start_date=datetime(2019, 1, 1),
    default_args={
        'provide_context': True,
        'on_failure_callback': slack_failure_callback,
    },
    schedule_interval=None
)


"""
######################################################################
############################# OPERATORS ##############################
######################################################################
"""

calculate_mechanical_royalty_amounts_operator = PythonOperator(
    dag=dag,
    python_callable=snowflake_calculate_mechanical_royalty_amounts_task,
    task_id='calculate_mechanical_royalty_amounts'
)

notify_failure_operator = PythonOperator(
    dag=dag,
    task_id='notify_failure',
    trigger_rule='one_failed',
    python_callable=notify_failure_task
)

notify_started_operator = PythonOperator(
    dag=dag,
    python_callable=notify_started_task,
    task_id='notify_started'
)

notify_success_operator = PythonOperator(
    dag=dag,
    python_callable=notify_success_task,
    task_id='notify_success',
    trigger_rule='none_failed'
)

snapshot_txns_eligible_for_mech_deductions_operator = PythonOperator(
    dag=dag,
    python_callable=snowflake_snapshot_mechanical_transactions_task,
    task_id='snapshot_txns_eligible_for_mech_deductions'
)

snapshot_album_mechanical_transaction_details_operator = PythonOperator(
    dag=dag,
    op_kwargs={'album_or_track': constants.ALBUM},
    python_callable=snowflake_snapshot_mechanical_transaction_details_task,
    task_id='snapshot_album_mechanical_transaction_details'
)

snapshot_track_mechanical_transaction_details_operator = PythonOperator(
    dag=dag,
    op_kwargs={'album_or_track': constants.TRACK},
    python_callable=snowflake_snapshot_mechanical_transaction_details_task,
    task_id='snapshot_track_mechanical_transaction_details'
)


"""
######################################################################
################################ DAG #################################
######################################################################
"""


notify_started_operator >> \
    snapshot_txns_eligible_for_mech_deductions_operator >> \
    snapshot_album_mechanical_transaction_details_operator >> \
    snapshot_track_mechanical_transaction_details_operator >> \
    calculate_mechanical_royalty_amounts_operator >> \
    notify_success_operator >> \
    notify_failure_operator