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'
PYTHON_BIN   = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
SCRAPER_PATH = '/opt/scrapers/portal_inmobiliario_cl/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())
    try:
        client.connect(
            hostname=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)
        _, stdout, stderr = client.exec_command(cmd, timeout=21600)  # 6 h max

        out = stdout.read().decode('utf-8', errors='replace')
        err = stderr.read().decode('utf-8', errors='replace')

        print("=== STDOUT ===\n{}".format(out))
        if err.strip():
            print("=== STDERR ===\n{}".format(err))

        if 'Finalizado:' not in out:
            raise RuntimeError(
                "El scraper no imprimio 'Finalizado:'. "
                "STDOUT: {!r} | STDERR: {!r}".format(
                    out[-2000:], err[-1000:]
                )
            )
    finally:
        client.close()


with DAG(
    dag_id='portal_inmobiliario_cl_daily',
    default_args=default_args,
    description='Scraper diario de PortalInmobiliario.com (Chile - MercadoLibre)',
    schedule_interval='30 7 * * *',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['scraping', 'portal-inmobiliario', 'cl', 'inmuebles'],
) as dag:

    scrape_task = PythonOperator(
        task_id='run_portal_inmobiliario_scraper',
        python_callable=run_scraper,
    )