#!/usr/bin/env python3
"""
ETL automático de SINADEF (Datos Abiertos - MINSA Perú).

Descarga el ZIP publicado por MINSA, extrae el CSV y recarga tbl_SINADEF en DuckDB.
Reemplaza el flujo manual (descarga a mano + sinadef_duckdb_loader.py).

Uso:
    python3 sinadef_etl.py                 # solo recarga si MINSA publicó algo nuevo
    python3 sinadef_etl.py --force         # descarga y recarga siempre
    python3 sinadef_etl.py --skip-download # recarga con el CSV que ya está en downloads/
    python3 sinadef_etl.py --check         # solo consulta si hay versión nueva (exit 0=hay, 1=no)
"""

import argparse
import hashlib
import json
import logging
import multiprocessing
import os
import shutil
import sys
import zipfile
from datetime import datetime
from pathlib import Path

import duckdb
import requests

# ─────────────────────────── Configuración ───────────────────────────

BASE_DIR = Path(__file__).resolve().parent
DOWNLOAD_DIR = BASE_DIR / "downloads"
DB_PATH = BASE_DIR / "sinadef.duckdb"
STATE_FILE = BASE_DIR / ".sinadef_state.json"
LOG_FILE = BASE_DIR / "sinadef_etl.log"

# Recurso publicado por MINSA en su Nextcloud institucional.
# Enlazado desde https://www.datosabiertos.gob.pe/dataset/informaci%C3%B3n-de-fallecidos-del-
#   sistema-inform%C3%A1tico-nacional-de-defunciones-sinadef-ministerio
# Se actualiza a diario. Devuelve SINADEF_DATOS_ABIERTOS_2.zip
SINADEF_URL = "https://files.minsa.gob.pe/s/RjeWiJt2wX3pMdG/download"

# El WAF de MINSA responde 403 sin un User-Agent de navegador.
HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
    )
}

ZIP_PATH = DOWNLOAD_DIR / "SINADEF_DATOS_ABIERTOS.zip"
CSV_PATH = DOWNLOAD_DIR / "SINADEF_DATOS_ABIERTOS.csv"

TABLE = "tbl_SINADEF"

# Esquema publicado desde 2025-2026 (36 columnas). Las 4 primeras después de N
# (DEPARTAMENTO/PROVINCIA/DISTRITO_FALLECIMIENTO, TIPO_CDEF) no existían en el
# esquema anterior de 32 columnas.
COLUMNS = [
    "N", "DEPARTAMENTO_FALLECIMIENTO", "PROVINCIA_FALLECIMIENTO",
    "DISTRITO_FALLECIMIENTO", "TIPO_CDEF", "TIPO_SEGURO", "SEXO", "TIEMPO_EDAD",
    "EDAD", "ESTADO_CIVIL", "NIVEL_DE_INSTRUCCION", "ETNIA",
    "COD_UBIGEO_DOMICILIO", "PAIS_DOMICILIO", "DEPARTAMENTO_DOMICILIO",
    "PROVINCIA_DOMICILIO", "DISTRITO_DOMICILIO", "FECHA", "ANIO", "MES",
    "TIPO_LUGAR", "INSTITUCION", "MUERTE_VIOLENTA", "NECROPSIA",
    "DEBIDO_CAUSA_A", "CAUSA_A_CIEX", "DEBIDO_CAUSA_B", "CAUSA_B_CIEX",
    "DEBIDO_CAUSA_C", "CAUSA_C_CIEX", "DEBIDO_CAUSA_D", "CAUSA_D_CIEX",
    "DEBIDO_CAUSA_E", "CAUSA_E_CIEX", "DEBIDO_CAUSA_F", "CAUSA_F_CIEX",
]

# Piso de cordura: si el CSV trae menos filas que esto, algo salió mal en la
# descarga y no vale la pena reemplazar la tabla vigente.
MIN_ROWS_ESPERADAS = 1_000_000

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler(LOG_FILE), logging.StreamHandler(sys.stdout)],
)
log = logging.getLogger("sinadef")


# ─────────────────────────── Estado ───────────────────────────

def load_state():
    if STATE_FILE.exists():
        try:
            return json.loads(STATE_FILE.read_text())
        except (json.JSONDecodeError, OSError):
            log.warning("Estado ilegible, se ignora")
    return {}


def save_state(**kwargs):
    state = load_state()
    state.update(kwargs)
    STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False))


# ─────────────────────────── Descarga ───────────────────────────

