"""
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,
    )