68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""
|
|
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,
|
|
)
|