from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from scripts.extract_data import fetch_movie_data
from scripts.transform_data import clean_and_normalize
from scripts.compute_ms import calculate_movie_score

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2026, 4, 1),
    'retries': 1,
}

with DAG('movie_score_pipeline', default_args=default_args, schedule_interval='@daily', catchup=False) as dag:

    t1 = PythonOperator(
        task_id='extract_data',
        python_callable=fetch_movie_data
    )

    t2 = PythonOperator(
        task_id='transform_data',
        python_callable=clean_and_normalize
    )

    t3 = PythonOperator(
        task_id='compute_ms',
        python_callable=calculate_movie_score
    )

    t1 >> t2 >> t3