feat: DAG elempleo_co_daily (05:00 UTC)

2026-07-29 06:49:43 +02:00
parent 273c17a04a
commit a6ac875820
1 changed files with 99 additions and 0 deletions
@@ -0,0 +1,99 @@
+"""
+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,
+    )