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'
VENV_PY    = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
SCRAPER_PY = '/opt/scrapers/fincaraiz_co/scraper.py'

# ---------------------------------------------------------------------------
# DEFAULT ARGS
# ---------------------------------------------------------------------------
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),
}


# ---------------------------------------------------------------------------
# TASK
# ---------------------------------------------------------------------------
def run_scraper(**ctx):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(
        SSH_HOST,
        port=SSH_PORT,
        username=SSH_USER,
        password=SSH_PASS,
        timeout=30,
    )
    client.get_transport().set_keepalive(30)

    # stderr redirigido a stdout para capturar todo en un solo buffer
    cmd = f'{VENV_PY} {SCRAPER_PY} 2>&1'
    _, stdout, _ = client.exec_command(cmd, timeout=21600)   # 6 h max
    log_output = stdout.read().decode('utf-8', errors='replace')
    exit_code  = stdout.channel.recv_exit_status()
    client.close()

    # Mostrar las últimas 5 000 chars del log en el task log de Airflow
    tail = log_output[-5000:] if len(log_output) > 5000 else log_output
    print('[fincaraiz_co] STDOUT+STDERR:\n' + tail)

    if 'Finalizado:' not in log_output:
        raise RuntimeError(
            f'fincaraiz_co_scraper no imprimió "Finalizado:" '
            f'(exit_code={exit_code}). '
            f'Últimas líneas: {log_output[-400:]}'
        )

    if exit_code != 0:
        raise RuntimeError(
            f'fincaraiz_co_scraper terminó con exit_code={exit_code}'
        )


# ---------------------------------------------------------------------------
# DAG
# ---------------------------------------------------------------------------
with DAG(
    dag_id='fincaraiz_co_daily',
    default_args=default_args,
    description='Scraper diario FincaRaiz.com.co – venta y arriendo de apartamentos (Colombia)',
    schedule_interval='0 7 * * *',
    start_date=datetime(2025, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['scraping', 'fincaraiz', 'co', 'inmuebles'],
) as dag:

    scrape = PythonOperator(
        task_id='run_fincaraiz_scraper',
        python_callable=run_scraper,
    )