wip: DAG trabajando_cl_daily

2026-07-29 06:49:46 +02:00
parent 0f63449b0a
commit a89a08d336
1 changed files with 72 additions and 0 deletions
@@ -0,0 +1,72 @@
+from datetime import datetime, timedelta
+import paramiko
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+    'owner': 'airflow',
+    'depends_on_past': False,
+    'retries': 2,
+    'retry_delay': timedelta(minutes=15),
+    'email': ['jpuma@redneurocom.com'],
+    'email_on_failure': True,
+    'email_on_retry': False,
+    'execution_timeout': timedelta(hours=6),
+}
+
+VENV_PYTHON  = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
+SCRAPER_PATH = '/opt/scrapers/trabajando_cl/scraper.py'
+SSH_HOST     = '152.53.242.31'
+SSH_PORT     = 22
+SSH_USER     = 'root'
+SSH_PASS     = 'XpvvdHnNr0Bq9qU'
+
+
+def run_scraper(**ctx):
+    client = paramiko.SSHClient()
+    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    client.connect(
+        hostname=SSH_HOST,
+        port=SSH_PORT,
+        username=SSH_USER,
+        password=SSH_PASS,
+        timeout=30,
+        look_for_keys=False,
+        allow_agent=False,
+    )
+    transport = client.get_transport()
+    transport.set_keepalive(30)
+
+    cmd = f'{VENV_PYTHON} {SCRAPER_PATH}'
+    _, stdout, stderr = client.exec_command(cmd, timeout=21600)
+
+    output = stdout.read().decode('utf-8', errors='replace')
+    errors = stderr.read().decode('utf-8', errors='replace')
+    client.close()
+
+    if errors.strip():
+        print(f"[STDERR]\n{errors}")
+    print(f"[STDOUT]\n{output}")
+
+    if 'Finalizado:' not in output:
+        raise RuntimeError(
+            f"Scraper trabajando_cl no finalizo correctamente. "
+            f"Tail output: {output[-800:]}"
+        )
+
+
+with DAG(
+    dag_id='trabajando_cl_daily',
+    default_args=DEFAULT_ARGS,
+    description='Scraper diario de ofertas de empleo en Trabajando.cl (Chile)',
+    schedule_interval='0 6 * * *',
+    start_date=datetime(2025, 1, 1),
+    catchup=False,
+    max_active_runs=1,
+    tags=['scraping', 'trabajando', 'cl', 'empleos'],
+) as dag:
+
+    scrape = PythonOperator(
+        task_id='run_trabajando_cl_scraper',
+        python_callable=run_scraper,
+    )