def remote_metadata():
    """
    Cabeceras del ZIP publicado, solo informativas.

    OJO: el Last-Modified de MINSA NO sirve para detectar cambios. El Nextcloud
    arma el ZIP al vuelo y devuelve la hora de generación, así que dos consultas
    seguidas dan fechas distintas; y cuando sirve una copia cacheada devuelve la
    fecha de esa copia aunque el contenido ya haya cambiado. Comprobado el
    2026-08-06: tres consultas en 30 segundos dieron 14:24:11, 14:24:24 y
    14:24:38 GMT. Los cambios se detectan por el hash del CSV, no por aquí.
    """
    r = requests.head(SINADEF_URL, headers=HEADERS, allow_redirects=True, timeout=60)
    r.raise_for_status()
    return {
        "last_modified": r.headers.get("Last-Modified"),
        "content_length": r.headers.get("Content-Length"),
    }


def hash_csv(path: Path) -> str:
    """SHA-256 del CSV extraído: la única señal fiable de si la data cambió.

    Se hashea el CSV y no el ZIP porque el ZIP se regenera en cada descarga y
    sus timestamps internos cambian aunque el contenido sea idéntico.
    """
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for bloque in iter(lambda: f.read(8 * 1024 * 1024), b""):
            h.update(bloque)
    return h.hexdigest()


def descargar_zip():
    DOWNLOAD_DIR.mkdir(exist_ok=True)
    tmp = ZIP_PATH.with_suffix(".zip.part")

    log.info(f"Descargando {SINADEF_URL}")
    with requests.get(SINADEF_URL, headers=HEADERS, stream=True, timeout=(60, 600)) as r:
        r.raise_for_status()
        total = int(r.headers.get("Content-Length", 0))
        bajado = 0
        siguiente_aviso = 50 * 1024 * 1024
        with open(tmp, "wb") as f:
            for chunk in r.iter_content(chunk_size=1024 * 1024):
                f.write(chunk)
                bajado += len(chunk)
                if bajado >= siguiente_aviso:
                    pct = f" ({bajado / total:.0%})" if total else ""
                    log.info(f"  {bajado / 1024**2:,.0f} MB{pct}")
                    siguiente_aviso += 50 * 1024 * 1024

    if total and bajado != total:
        tmp.unlink(missing_ok=True)
        raise IOError(f"Descarga incompleta: {bajado} de {total} bytes")

    tmp.replace(ZIP_PATH)
    log.info(f"✓ ZIP descargado: {ZIP_PATH.stat().st_size / 1024**2:,.0f} MB")
    return ZIP_PATH


def extraer_csv(zip_path):
    """Extrae el único CSV del ZIP a downloads/SINADEF_DATOS_ABIERTOS.csv."""
    with zipfile.ZipFile(zip_path) as z:
        miembros = [n for n in z.namelist() if n.lower().endswith(".csv")]
        if not miembros:
            raise ValueError(f"El ZIP no contiene ningún CSV: {z.namelist()}")
        if len(miembros) > 1:
            log.warning(f"El ZIP trae {len(miembros)} CSV; se usa {miembros[0]}")
        miembro = miembros[0]

        log.info(f"Extrayendo {miembro}")
        # El CSV va dentro de una carpeta; se aplana a un nombre fijo.
        with z.open(miembro) as src, open(CSV_PATH, "wb") as dst:
            shutil.copyfileobj(src, dst, length=8 * 1024 * 1024)

    log.info(f"✓ CSV extraído: {CSV_PATH} ({CSV_PATH.stat().st_size / 1024**2:,.0f} MB)")
    return CSV_PATH


# ─────────────────────────── Carga a DuckDB ───────────────────────────

def verificar_esquema(conn, csv_path):
    """Avisa si MINSA cambió las columnas respecto a COLUMNS."""
    reales = [
        r[0] for r in conn.execute(
            f"DESCRIBE SELECT * FROM read_csv('{csv_path}', all_varchar=true, "
            f"header=true, sample_size=1000)"
        ).fetchall()
    ]
    if reales != COLUMNS:
        faltan = [c for c in COLUMNS if c not in reales]
        nuevas = [c for c in reales if c not in COLUMNS]
        log.warning(
            f"⚠ El esquema del CSV cambió respecto al esperado. "
            f"Columnas nuevas: {nuevas or 'ninguna'}. Desaparecidas: {faltan or 'ninguna'}. "
            f"Se cargará el esquema real ({len(reales)} columnas); revisar main.py y el dashboard."
        )
    return reales


