from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
import sys
sys.path.insert(0, '/opt/airflow/dags/save_url_ftp')
from SaveUrl import run_save_url

# ==========================================
# CONFIGURACIÓN DEL DAG
# ==========================================
DAG_ID          = 'save_url_ftp'
DAG_DESCRIPTION = 'Genera y guarda URLs del FTP en SQL Server para todas las BDs'
DAG_SCHEDULE =  '0 */2 * * *'  # Cada 2 horas El formato cron es minuto hora día mes día_semana
TAGS            = ['ftp', 'url', 'sql', 'kafka', 'notification', 'glmpmi', 'glmccf', 'glmea', 'glmcp', 'glmcl', 'is001', 'is003', 'is004', 'is005']

default_args = {
    'start_date':   datetime(2026, 4, 1),
    'retries':      2,
    'retry_delay':  timedelta(minutes=5),
}

with DAG(
    dag_id=DAG_ID,
    description=DAG_DESCRIPTION,
    schedule_interval=DAG_SCHEDULE,
    default_args=default_args,
    catchup=False,
    max_active_runs=1,
    tags=TAGS,
) as dag:

    start = EmptyOperator(task_id='inicio')
    end   = EmptyOperator(task_id='fin')

    save_urls = PythonOperator(
        task_id='save_url_ftp',
        python_callable=run_save_url,
    )

    start >> save_urls >> end