"""DAG de backfill – Metrocuadrado CO (Colombia) Reanudable via checkpoint en /opt/scrapers/metrocuadrado_co/state.json 156 categorías × (1 pagina + ~24 barrios) × ~55 listings cada una. """ from datetime import datetime, timedelta import paramiko from airflow import DAG from airflow.operators.python import PythonOperator VPS_HOST = '152.53.242.31' VPS_USER = 'root' VPS_KEY = '/opt/airflow/config/.ssh/id_rsa' SCRAPER_CMD = ( 'timeout 36000 ' '/home/jpuma/proyectos/bot-sunat/.venv/bin/python ' '/opt/scrapers/metrocuadrado_co/scraper.py 2>&1' ) def run_backfill(**ctx): client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect(VPS_HOST, username=VPS_USER, key_filename=VPS_KEY, timeout=30) _, stdout, stderr = client.exec_command(SCRAPER_CMD, timeout=36000) output = stdout.read().decode('utf-8', errors='replace') err = stderr.read().decode('utf-8', errors='replace') client.close() print(output[-6000:] if len(output) > 6000 else output) if err: print('STDERR:', err[-500:]) if 'Finalizado:' not in output: raise RuntimeError('Scraper no imprimió "Finalizado:" — posible error') return output.split('Finalizado:')[-1].strip() default_args = { 'owner': 'jpuma', 'retries': 1, 'retry_delay': timedelta(minutes=15), 'execution_timeout': timedelta(hours=11), } with DAG( dag_id='metrocuadrado_co_backfill', default_args=default_args, description='Backfill histórico Metrocuadrado CO — reanudable via checkpoint', schedule_interval=None, start_date=datetime(2026, 8, 1), catchup=False, max_active_runs=1, tags=['colombia', 'backfill', 'inmuebles'], ) as dag: PythonOperator( task_id='backfill_metrocuadrado_co', python_callable=run_backfill, )