def cargar_duckdb(csv_path):
    """
    Construye la BD en un archivo aparte y la pone en su sitio con os.replace().

    El swap es atómico, así que la API puede seguir sirviendo durante toda la
    carga: las conexiones abiertas conservan el inode anterior hasta cerrarse y
    las nuevas toman la BD recién construida. Sin esto, DuckDB (un solo escritor)
    haría fallar el cron cada vez que uvicorn tuviera el archivo abierto.
    """
    build_path = DB_PATH.with_suffix(".building.duckdb")
    build_path.unlink(missing_ok=True)
    Path(str(build_path) + ".wal").unlink(missing_ok=True)

    conn = duckdb.connect(str(build_path))
    try:
        threads = max(1, int(multiprocessing.cpu_count() * 0.9))
        conn.execute(f"SET threads={threads}")
        conn.execute("SET preserve_insertion_order=FALSE")

        verificar_esquema(conn, csv_path)

        log.info("Cargando CSV en DuckDB...")
        conn.execute(f"""
            CREATE TABLE {TABLE} AS
            SELECT * FROM read_csv(
                '{csv_path}',
                all_varchar = true,
                header = true,
                delim = ',',
                quote = '"',
                escape = '"'
            )
        """)

        filas = conn.execute(f"SELECT COUNT(*) FROM {TABLE}").fetchone()[0]
        if filas < MIN_ROWS_ESPERADAS:
            raise ValueError(
                f"Solo {filas:,} filas cargadas (mínimo esperado {MIN_ROWS_ESPERADAS:,}). "
                f"Se aborta y se conserva la base anterior."
            )

        fmin, fmax = conn.execute(
            f"SELECT MIN(FECHA), MAX(FECHA) FROM {TABLE}"
        ).fetchone()
        conn.execute("CHECKPOINT")
    except Exception:
        conn.close()
        build_path.unlink(missing_ok=True)
        raise
    finally:
        try:
            conn.close()
        except Exception:
            pass

    previas = filas_actuales()
    os.replace(build_path, DB_PATH)  # atómico: mismo filesystem

    log.info(
        "\n"
        "  ═══════════════ RESUMEN DE CARGA ═══════════════\n"
        f"  Registros anteriores : {previas:,}\n"
        f"  Registros cargados   : {filas:,}  ({filas - previas:+,})\n"
        f"  Rango de fechas      : {fmin} → {fmax}\n"
        "  ════════════════════════════════════════════════"
    )
    return {"filas": filas, "fecha_min": fmin, "fecha_max": fmax}


def filas_actuales():
    """Filas de la BD vigente, para reportar el delta. 0 si aún no existe."""
    if not DB_PATH.exists():
        return 0
    try:
        c = duckdb.connect(str(DB_PATH), read_only=True)
        try:
            return c.execute(f"SELECT COUNT(*) FROM {TABLE}").fetchone()[0]
        finally:
            c.close()
    except Exception:
        return 0


# ─────────────────────────── Main ───────────────────────────

def main():
    p = argparse.ArgumentParser(description="ETL automático de SINADEF")
    p.add_argument("--force", action="store_true",
                   help="descargar y recargar aunque no haya versión nueva")
    p.add_argument("--skip-download", action="store_true",
                   help="usar el CSV ya presente en downloads/")
    p.add_argument("--check", action="store_true",
                   help="solo verificar si hay versión nueva (exit 0=sí, 1=no)")
    p.add_argument("--keep-zip", action="store_true",
                   help="conservar el ZIP descargado")
    args = p.parse_args()

    state = load_state()
    meta = {}

    if args.skip_download:
        if not CSV_PATH.exists():
            log.error(f"✗ No existe {CSV_PATH}; no se puede usar --skip-download")
            return 1
    else:
        meta = remote_metadata()
        log.info(f"Publicación remota (informativo): {meta['last_modified']}")

        # Hay que descargar para saber si cambió: MINSA no expone ninguna señal
        # fiable en las cabeceras. Son ~610 MB, unos 25 segundos.
        zip_path = descargar_zip()
        extraer_csv(zip_path)
        if not args.keep_zip:
            zip_path.unlink(missing_ok=True)

    log.info("Calculando hash del CSV...")
    huella = hash_csv(CSV_PATH)
    huella_previa = state.get("csv_sha256")
    cambio = huella != huella_previa
    log.info(f"SHA-256: {huella[:16]}… (anterior: {(huella_previa or 'ninguno')[:16]})")

    if args.check:
        log.info("Hay data nueva" if cambio else "El CSV es idéntico al ya cargado")
        return 0 if cambio else 1

    if not cambio and not args.force:
        log.info(
            f"El CSV publicado es idéntico al que ya está cargado "
            f"({state.get('cargado_en', 'desconocida')}). No se recarga la base."
        )
        # Se registra la comprobación para dejar constancia de que el cron corrió.
        save_state(ultima_verificacion=datetime.now().isoformat(timespec="seconds"))
        return 0

    resultado = cargar_duckdb(CSV_PATH)

    save_state(
        last_modified=meta.get("last_modified", state.get("last_modified")),
        csv_sha256=huella,
        cargado_en=datetime.now().isoformat(timespec="seconds"),
        ultima_verificacion=datetime.now().isoformat(timespec="seconds"),
        filas=resultado["filas"],
        fecha_max=resultado["fecha_max"],
    )
    log.info("✅ ETL completado")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as e:
        log.error(f"✗ ETL fallido: {e}", exc_info=True)
        sys.exit(2)