100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
"""
|
|
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,
|
|
)
|