diff --git a/dags/argenprop_ar_backfill.py b/dags/argenprop_ar_backfill.py new file mode 100644 index 0000000..59c17ff --- /dev/null +++ b/dags/argenprop_ar_backfill.py @@ -0,0 +1,57 @@ +"""DAG de backfill – ArgenProp AR (Argentina) +Corre manualmente para cubrir el histórico completo de los 12 sub-sitemaps. +Reanudable: checkpoint en /opt/scrapers/argenprop_ar/state.json +""" +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 43200 ' + '/home/jpuma/proyectos/bot-sunat/.venv/bin/python ' + '/opt/scrapers/argenprop_ar/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=43200) + 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[-1000:]) + 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=30), + 'execution_timeout': timedelta(hours=13), +} + +with DAG( + dag_id='argenprop_ar_backfill', + default_args=default_args, + description='Backfill histórico ArgenProp AR – reanudable via checkpoint', + schedule_interval=None, + start_date=datetime(2026, 8, 1), + catchup=False, + max_active_runs=1, + tags=['argentina', 'backfill', 'inmuebles'], +) as dag: + PythonOperator( + task_id='backfill_argenprop_ar', + python_callable=run_backfill, + )