from datetime import datetime, timedelta
import paramiko
from airflow import DAG
from airflow.operators.python import PythonOperator

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

VENV_PYTHON  = '/home/jpuma/proyectos/bot-sunat/.venv/bin/python'
SCRAPER_PATH = '/opt/scrapers/trabajando_cl/scraper.py'
SSH_HOST     = '152.53.242.31'
SSH_PORT     = 22
SSH_USER     = 'root'
SSH_PASS     = 'XpvvdHnNr0Bq9qU'


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,
        look_for_keys=False,
        allow_agent=False,
    )
    transport = client.get_transport()
    transport.set_keepalive(30)

    cmd = f'{VENV_PYTHON} {SCRAPER_PATH}'
    _, stdout, stderr = client.exec_command(cmd, timeout=21600)

    output = stdout.read().decode('utf-8', errors='replace')
    errors = stderr.read().decode('utf-8', errors='replace')
    client.close()

    if errors.strip():
        print(f"[STDERR]\n{errors}")
    print(f"[STDOUT]\n{output}")

    if 'Finalizado:' not in output:
        raise RuntimeError(
            f"Scraper trabajando_cl no finalizo correctamente. "
            f"Tail output: {output[-800:]}"
        )


with DAG(
    dag_id='trabajando_cl_daily',
    default_args=DEFAULT_ARGS,
    description='Scraper diario de ofertas de empleo en Trabajando.cl (Chile)',
    schedule_interval='0 6 * * *',
    start_date=datetime(2025, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['scraping', 'trabajando', 'cl', 'empleos'],
) as dag:

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