diff --git a/dags/fincaraiz_co_daily.py b/dags/fincaraiz_co_daily.py new file mode 100644 index 0000000..d508103 --- /dev/null +++ b/dags/fincaraiz_co_daily.py @@ -0,0 +1,87 @@ +from datetime import datetime, timedelta + +import paramiko +from airflow import DAG +from airflow.operators.python import PythonOperator + +# --------------------------------------------------------------------------- +# CONFIG SSH + SCRAPER +# --------------------------------------------------------------------------- +SSH_HOST = '152.53.242.31' +SSH_PORT = 22 +SSH_USER = 'root' +SSH_PASS = 'XpvvdHnNr0Bq9qU' +VENV_PY = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python' +SCRAPER_PY = '/opt/scrapers/fincaraiz_co/scraper.py' + +# --------------------------------------------------------------------------- +# DEFAULT ARGS +# --------------------------------------------------------------------------- +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), +} + + +# --------------------------------------------------------------------------- +# TASK +# --------------------------------------------------------------------------- +def run_scraper(**ctx): + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + client.connect( + SSH_HOST, + port=SSH_PORT, + username=SSH_USER, + password=SSH_PASS, + timeout=30, + ) + client.get_transport().set_keepalive(30) + + # stderr redirigido a stdout para capturar todo en un solo buffer + cmd = f'{VENV_PY} {SCRAPER_PY} 2>&1' + _, stdout, _ = client.exec_command(cmd, timeout=21600) # 6 h max + log_output = stdout.read().decode('utf-8', errors='replace') + exit_code = stdout.channel.recv_exit_status() + client.close() + + # Mostrar las últimas 5 000 chars del log en el task log de Airflow + tail = log_output[-5000:] if len(log_output) > 5000 else log_output + print('[fincaraiz_co] STDOUT+STDERR:\n' + tail) + + if 'Finalizado:' not in log_output: + raise RuntimeError( + f'fincaraiz_co_scraper no imprimió "Finalizado:" ' + f'(exit_code={exit_code}). ' + f'Últimas líneas: {log_output[-400:]}' + ) + + if exit_code != 0: + raise RuntimeError( + f'fincaraiz_co_scraper terminó con exit_code={exit_code}' + ) + + +# --------------------------------------------------------------------------- +# DAG +# --------------------------------------------------------------------------- +with DAG( + dag_id='fincaraiz_co_daily', + default_args=default_args, + description='Scraper diario FincaRaiz.com.co – venta y arriendo de apartamentos (Colombia)', + schedule_interval='0 7 * * *', + start_date=datetime(2025, 1, 1), + catchup=False, + max_active_runs=1, + tags=['scraping', 'fincaraiz', 'co', 'inmuebles'], +) as dag: + + scrape = PythonOperator( + task_id='run_fincaraiz_scraper', + python_callable=run_scraper, + )