77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
"""
|
|
DAG: corotos_do_daily
|
|
Scraper diario de inmuebles de Corotos.com.do (Republica Dominicana).
|
|
Ejecuta el scraper via SSH sobre el ARM VPS host (152.53.242.31).
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
|
|
import paramiko
|
|
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/corotos_do/scraper.py'
|
|
|
|
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),
|
|
}
|
|
|
|
|
|
def run_scraper(**ctx):
|
|
client = paramiko.SSHClient()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
client.connect(
|
|
hostname=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}'
|
|
_, stdout, stderr = client.exec_command(cmd, timeout=21600)
|
|
|
|
log_out = stdout.read().decode('utf-8', errors='replace')
|
|
log_err = stderr.read().decode('utf-8', errors='replace')
|
|
client.close()
|
|
|
|
print('=== STDOUT ===')
|
|
print(log_out)
|
|
if log_err.strip():
|
|
print('=== STDERR ===')
|
|
print(log_err)
|
|
|
|
full_log = log_out + log_err
|
|
if 'Finalizado:' not in full_log:
|
|
raise RuntimeError(
|
|
"Scraper corotos_do no imprimio 'Finalizado:'. "
|
|
f"Ultimas 500 chars: {full_log[-500:]!r}"
|
|
)
|
|
|
|
|
|
with DAG(
|
|
dag_id='corotos_do_daily',
|
|
default_args=default_args,
|
|
description='Scraper diario de inmuebles Corotos.com.do (Republica Dominicana)',
|
|
schedule_interval='30 8 * * *',
|
|
start_date=datetime(2025, 1, 1),
|
|
catchup=False,
|
|
max_active_runs=1,
|
|
tags=['scraping', 'corotos', 'do', 'inmuebles'],
|
|
) as dag:
|
|
|
|
t_scrape = PythonOperator(
|
|
task_id='run_corotos_do_scraper',
|
|
python_callable=run_scraper,
|
|
) |