98 lines
3.0 KiB
Python
98 lines
3.0 KiB
Python
"""
|
||
DAG: yapo_cl_daily
|
||
Portal: Yapo.cl (Chile – Schibsted clasificados)
|
||
Schedule: 08:00 UTC diario
|
||
"""
|
||
|
||
import paramiko
|
||
from datetime import datetime, timedelta
|
||
from airflow import DAG
|
||
from airflow.operators.python import PythonOperator
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DEFAULT ARGS
|
||
# ---------------------------------------------------------------------------
|
||
default_args = {
|
||
'owner': 'airflow',
|
||
'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),
|
||
}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CONSTANTES SSH
|
||
# ---------------------------------------------------------------------------
|
||
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/yapo_cl/scraper.py'
|
||
CMD_TIMEOUT = 21600 # 6 horas en segundos
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CALLABLE
|
||
# ---------------------------------------------------------------------------
|
||
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 = '{} {}'.format(PYTHON_BIN, SCRAPER_PATH)
|
||
print('Ejecutando en VPS: {}'.format(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('[STDOUT]\n{}'.format(output))
|
||
if errors:
|
||
print('[STDERR]\n{}'.format(errors))
|
||
|
||
log_text = output + errors
|
||
if 'Finalizado:' not in log_text:
|
||
raise RuntimeError(
|
||
'Scraper yapo_cl no completó correctamente. '
|
||
'Últimas líneas del log:\n{}'.format(log_text[-3000:])
|
||
)
|
||
|
||
print('Scraper Yapo.cl completado exitosamente.')
|
||
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DAG
|
||
# ---------------------------------------------------------------------------
|
||
with DAG(
|
||
dag_id='yapo_cl_daily',
|
||
default_args=default_args,
|
||
description='Scraper diario de bienes raíces en Yapo.cl (Chile)',
|
||
schedule_interval='0 8 * * *',
|
||
start_date=datetime(2024, 1, 1),
|
||
catchup=False,
|
||
max_active_runs=1,
|
||
tags=['scraping', 'yapo', 'cl', 'inmuebles'],
|
||
) as dag:
|
||
|
||
scrape_task = PythonOperator(
|
||
task_id='run_yapo_scraper',
|
||
python_callable=run_scraper,
|
||
provide_context=True,
|
||
)
|