From 6b1b2832a16e6d514c73c1e4a6db170ee131f597 Mon Sep 17 00:00:00 2001
From: claude <claude@redneurocom.com>
Date: Wed, 29 Jul 2026 06:49:50 +0200
Subject: [PATCH] feat: DAG corotos_do_daily (08:30 UTC)

---
 dags/corotos_do_daily.py | 77 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 77 insertions(+)
 create mode 100644 dags/corotos_do_daily.py

diff --git a/dags/corotos_do_daily.py b/dags/corotos_do_daily.py
new file mode 100644
index 0000000..aab9c10
--- /dev/null
+++ b/dags/corotos_do_daily.py
@@ -0,0 +1,77 @@
+"""
+DAG: corotos_do_daily
+Scraper diario de inmuebles de Corotos.com.do (Republica Dominicana).
+Ejecuta el scraper via SSH sobre el ARM VPS host (152.53.242.31).
+"""
+from datetime import datetime, timedelta
+
+import paramiko
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+
+SSH_HOST     = '152.53.242.31'
+SSH_PORT     = 22
+SSH_USER     = 'root'
+SSH_PASS     = 'XpvvdHnNr0Bq9qU'
+PYTHON_BIN   = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
+SCRAPER_PATH = '/opt/scrapers/corotos_do/scraper.py'
+
+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),
+}
+
+
+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,
+    )
+    client.get_transport().set_keepalive(30)
+
+    cmd = f'{PYTHON_BIN} {SCRAPER_PATH}'
+    _, stdout, stderr = client.exec_command(cmd, timeout=21600)
+
+    log_out = stdout.read().decode('utf-8', errors='replace')
+    log_err = stderr.read().decode('utf-8', errors='replace')
+    client.close()
+
+    print('=== STDOUT ===')
+    print(log_out)
+    if log_err.strip():
+        print('=== STDERR ===')
+        print(log_err)
+
+    full_log = log_out + log_err
+    if 'Finalizado:' not in full_log:
+        raise RuntimeError(
+            "Scraper corotos_do no imprimio 'Finalizado:'. "
+            f"Ultimas 500 chars: {full_log[-500:]!r}"
+        )
+
+
+with DAG(
+    dag_id='corotos_do_daily',
+    default_args=default_args,
+    description='Scraper diario de inmuebles Corotos.com.do (Republica Dominicana)',
+    schedule_interval='30 8 * * *',
+    start_date=datetime(2025, 1, 1),
+    catchup=False,
+    max_active_runs=1,
+    tags=['scraping', 'corotos', 'do', 'inmuebles'],
+) as dag:
+
+    t_scrape = PythonOperator(
+        task_id='run_corotos_do_scraper',
+        python_callable=run_scraper,
+    )
\ No newline at end of file