"""
DAG: fincaraiz_co_backfill
Backfill manual del catalogo completo de FincaRaiz.com.co Colombia.
Dos pasadas: venta de apartamentos + arriendo de apartamentos.
Reanudable: checkpoint en /opt/scrapers/fincaraiz_co/backfill_state.json
Para reiniciar desde cero: borrar ese archivo antes de disparar el DAG.
"""
from datetime import datetime, timedelta
import paramiko
from airflow import DAG
from airflow.operators.python import PythonOperator

VPS_HOST     = '152.53.242.31'
VPS_PORT     = 22
VPS_USER     = 'root'
VPS_PASS     = 'XpvvdHnNr0Bq9qU'
VENV_PYTHON  = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
BACKFILL_PY  = '/opt/scrapers/fincaraiz_co/backfill.py'

default_args = {
    'owner':             'jpuma',
    'retries':           0,
    'email':             ['jpuma@redneurocom.com'],
    'email_on_failure':  True,
    'email_on_retry':    False,
    'execution_timeout': timedelta(hours=12),
}


def run_backfill(**ctx):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(VPS_HOST, port=VPS_PORT, username=VPS_USER, password=VPS_PASS, timeout=30)
    ssh.get_transport().set_keepalive(30)

    cmd = f'BF_MAX_PAGES=300 BF_STREAK=8 {VENV_PYTHON} {BACKFILL_PY} 2>&1'
    _, stdout, _ = ssh.exec_command(cmd, timeout=43200)
    output    = stdout.read().decode('utf-8', errors='replace')
    exit_code = stdout.channel.recv_exit_status()
    ssh.close()

    tail = output[-5000:] if len(output) > 5000 else output
    print('[fincaraiz_co_backfill] OUTPUT:\n' + tail)

    if 'Finalizado:' not in output:
        raise RuntimeError(
            f'Backfill no finalizó correctamente (exit={exit_code}). '
            f'Ultimas lineas: {output[-600:]}'
        )
    if exit_code != 0:
        raise RuntimeError(f'Backfill terminó con exit code {exit_code}')


with DAG(
    dag_id            = 'fincaraiz_co_backfill',
    default_args      = default_args,
    description       = 'Backfill catalogo completo FincaRaiz.com.co Colombia (venta+arriendo) — disparo manual, reanudable',
    schedule_interval = None,
    start_date        = datetime(2026, 7, 1),
    catchup           = False,
    max_active_runs   = 1,
    tags              = ['backfill', 'fincaraiz', 'co', 'inmuebles'],
) as dag:

    PythonOperator(
        task_id         = 'run_fincaraiz_backfill',
        python_callable = run_backfill,
    )