diff --git a/dags/encuentra24_ca_daily.py b/dags/encuentra24_ca_daily.py new file mode 100644 index 0000000..0e55add --- /dev/null +++ b/dags/encuentra24_ca_daily.py @@ -0,0 +1,79 @@ +""" +DAG: encuentra24_ca_daily +Portal: Encuentra24.com – Bienes Raices Centroamerica (CR, GT, HN, NI, SV) +Schedule: 08:30 UTC diario +""" + +import paramiko +from datetime import datetime, timedelta +from airflow import DAG +from airflow.operators.python import PythonOperator + +SSH_HOST = '152.53.242.31' +SSH_PORT = 22 +SSH_USER = 'root' +SSH_PASS = 'XpvvdHnNr0Bq9qU' +PYTHON_BIN = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python' +SCRAPER_PATH = '/opt/scrapers/encuentra24_ca/scraper.py' +CMD_TIMEOUT = 21600 # 6 horas + +default_args = { + 'owner': 'jpuma', + 'depends_on_past': False, + 'retries': 2, + 'retry_delay': timedelta(minutes=15), + 'email': ['jpuma@redneurocom.com'], + 'email_on_failure': True, + 'email_on_retry': False, + 'execution_timeout': timedelta(hours=6), +} + + +def run_scraper(**ctx): + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + try: + client.connect(SSH_HOST, port=SSH_PORT, username=SSH_USER, + password=SSH_PASS, timeout=30) + client.get_transport().set_keepalive(30) + + cmd = f'{PYTHON_BIN} {SCRAPER_PATH}' + print(f'Ejecutando en VPS: {cmd}') + + _, stdout, stderr = client.exec_command(cmd, timeout=CMD_TIMEOUT) + output = stdout.read().decode('utf-8', errors='replace') + errors = stderr.read().decode('utf-8', errors='replace') + + if output: + print(f'[STDOUT]\n{output}') + if errors: + print(f'[STDERR]\n{errors}') + + log_text = output + errors + if 'Finalizado:' not in log_text: + raise RuntimeError( + 'Scraper encuentra24_ca no completó correctamente. ' + f'Últimas líneas:\n{log_text[-3000:]}' + ) + + print('Scraper Encuentra24 CA finalizado correctamente.') + finally: + client.close() + + +with DAG( + dag_id='encuentra24_ca_daily', + default_args=default_args, + description='Scraper diario de bienes raices en Encuentra24 CA (CR/GT/HN/NI/SV)', + schedule_interval='30 8 * * *', + start_date=datetime(2024, 1, 1), + catchup=False, + max_active_runs=1, + tags=['scraping', 'encuentra24', 'ca', 'inmuebles'], +) as dag: + + PythonOperator( + task_id='run_encuentra24_ca_scraper', + python_callable=run_scraper, + provide_context=True, + )