"""DAG for closing an accounting period."""

from datetime import datetime

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

from lib import constants
from tasks.accounting_period_close.close_accounting_period \
    import close_accounting_period_task
from tasks.accounting_period_close.snowflake_rollover_unallocated_sales \
    import snowflake_rollover_unallocated_sales_task
from tasks.accounting_period_close.update_abacus_state import update_abacus_state_task

dag = DAG(
    constants.DAG_ACCOUNTING_PERIOD_CLOSE_NAME,
    description='Close an accounting period: automatically rollover unallocated sales',
    start_date=datetime(2025, 1, 1),
    default_args={'provide_context': True},
    schedule_interval=None
)

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

close_accounting_period_operator = PythonOperator(
    dag=dag,
    python_callable=close_accounting_period_task,
    task_id='close_accounting_period'
)

notify_failure_operator = PythonOperator(
    dag=dag,
    op_kwargs={'action_status': constants.ABACUS_STATE_STATUSES.ERROR},
    python_callable=update_abacus_state_task,
    task_id='notify_failure',
    trigger_rule='one_failed'
)

notify_started_operator = PythonOperator(
    dag=dag,
    op_kwargs={'action_status': constants.ABACUS_STATE_STATUSES.RUNNING},
    python_callable=update_abacus_state_task,
    task_id='notify_started'
)

notify_success_operator = PythonOperator(
    dag=dag,
    op_kwargs={'action_status': constants.ABACUS_STATE_STATUSES.COMPLETE},
    python_callable=update_abacus_state_task,
    task_id='notify_success',
    trigger_rule='none_failed'
)

rollover_unallocated_sales_operator = PythonOperator(
    dag=dag,
    python_callable=snowflake_rollover_unallocated_sales_task,
    task_id='rollover_unallocated_sales'
)

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

notify_started_operator >> \
    rollover_unallocated_sales_operator >> \
    close_accounting_period_operator >> [
        notify_success_operator,
        notify_failure_operator
    ]