diff --git a/dags/corotos_do_backfill.py b/dags/corotos_do_backfill.py new file mode 100644 index 0000000..c099e26 --- /dev/null +++ b/dags/corotos_do_backfill.py @@ -0,0 +1,67 @@ +""" +DAG: corotos_do_backfill +Backfill manual del catalogo completo de Corotos.com.do Republica Dominicana. +Reanudable: checkpoint en /opt/scrapers/corotos_do/backfill_state.json +Para reiniciar desde cero: borrar ese archivo antes de disparar el DAG. +""" +from datetime import datetime, timedelta +import paramiko +from airflow import DAG +from airflow.operators.python import PythonOperator + +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' +BACKFILL_PY = '/opt/scrapers/corotos_do/backfill.py' + +default_args = { + 'owner': 'jpuma', + 'retries': 0, + 'email': ['jpuma@redneurocom.com'], + 'email_on_failure': True, + 'email_on_retry': False, + 'execution_timeout': timedelta(hours=8), +} + + +def run_backfill(**ctx): + ssh = paramiko.SSHClient() + ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + ssh.connect(VPS_HOST, port=VPS_PORT, username=VPS_USER, password=VPS_PASS, timeout=30) + ssh.get_transport().set_keepalive(30) + + cmd = f'BF_MAX_PAGES=200 BF_STREAK=6 {VENV_PYTHON} {BACKFILL_PY} 2>&1' + _, stdout, _ = ssh.exec_command(cmd, timeout=28800) # 8 h + output = stdout.read().decode('utf-8', errors='replace') + exit_code = stdout.channel.recv_exit_status() + ssh.close() + + tail = output[-5000:] if len(output) > 5000 else output + print('[corotos_do_backfill] OUTPUT:\n' + tail) + + if 'Finalizado:' not in output: + raise RuntimeError( + f'Backfill no finalizó correctamente (exit={exit_code}). ' + f'Ultimas lineas: {output[-600:]}' + ) + if exit_code != 0: + raise RuntimeError(f'Backfill terminó con exit code {exit_code}') + + +with DAG( + dag_id = 'corotos_do_backfill', + default_args = default_args, + description = 'Backfill catalogo completo Corotos.com.do Rep. Dominicana — disparo manual, reanudable', + schedule_interval = None, + start_date = datetime(2026, 7, 1), + catchup = False, + max_active_runs = 1, + tags = ['backfill', 'corotos', 'do', 'inmuebles'], +) as dag: + + PythonOperator( + task_id = 'run_corotos_backfill', + python_callable = run_backfill, + )