"""DAG for adjustment file upload process."""

from datetime import datetime

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

from lib import constants
from tasks.adjustment_file_upload.adjustment_file_validation_check import \
    adjustment_file_validation_check
from tasks.adjustment_file_upload.check_adjustment_file_error_report import \
    check_adjustment_file_error_report_task
from tasks.adjustment_file_upload.invoke_adjustment_file_validation_lambda import \
    invoke_adjustment_file_validation_lambda_task
from tasks.adjustment_file_upload.notify_failure import notify_failure_task
from tasks.adjustment_file_upload.notify_started import notify_started_task
from tasks.adjustment_file_upload.notify_success import notify_success_task
from tasks.adjustment_file_upload.validate_format import validate_format_task


dag = DAG(
    constants.DAG_ADJUSTMENT_FILE_UPLOAD_NAME,
    description='Handles adjustment file upload flow and move the file to S3.',  # noqa: E501
    start_date=datetime(2023, 1, 1),
    default_args={'provide_context': True},
    schedule_interval=None
)


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

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

invoke_adjustment_file_validation_lambda_operator = PythonOperator(
    task_id='invoke_adjustment_file_validation_lambda',
    python_callable=invoke_adjustment_file_validation_lambda_task,
    dag=dag
)

adjustment_file_validation_check_result_operator = PythonSensor(
    task_id='adjustment_file_validation_check_result',
    dag=dag,
    python_callable=adjustment_file_validation_check,
    timeout=900,
    poke_interval=30,
)

check_adjustment_file_error_report_operator = PythonOperator(
    task_id='check_adjustment_file_error_report',
    python_callable=check_adjustment_file_error_report_task,
    dag=dag
)

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

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

validate_format_operator = PythonOperator(
    dag=dag,
    python_callable=validate_format_task,
    task_id='validate_format'
)

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

notify_started_operator >>\
    validate_format_operator >>\
    invoke_adjustment_file_validation_lambda_operator >>\
    adjustment_file_validation_check_result_operator >>\
    check_adjustment_file_error_report_operator >>\
    notify_success_operator >>\
    notify_failure_operator