feat: DAG fincaraiz_co_daily (07:00 UTC)

2026-07-29 06:49:47 +02:00
parent a89a08d336
commit d67e9839b1
1 changed files with 87 additions and 0 deletions
@@ -0,0 +1,87 @@
+from datetime import datetime, timedelta
+
+import paramiko
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+
+# ---------------------------------------------------------------------------
+# CONFIG SSH + SCRAPER
+# ---------------------------------------------------------------------------
+SSH_HOST   = '152.53.242.31'
+SSH_PORT   = 22
+SSH_USER   = 'root'
+SSH_PASS   = 'XpvvdHnNr0Bq9qU'
+VENV_PY    = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
+SCRAPER_PY = '/opt/scrapers/fincaraiz_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),
+}
+
+
+# ---------------------------------------------------------------------------
+# TASK
+# ---------------------------------------------------------------------------
+def run_scraper(**ctx):
+    client = paramiko.SSHClient()
+    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    client.connect(
+        SSH_HOST,
+        port=SSH_PORT,
+        username=SSH_USER,
+        password=SSH_PASS,
+        timeout=30,
+    )
+    client.get_transport().set_keepalive(30)
+
+    # stderr redirigido a stdout para capturar todo en un solo buffer
+    cmd = f'{VENV_PY} {SCRAPER_PY} 2>&1'
+    _, stdout, _ = client.exec_command(cmd, timeout=21600)   # 6 h max
+    log_output = stdout.read().decode('utf-8', errors='replace')
+    exit_code  = stdout.channel.recv_exit_status()
+    client.close()
+
+    # Mostrar las últimas 5 000 chars del log en el task log de Airflow
+    tail = log_output[-5000:] if len(log_output) > 5000 else log_output
+    print('[fincaraiz_co] STDOUT+STDERR:\n' + tail)
+
+    if 'Finalizado:' not in log_output:
+        raise RuntimeError(
+            f'fincaraiz_co_scraper no imprimió "Finalizado:" '
+            f'(exit_code={exit_code}). '
+            f'Últimas líneas: {log_output[-400:]}'
+        )
+
+    if exit_code != 0:
+        raise RuntimeError(
+            f'fincaraiz_co_scraper terminó con exit_code={exit_code}'
+        )
+
+
+# ---------------------------------------------------------------------------
+# DAG
+# ---------------------------------------------------------------------------
+with DAG(
+    dag_id='fincaraiz_co_daily',
+    default_args=default_args,
+    description='Scraper diario FincaRaiz.com.co – venta y arriendo de apartamentos (Colombia)',
+    schedule_interval='0 7 * * *',
+    start_date=datetime(2025, 1, 1),
+    catchup=False,
+    max_active_runs=1,
+    tags=['scraping', 'fincaraiz', 'co', 'inmuebles'],
+) as dag:
+
+    scrape = PythonOperator(
+        task_id='run_fincaraiz_scraper',
+        python_callable=run_scraper,
+    )