"""
DAG: elempleo_co_daily
Ejecuta el scraper de ElEmpleo.com Colombia via SSH en el ARM VPS (152.53.242.31).
Schedule  : 05:00 UTC diario
Autor     : jpuma
"""

from datetime import datetime, timedelta

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

# ---------------------------------------------------------------------------
# Constantes de conexión
# ---------------------------------------------------------------------------
VPS_HOST    = '152.53.242.31'
VPS_PORT    = 22
VPS_USER    = 'root'
VPS_PASS    = 'XpvvdHnNr0Bq9qU'

VENV_PYTHON  = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
SCRAPER_PATH = '/opt/scrapers/elempleo_co/scraper.py'

# ---------------------------------------------------------------------------
# default_args
# ---------------------------------------------------------------------------
default_args = {
    'owner'            : 'jpuma',
    'retries'          : 2,
    'retry_delay'      : timedelta(minutes=15),
    'email'            : ['jpuma@redneurocom.com'],
    'email_on_failure' : True,
    'email_on_retry'   : False,
    'execution_timeout': timedelta(hours=6),
}


# ---------------------------------------------------------------------------
# Callable
# ---------------------------------------------------------------------------
def run_scraper(**ctx):
    """Conecta al VPS via SSH y ejecuta el scraper. Valida el patrón de éxito."""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(
        hostname=VPS_HOST,
        port=VPS_PORT,
        username=VPS_USER,
        password=VPS_PASS,
        timeout=30,
    )
    ssh.get_transport().set_keepalive(30)

    cmd = '{} {} 2>&1'.format(VENV_PYTHON, SCRAPER_PATH)

    # exec_command con timeout de 6 horas (igual que execution_timeout del DAG)
    _, stdout, _ = ssh.exec_command(cmd, timeout=21600)
    log       = stdout.read().decode('utf-8', errors='replace')
    exit_code = stdout.channel.recv_exit_status()
    ssh.close()

    # Volcamos el log completo para visibilidad en Airflow
    print("===== LOG SCRAPER ELEMPLEO =====")
    print(log)
    print("===== EXIT CODE: {} =====".format(exit_code))

    # Validación del patrón de éxito obligatorio
    if 'Finalizado:' not in log:
        raise RuntimeError(
            "Scraper ElEmpleo no finalizó correctamente "
            "(exit={}). Ultimas lineas: {}".format(exit_code, log[-800:])
        )

    if exit_code != 0:
        raise RuntimeError(
            "Scraper terminó con exit code {}. Log: {}".format(exit_code, log[-800:])
        )


# ---------------------------------------------------------------------------
# DAG
# ---------------------------------------------------------------------------
with DAG(
    dag_id            = 'elempleo_co_daily',
    default_args      = default_args,
    description       = 'Scraper diario de ofertas de empleo — ElEmpleo.com Colombia',
    schedule_interval = '0 5 * * *',
    start_date        = datetime(2025, 1, 1),
    catchup           = False,
    max_active_runs   = 1,
    tags              = ['scraping', 'elempleo', 'co', 'empleos'],
) as dag:

    scrape_task = PythonOperator(
        task_id         = 'run_elempleo_scraper',
        python_callable = run_scraper,
        provide_context = True,
    )