"""
API SINADEF — endpoints agregados sobre DuckDB.

Principio de diseño: NO se crean atributos derivados. Todo lo que devuelve esta
API es un conteo sobre columnas tal como las publica MINSA. No existe ninguna
categoría inventada (feminicidio, "es violenta", grupos etarios con nombre, etc.).
Quien consuma la API aplica los filtros que considere y saca sus conclusiones.

Levantar:
    uvicorn main:app --host 0.0.0.0 --port 8009
"""

import json
import os
import threading
from pathlib import Path
from typing import Optional

import duckdb
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

BASE_DIR = Path(__file__).resolve().parent
DB_PATH = BASE_DIR / "sinadef.duckdb"
FRONTEND_DIST = BASE_DIR / "frontend" / "dist"

app = FastAPI(
    title="SINADEF API",
    description="Agregados de defunciones (MINSA Perú) sobre las columnas originales de SINADEF.",
    version="2.0",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET"],
    allow_headers=["*"],
)

# ── Conexión ────────────────────────────────────────────────────────────
# El ETL diario reemplaza el archivo con os.replace() (swap atómico). Se
# cachea la conexión y se reabre solo cuando cambia el inode, de modo que la
# recarga de las 6am no requiere reiniciar la API ni provoca downtime: las
# consultas en vuelo siguen leyendo el inode anterior hasta cerrarse.

_conn = None
_conn_inode = None
_lock = threading.Lock()


def get_conn():
    """Conexión base, compartida. Nunca se consulta directamente: ver q()."""
    global _conn, _conn_inode
    if not DB_PATH.exists():
        raise HTTPException(503, "Base de datos no disponible")
    inode = os.stat(DB_PATH).st_ino
    with _lock:
        if _conn is None or inode != _conn_inode:
            # La anterior se cierra ANTES de abrir la nueva, y el orden importa:
            # DuckDB cachea una instancia de base por ruta dentro del proceso.
            # Mientras quede una conexión viva sobre sinadef.duckdb, un connect()
            # posterior reutiliza esa instancia y sigue leyendo el archivo que el
            # ETL ya reemplazó — la API serviría datos de ayer indefinidamente.
            if _conn is not None:
                try:
                    _conn.close()
                except Exception:
                    pass
                _conn = None
            _conn = duckdb.connect(str(DB_PATH), read_only=True)
            _conn_inode = inode
        return _conn


def q(sql: str, params: list | None = None):
    """
    Ejecuta sobre un cursor propio y devuelve lista de dicts.

    El cursor es obligatorio, no una optimización: FastAPI corre los endpoints
    síncronos en un threadpool, y la app pide varios endpoints en paralelo al
    cargar. Compartir una sola DuckDBPyConnection entre esos hilos la bloquea de
    forma permanente — el proceso queda inservible, no solo lento. cursor() da a
    cada hilo una conexión hija independiente sobre la misma base.

    El reintento cubre la ventana de milisegundos en que el ETL reemplaza la
    base: al reabrirse la conexión, un cursor en vuelo queda invalidado.
    """
    for intento in (1, 2):
        cur = get_conn().cursor()
        try:
            cur.execute(sql, params or [])
            cols = [d[0] for d in cur.description]
            return [dict(zip(cols, row)) for row in cur.fetchall()]
        except duckdb.Error:
            if intento == 2:
                raise
            with _lock:
                globals()["_conn_inode"] = None  # fuerza reapertura
        finally:
            try:
                cur.close()
            except Exception:
                pass


# ── Filtros ─────────────────────────────────────────────────────────────
# Los valores llegan como texto libre pero siempre viajan como parámetros
# enlazados; los nombres de columna salen de una whitelist.

DIMENSIONES = {
    "tipo": "MUERTE_VIOLENTA",
    "sexo": "SEXO",
    "departamento": "DEPARTAMENTO_FALLECIMIENTO",
    "provincia": "PROVINCIA_FALLECIMIENTO",
    "distrito": "DISTRITO_FALLECIMIENTO",
    "lugar": "TIPO_LUGAR",
    "institucion": "INSTITUCION",
    "necropsia": "NECROPSIA",
    "seguro": "TIPO_SEGURO",
    "estado_civil": "ESTADO_CIVIL",
    "instruccion": "NIVEL_DE_INSTRUCCION",
    "etnia": "ETNIA",
    "tipo_cdef": "TIPO_CDEF",
}


def build_where(
    tipo: Optional[str],
    sexo: Optional[str],
    departamento: Optional[str],
    lugar: Optional[str],
    desde: Optional[str],
    hasta: Optional[str],
):
    clauses, params = ["FECHA IS NOT NULL"], []
    for valor, columna in (
        (tipo, "MUERTE_VIOLENTA"),
        (sexo, "SEXO"),
        (departamento, "DEPARTAMENTO_FALLECIMIENTO"),
        (lugar, "TIPO_LUGAR"),
    ):
        if valor:
            opciones = [v.strip() for v in valor.split(",") if v.strip()]
            if opciones:
                marcadores = ",".join("?" * len(opciones))
                clauses.append(f"{columna} IN ({marcadores})")
                params.extend(opciones)
    if desde:
        clauses.append("ANIO >= ?")
        params.append(str(desde))
    if hasta:
        clauses.append("ANIO <= ?")
        params.append(str(hasta))
    return " AND ".join(clauses), params


def filtros(
    tipo: Optional[str] = Query(None, description="MUERTE_VIOLENTA, separado por comas"),
    sexo: Optional[str] = Query(None, description="SEXO, separado por comas"),
    departamento: Optional[str] = Query(None, description="DEPARTAMENTO_FALLECIMIENTO"),
    lugar: Optional[str] = Query(None, description="TIPO_LUGAR"),
    desde: Optional[str] = Query(None, description="Año inicial, ej. 2023"),
    hasta: Optional[str] = Query(None, description="Año final, ej. 2026"),
):
    return build_where(tipo, sexo, departamento, lugar, desde, hasta)


# ── Endpoints ───────────────────────────────────────────────────────────


@app.get("/api/meta")
def meta():
    """Valores disponibles para armar los filtros, tal como están en la data."""
    base = q("SELECT MIN(ANIO) anio_min, MAX(ANIO) anio_max, COUNT(*) filas, MAX(FECHA) fecha_max FROM tbl_SINADEF")[0]

    # Último mes con data: se marca como parcial para no dibujar caídas falsas.
    ultimo_mes = q("SELECT max(strftime(TRY_CAST(FECHA AS DATE), '%Y-%m')) m FROM tbl_SINADEF")[0]["m"]

    def valores(col):
        return [r["v"] for r in q(
            f"SELECT {col} AS v, COUNT(*) c FROM tbl_SINADEF WHERE {col} IS NOT NULL GROUP BY 1 ORDER BY c DESC"
        )]

    # Cuándo se recargó la base y cuándo se comprobó por última vez si MINSA
    # publicó algo. Son cosas distintas: el ETL corre a diario, pero solo
    # recarga si el contenido cambió, así que sin el segundo dato no hay forma
    # de distinguir "no hay novedades" de "el cron dejó de correr".
    verificado = None
    try:
        estado_etl = json.loads((BASE_DIR / ".sinadef_state.json").read_text())
        verificado = estado_etl.get("ultima_verificacion") or estado_etl.get("cargado_en")
    except Exception:
        pass

    return {
        **base,
        "ultimo_mes": ultimo_mes,
        "anios": [r["a"] for r in q("SELECT DISTINCT ANIO a FROM tbl_SINADEF ORDER BY a")],
        "tipos_muerte": valores("MUERTE_VIOLENTA"),
        "sexos": valores("SEXO"),
        "lugares": valores("TIPO_LUGAR"),
        "departamentos": valores("DEPARTAMENTO_FALLECIMIENTO"),
        "actualizado": os.path.getmtime(DB_PATH),
        "verificado": verificado,
    }


@app.get("/api/resumen")
def resumen(
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    desde: Optional[str] = None,
    hasta: Optional[str] = None,
):
    """Totales del periodo filtrado y comparación con el periodo anterior."""
    where, params = build_where(tipo, sexo, departamento, lugar, desde, hasta)

    total = q(f"SELECT COUNT(*) n FROM tbl_SINADEF WHERE {where}", params)[0]["n"]

    por_anio = q(
        f"SELECT ANIO, COUNT(*) n FROM tbl_SINADEF WHERE {where} GROUP BY 1 ORDER BY 1",
        params,
    )

    # Variación del último año cerrado frente al anterior. El año en curso se
    # excluye del cálculo porque está incompleto.
    anio_max_data = q("SELECT MAX(ANIO) a FROM tbl_SINADEF")[0]["a"]
    cerrados = [r for r in por_anio if r["ANIO"] != anio_max_data]
    variacion = None
    if len(cerrados) >= 2:
        ult, prev = cerrados[-1]["n"], cerrados[-2]["n"]
        if prev:
            variacion = {
                "anio": cerrados[-1]["ANIO"],
                "anio_previo": cerrados[-2]["ANIO"],
                "valor": ult,
                "valor_previo": prev,
                "pct": round(100.0 * (ult - prev) / prev, 1),
            }

    return {"total": total, "por_anio": por_anio, "variacion": variacion, "anio_en_curso": anio_max_data}


@app.get("/api/serie-mensual")
def serie_mensual(
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    desde: Optional[str] = None,
    hasta: Optional[str] = None,
    serie_por: Optional[str] = Query(None, description="Dimensión para abrir en múltiples series"),
):
    """Conteo mensual. Con serie_por devuelve una serie por cada valor de esa columna."""
    where, params = build_where(tipo, sexo, departamento, lugar, desde, hasta)

    if serie_por:
        if serie_por not in DIMENSIONES:
            raise HTTPException(400, f"serie_por inválido. Opciones: {list(DIMENSIONES)}")
        col = DIMENSIONES[serie_por]
        filas = q(
            f"""SELECT ANIO || '-' || MES AS mes, {col} AS serie, COUNT(*) n
                FROM tbl_SINADEF WHERE {where}
                GROUP BY 1, 2 ORDER BY 1, 2""",
            params,
        )
    else:
        filas = q(
            f"""SELECT ANIO || '-' || MES AS mes, COUNT(*) n
                FROM tbl_SINADEF WHERE {where}
                GROUP BY 1 ORDER BY 1""",
            params,
        )

    ultimo_mes = q("SELECT max(ANIO || '-' || MES) m FROM tbl_SINADEF")[0]["m"]
    return {"datos": filas, "mes_parcial": ultimo_mes}


@app.get("/api/dimension/{dim}")
def por_dimension(
    dim: str,
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    desde: Optional[str] = None,
    hasta: Optional[str] = None,
    limite: int = Query(30, ge=1, le=200),
):
    """Conteo agrupado por cualquier columna de la whitelist."""
    if dim not in DIMENSIONES:
        raise HTTPException(400, f"Dimensión inválida. Opciones: {list(DIMENSIONES)}")
    col = DIMENSIONES[dim]
    where, params = build_where(tipo, sexo, departamento, lugar, desde, hasta)
    return q(
        f"""SELECT {col} AS categoria, COUNT(*) n
            FROM tbl_SINADEF WHERE {where}
            GROUP BY 1 ORDER BY n DESC LIMIT {limite}""",
        params,
    )


@app.get("/api/estacionalidad")
def estacionalidad(
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    desde: Optional[str] = None,
    hasta: Optional[str] = None,
):
    """Matriz año × mes, para ver el patrón intranual."""
    where, params = build_where(tipo, sexo, departamento, lugar, desde, hasta)
    return q(
        f"""SELECT ANIO, MES, COUNT(*) n
            FROM tbl_SINADEF WHERE {where}
            GROUP BY 1, 2 ORDER BY 1, 2""",
        params,
    )


@app.get("/api/interanual")
def interanual(
    mes_corte: str = Query(..., description="Mes de corte, '01'–'12'"),
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
):
    """
    Compara un mismo mes y un mismo acumulado enero–mes entre todos los años.

    Deliberadamente ignora el filtro de años de la pantalla: la comparación sirve
    justamente para mirar el histórico completo. Devuelve ambas series en una
    sola consulta.
    """
    if mes_corte not in {f"{m:02d}" for m in range(1, 13)}:
        raise HTTPException(400, "mes_corte debe ser '01'–'12'")

    where, params = build_where(tipo, sexo, departamento, lugar, None, None)

    filas = q(
        f"""SELECT ANIO,
                   COUNT(*) FILTER (WHERE MES = ?) AS mes_n,
                   COUNT(*) FILTER (WHERE MES <= ?) AS ytd_n
            FROM tbl_SINADEF WHERE {where}
            GROUP BY 1 ORDER BY 1""",
        [mes_corte, mes_corte] + params,
    )

    # Un año solo entra en la comparación si ya transcurrió el mes de corte;
    # de lo contrario su acumulado no es comparable con el de los demás.
    ultimo = q("SELECT MAX(ANIO) a, MAX(MES) FILTER (WHERE ANIO = (SELECT MAX(ANIO) FROM tbl_SINADEF)) m FROM tbl_SINADEF")[0]
    filas = [f for f in filas if f["ANIO"] < ultimo["a"] or mes_corte <= ultimo["m"]]

    def var(campo):
        if len(filas) < 2:
            return None
        ult, prev = filas[-1], filas[-2]
        if not prev[campo]:
            return None
        previos = [f[campo] for f in filas[:-1]]
        return {
            "anio": ult["ANIO"],
            "valor": ult[campo],
            "anio_previo": prev["ANIO"],
            "valor_previo": prev[campo],
            "pct": round(100.0 * (ult[campo] - prev[campo]) / prev[campo], 1),
            "promedio_historico": round(sum(previos) / len(previos)),
            "pct_vs_promedio": round(
                100.0 * (ult[campo] - sum(previos) / len(previos)) / (sum(previos) / len(previos)), 1
            ),
        }

    return {
        "mes_corte": mes_corte,
        "datos": filas,
        "variacion_mes": var("mes_n"),
        "variacion_ytd": var("ytd_n"),
        "anio_en_curso": ultimo["a"],
    }


@app.get("/api/consolidacion")
def consolidacion():
    """
    Cuán completo está cada uno de los últimos meses.

    SINADEF sigue recibiendo registros con retraso, así que un mes recién
    cerrado está subvalorado. Se reporta la fecha de corte de la data y qué
    meses aún no deberían leerse como definitivos, sin ajustar ningún conteo.
    """
    fecha_max = q("SELECT MAX(FECHA) f FROM tbl_SINADEF")[0]["f"]
    meses = q(
        """SELECT ANIO || '-' || MES AS mes, COUNT(*) n
           FROM tbl_SINADEF GROUP BY 1 ORDER BY 1 DESC LIMIT 6"""
    )
    return {"fecha_corte": fecha_max, "ultimos_meses": meses}


@app.get("/api/edades")
def edades(
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    desde: Optional[str] = None,
    hasta: Optional[str] = None,
):
    """
    Distribución de EDAD en décadas. Solo TIEMPO_EDAD='AÑOS'; los registros en
    meses/días/horas se devuelven aparte sin reclasificarlos.
    """
    where, params = build_where(tipo, sexo, departamento, lugar, desde, hasta)
    bins = q(
        # // es división entera: con / DuckDB devuelve flotante y cada edad
        # quedaría en su propio "bin".
        f"""SELECT (TRY_CAST(EDAD AS INT) // 10) * 10 AS decada, COUNT(*) n
            FROM tbl_SINADEF
            WHERE {where} AND TIEMPO_EDAD = 'AÑOS' AND TRY_CAST(EDAD AS INT) IS NOT NULL
            GROUP BY 1 ORDER BY 1""",
        params,
    )
    otros = q(
        f"""SELECT TIEMPO_EDAD, COUNT(*) n FROM tbl_SINADEF
            WHERE {where} AND TIEMPO_EDAD <> 'AÑOS'
            GROUP BY 1 ORDER BY n DESC""",
        params,
    )
    return {"decadas": bins, "otras_unidades": otros}


@app.get("/api/registros")
def registros(
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    desde: Optional[str] = None,
    hasta: Optional[str] = None,
    limite: int = Query(100, ge=1, le=1000),
):
    """Muestra de registros crudos, para inspeccionar la data detrás de un filtro."""
    where, params = build_where(tipo, sexo, departamento, lugar, desde, hasta)
    return q(
        f"""SELECT FECHA, SEXO, EDAD, TIEMPO_EDAD, MUERTE_VIOLENTA, TIPO_LUGAR,
                   DEPARTAMENTO_FALLECIMIENTO, PROVINCIA_FALLECIMIENTO, DISTRITO_FALLECIMIENTO,
                   NECROPSIA, DEBIDO_CAUSA_A, CAUSA_A_CIEX
            FROM tbl_SINADEF WHERE {where}
            ORDER BY FECHA DESC LIMIT {limite}""",
        params,
    )


# ── Proyección ──────────────────────────────────────────────────────────


def _cierre_anual(anuales, serie, res, anio_proyectado, corte_efectivo):
    """
    Cómo cerraría el año proyectado, comparado con los años ya cerrados.

    El año en curso se parte en dos: lo que SINADEF ya registró hasta el mes de
    corte y lo que proyecta el modelo de ahí a diciembre. Se mantienen separados
    para que se vea qué parte es dato y qué parte es estimación.
    """
    observado_curso = int(
        sum(v for p, v in serie.items() if str(p).startswith(anio_proyectado))
    )
    proyectado = float(sum(res["central"]))
    bandas = res.get("bandas") or {}
    lo = observado_curso + float(sum(bandas.get("p10", []))) if bandas.get("p10") else None
    hi = observado_curso + float(sum(bandas.get("p90", []))) if bandas.get("p90") else None

    filas = []
    for a in anuales:
        if a["ANIO"] < anio_proyectado:
            filas.append({
                "anio": a["ANIO"], "observado": a["n"], "proyectado": 0,
                "total": a["n"], "es_proyeccion": False,
            })
    cierre = observado_curso + proyectado
    filas.append({
        "anio": anio_proyectado,
        "observado": observado_curso,
        "proyectado": round(proyectado, 1),
        "total": round(cierre, 1),
        "es_proyeccion": True,
        "rango": f"{round(lo):,} – {round(hi):,}".replace(",", " ") if lo and hi else None,
    })

    previos = [f["total"] for f in filas if not f["es_proyeccion"]]
    anterior = previos[-1] if previos else None
    promedio = sum(previos) / len(previos) if previos else None

    return {
        "filas": filas,
        "anio": anio_proyectado,
        "observado": observado_curso,
        "proyectado": round(proyectado, 1),
        "cierre": round(cierre),
        "rango_lo": round(lo) if lo else None,
        "rango_hi": round(hi) if hi else None,
        "meses_observados": sum(1 for p in serie.index if str(p).startswith(anio_proyectado)),
        "mes_corte": str(corte_efectivo),
        "vs_anterior": (
            {
                "anio": filas[-2]["anio"],
                "valor": anterior,
                "pct": round(100.0 * (cierre - anterior) / anterior, 1),
            }
            if anterior
            else None
        ),
        "vs_promedio": (
            {"valor": round(promedio), "pct": round(100.0 * (cierre - promedio) / promedio, 1)}
            if promedio
            else None
        ),
    }


@app.post("/api/forecast")
def lanzar_forecast(
    mes_corte: str = Query(..., description="Último mes que entra al modelo, 'YYYY-MM'"),
    hasta: str = Query("2026-12", description="Último mes a proyectar, 'YYYY-MM'"),
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    excluir_meses: int = Query(
        0, ge=0, le=6,
        description="Meses finales a descartar del entrenamiento por registro tardío",
    ),
    ventana: int = Query(3, ge=2, le=6, description="Meses por ventana de test"),
    origenes: int = Query(6, ge=1, le=12, description="Ventanas de rolling origin"),
):
    """
    Encola una proyección. Devuelve el id del trabajo; el avance se consulta en
    /api/forecast/{job_id}.
    """
    import time

    import pandas as pd

    import forecast as fc
    import jobs
    import store

    try:
        corte = pd.Period(mes_corte, freq="M")
        fin = pd.Period(hasta, freq="M")
    except Exception:
        raise HTTPException(400, "mes_corte y hasta deben tener formato 'YYYY-MM'")

    corte_efectivo = corte - excluir_meses
    h = (fin - corte_efectivo).n
    if h < 1:
        raise HTTPException(400, f"El horizonte debe ser al menos 1 mes (calculado: {h})")
    if h > 36:
        raise HTTPException(400, "Horizonte máximo de 36 meses")

    # Firma de la versión actual de la base. Va dentro de la clave de caché, así
    # que cuando el ETL de las 6am traiga datos nuevos —o corrija meses viejos,
    # que ocurre a diario— la firma cambia y la proyección se recalcula desde
    # cero en lugar de servir un resultado apoyado en datos que ya no existen.
    base = q("SELECT MAX(FECHA) f, COUNT(*) n FROM tbl_SINADEF")[0]
    firma = store.firma_datos(base["f"], base["n"])

    params = {
        "mes_corte": str(corte_efectivo), "hasta": hasta, "tipo": tipo, "sexo": sexo,
        "departamento": departamento, "lugar": lugar,
        "ventana": ventana, "origenes": origenes,
        "excluir_meses": excluir_meses,
        "firma_datos": firma,
        "algo_v": fc.VERSION,
    }

    # El repositorio en disco sobrevive a los reinicios y es compartido por
    # todos los usuarios: si alguien ya corrió esta misma combinación sobre esta
    # misma versión de los datos, se devuelve al instante.
    guardada = store.buscar(store.clave(params, firma))
    if guardada:
        return jobs.estado(jobs.desde_repositorio(params, guardada)["id"])

    where, sql_params = build_where(tipo, sexo, departamento, lugar, None, None)
    filas = q(
        f"""SELECT ANIO || '-' || MES AS mes, COUNT(*) n
            FROM tbl_SINADEF WHERE {where} AND ANIO || '-' || MES <= ?
            GROUP BY 1 ORDER BY 1""",
        sql_params + [str(corte_efectivo)],
    )
    if len(filas) < 40:
        raise HTTPException(
            400,
            f"La serie filtrada tiene {len(filas)} meses; se necesitan al menos 40 "
            f"(3 años de entrenamiento más las ventanas de validación).",
        )

    serie = pd.Series(
        [f["n"] for f in filas],
        index=pd.PeriodIndex([f["mes"] for f in filas], freq="M"),
    ).sort_index()
    # Un mes sin defunciones no viene como fila: se rellena para no dejar huecos.
    serie = serie.reindex(
        pd.period_range(serie.index[0], serie.index[-1], freq="M"), fill_value=0
    )

    historico = [{"mes": str(p), "n": int(v)} for p, v in serie.items()]

    # Totales anuales con los mismos filtros: son la base para comparar cómo
    # cerraría el año en curso contra los años ya cerrados.
    anuales = q(
        f"SELECT ANIO, COUNT(*) n FROM tbl_SINADEF WHERE {where} GROUP BY 1 ORDER BY 1",
        sql_params,
    )
    anio_proyectado = str(fin.year)

    def trabajo(avance):
        t0 = time.time()
        res = fc.proyectar(
            serie, h, n_jobs=8, ventana=ventana, k_origenes=origenes, progreso=avance
        )
        res["historico"] = historico
        res["filtros"] = {
            "tipo": tipo or "(todas)", "sexo": sexo or "(todos)",
            "departamento": departamento or "(todos)", "lugar": lugar or "(todos)",
        }
        res["mes_corte_usado"] = str(corte_efectivo)
        res["meses_excluidos"] = excluir_meses
        res["horizonte"] = h
        res["cierre_anual"] = _cierre_anual(
            anuales, serie, res, anio_proyectado, corte_efectivo
        )
        res["firma_datos"] = firma
        res["datos_hasta"] = base["f"]
        store.guardar(store.clave(params, firma), firma, params, res, time.time() - t0)
        return res

    # Se devuelve vía estado() para que POST y GET tengan exactamente la misma
    # forma y el frontend no tenga que tratarlos distinto.
    return jobs.estado(jobs.lanzar(params, trabajo)["id"])


def _pendiente(valores: list[float]) -> dict:
    """
    Regresión lineal sobre la serie mensual del periodo.

    La tendencia se expresa en % anual sobre el nivel medio del periodo, no en
    casos por año. Así es comparable entre territorios de distinto tamaño: +5
    casos/año significan cosas muy distintas en Lima que en Pataz, mientras que
    +12% anual se lee igual en ambos. Además evita el problema de redondear
    pendientes pequeñas a cero.
    """
    import numpy as np

    n = len(valores)
    vacio = {"anual": None, "pct": None, "t": None, "direccion": "sin datos"}
    if n < 6:
        return vacio
    x = np.arange(n, dtype=float)
    y = np.asarray(valores, dtype=float)
    mx, my = x.mean(), y.mean()
    sxx = float(((x - mx) ** 2).sum())
    if sxx == 0 or my == 0:
        return vacio
    b = float(((x - mx) * (y - my)).sum() / sxx)
    a = my - b * mx
    sse = float(((y - (a + b * x)) ** 2).sum())
    se = (sse / (n - 2) / sxx) ** 0.5 if n > 2 and sse > 0 else 0.0
    t = b / se if se > 0 else (0.0 if b == 0 else float(np.sign(b)) * 99)
    return {
        "anual": round(b * 12, 2),               # casos/año, para referencia
        "pct": round(b * 12 / my * 100, 1),      # % anual sobre el nivel medio
        "t": round(float(t), 2),
        "direccion": "estable" if abs(t) < 2 else ("sube" if b > 0 else "baja"),
        # Coeficientes crudos: con ellos se evalúa la recta en cualquier punto,
        # que es lo que permite calcular el nivel al inicio, al final y el
        # contrafactual.
        "_a": a, "_b": b, "_n": n,
    }


@app.get("/api/gestion")
def gestion(
    nivel: str = Query("departamento", description="'departamento' o 'provincia'"),
    p1_desde: str = Query("2017-01"), p1_hasta: str = Query("2022-12"),
    p2_desde: str = Query("2023-01"), p2_hasta: Optional[str] = None,
    tipo: Optional[str] = "HOMICIDIO",
    sexo: Optional[str] = None,
    lugar: Optional[str] = None,
    minimo: int = Query(10, ge=0, description="Casos mínimos en el periodo 1 para incluir el territorio"),
):
    """
    Compara dos periodos por territorio: el que la autoridad heredó y el suyo.

    Lima se parte en dos: la provincia de Lima es Lima Metropolitana, cuyo
    alcalde ejerce además funciones regionales; el resto del departamento es la
    región Lima Provincias, que tiene gobernador propio.
    """
    if nivel not in ("departamento", "provincia", "distrito"):
        raise HTTPException(400, "nivel debe ser 'departamento', 'provincia' o 'distrito'")

    if not p2_hasta:
        # El último mes con datos siempre está incompleto por el rezago de
        # registro. Incluirlo mete un punto hundido al final de la serie que
        # inclina la recta hacia abajo y fabrica mejoras inexistentes, así que
        # el periodo se cierra en el último mes ya transcurrido por completo.
        ultimo = q("SELECT MAX(ANIO || '-' || MES) m FROM tbl_SINADEF")[0]["m"]
        a, m = int(ultimo[:4]), int(ultimo[5:7])
        m -= 1
        if m == 0:
            m, a = 12, a - 1
        p2_hasta = f"{a:04d}-{m:02d}"

    clauses, params = ["DEPARTAMENTO_FALLECIMIENTO <> 'SIN REGISTRO'"], []
    for valor, col in ((tipo, "MUERTE_VIOLENTA"), (sexo, "SEXO"), (lugar, "TIPO_LUGAR")):
        if valor:
            ops = [v.strip() for v in valor.split(",") if v.strip()]
            if ops:
                clauses.append(f"{col} IN ({','.join('?' * len(ops))})")
                params.extend(ops)
    where = " AND ".join(clauses)

    if nivel == "departamento":
        territorio = """CASE WHEN DEPARTAMENTO_FALLECIMIENTO='LIMA' AND PROVINCIA_FALLECIMIENTO='LIMA'
                THEN 'LIMA METROPOLITANA'
                WHEN DEPARTAMENTO_FALLECIMIENTO='LIMA' THEN 'LIMA PROVINCIAS'
                ELSE DEPARTAMENTO_FALLECIMIENTO END"""
        extra = ", '' AS padre"
        excluir = ""
    elif nivel == "provincia":
        territorio = "PROVINCIA_FALLECIMIENTO"
        extra = ", DEPARTAMENTO_FALLECIMIENTO AS padre"
        excluir = " AND PROVINCIA_FALLECIMIENTO <> 'SIN REGISTRO'"
    else:
        territorio = "DISTRITO_FALLECIMIENTO"
        # 115 nombres de distrito se repiten entre provincias (hay varios
        # "San Juan", "Independencia", etc.), así que el padre no es decorativo:
        # sin él se fundirían distritos distintos en una sola fila.
        extra = ", PROVINCIA_FALLECIMIENTO || ' · ' || DEPARTAMENTO_FALLECIMIENTO AS padre"
        excluir = " AND DISTRITO_FALLECIMIENTO <> 'SIN REGISTRO'"

    filas = q(
        f"""SELECT {territorio} AS territorio{extra}, ANIO || '-' || MES AS mes, COUNT(*) n
            FROM tbl_SINADEF WHERE {where}{excluir}
              AND ANIO || '-' || MES >= ? AND ANIO || '-' || MES <= ?
            GROUP BY 1, 2, 3 ORDER BY 1, 3""",
        params + [p1_desde, p2_hasta],
    )

    # La clave agrupa territorio + padre, nunca el nombre solo.
    por_terr: dict[tuple[str, str], dict] = {}
    for f in filas:
        d = por_terr.setdefault(
            (f["territorio"], f["padre"]), {"padre": f["padre"], "meses": {}}
        )
        d["meses"][f["mes"]] = f["n"]

    def rango(mm: dict, desde: str, hasta: str) -> list[float]:
        """Serie mensual completa del tramo; los meses sin casos cuentan 0."""
        out, a, m = [], int(desde[:4]), int(desde[5:7])
        while f"{a:04d}-{m:02d}" <= hasta:
            out.append(float(mm.get(f"{a:04d}-{m:02d}", 0)))
            m += 1
            if m == 13:
                m, a = 1, a + 1
        return out

    salida = []
    for (terr, _padre), d in por_terr.items():
        s1, s2 = rango(d["meses"], p1_desde, p1_hasta), rango(d["meses"], p2_desde, p2_hasta)
        n1, n2 = sum(s1), sum(s2)
        if n1 < minimo:
            continue
        pr1 = n1 / len(s1) if s1 else 0
        pr2 = n2 / len(s2) if s2 else 0
        t1, t2 = _pendiente(s1), _pendiente(s2)

        # Los niveles son CONTEOS OBSERVADOS de doce meses, no puntos de una
        # recta ajustada. Una recta extrapola y puede devolver cifras que nunca
        # ocurrieron (Talara "recibía 5.9" cuando en 2022 tuvo 8 casos reales).
        # Ambas ventanas son de doce meses, así que la estacionalidad se cancela
        # y las cifras son verificables una por una contra el registro.
        recibido = float(sum(s1[-12:])) if len(s1) >= 12 else None
        actual = float(sum(s2[-12:])) if len(s2) >= 12 else None

        # Contrafactual: el nivel heredado más la pendiente previa prolongada por
        # los años transcurridos.
        #
        # La pendiente solo se aplica si es estadísticamente significativa. Si la
        # tendencia heredada se declaró "plana", proyectar con su pendiente cruda
        # es contradecirse: Virú aparecía como "lo heredó plano" y a la vez con
        # una proyección de 19 sobre un nivel de 13, es decir arrastrando un alza
        # que la propia etiqueta niega.
        contraf = None
        if recibido is not None and t1.get("anual") is not None:
            anios = len(s2) / 12
            pendiente = t1["anual"] if t1["direccion"] != "estable" else 0.0
            contraf = max(0.0, recibido + pendiente * 12 * anios)

        # El veredicto se decide contra el CONTRAFACTUAL, no por la
        # significancia de la pendiente. Con series pequeñas el test t casi
        # nunca alcanza |t|>2, así que usarlo como criterio hacía que "frenó"
        # fuese el resultado por defecto de la falta de datos: Talara aparecía
        # "frenando" estando 12% por encima de su propia proyección, y Pataz
        # estando 151% por encima.
        recibio_alza = t1["direccion"] == "sube"
        # El ratio se calcula con los MISMOS valores redondeados que se muestran.
        # Con los decimales internos, Virú daba 0.852 (→ "sin cambio") mientras
        # en pantalla se leía 15/18 = 0.83, que cae del otro lado del umbral:
        # quien rehiciera la cuenta vería una contradicción.
        r_act = round(actual) if actual is not None else None
        r_ctf = round(contraf) if contraf is not None else None
        ratio = (r_act / r_ctf) if (r_ctf and r_ctf > 0 and r_act is not None) else None

        # Cinco niveles y no más: siete categorías no se pueden colorear de forma
        # distinguible (el validador de paleta rechaza cualquier escala de siete
        # pasos, y el par naranja/rojo quedaba en ΔE 5.1 sobre fondo oscuro).
        # Lo que se pierde al agrupar —si el territorio heredó o no una escalada—
        # ya está en la columna "Lo heredó".
        # El ratio manda siempre, haya heredado o no una escalada. Tratar esos
        # dos casos por separado producía veredictos falsos: San Juan de
        # Miraflores pasó de 29 a 60 homicidios —un 82% por encima de lo
        # proyectado— y salía como "Sin cambio" solo porque su tendencia previa
        # no era significativa.
        if ratio is None:
            # Sin base con la que comparar (2022 en cero, o proyección nula):
            # el ratio no existe, así que se decide por los niveles directos.
            # Antes caía en "sin cambio" por defecto, y territorios que pasaron
            # de 0 a 8 homicidios aparecían como si nada hubiera ocurrido.
            if r_act is None or r_act == recibido:
                veredicto = "sigue"
            elif r_act < recibido:
                veredicto = "bajo"
            elif recibido == 0:
                # Aparece violencia donde no la había registrada.
                veredicto = "disparo" if r_act >= 3 else "empeoro"
            else:
                veredicto = "disparo" if r_act > recibido * 1.5 else "empeoro"
        elif ratio < 0.85:
            # "Bajó" exige que el nivel actual sea MENOR que el heredado: es lo
            # primero que verifica quien lee la fila. Un territorio que subió de
            # 13 a 15 no "bajó", aunque haya crecido menos de lo proyectado —
            # eso es haber frenado, no haber bajado.
            veredicto = "bajo" if (r_act is not None and r_act < recibido) else "freno"
        elif ratio <= 1.15:
            veredicto = "sigue"
        elif ratio <= 1.5:
            veredicto = "empeoro"
        else:
            veredicto = "disparo"

        for t in (t1, t2):
            for k in ("_a", "_b", "_n"):
                t.pop(k, None)

        salida.append({
            # id estable para seleccionar en la UI sin confundir homónimos.
            "id": f"{terr}|{d['padre']}",
            "territorio": terr, "padre": d["padre"],
            "n1": int(n1), "n2": int(n2),
            # Los tres números que hacen entendible el veredicto.
            "recibido": round(recibido) if recibido is not None else None,
            "actual": round(actual) if actual is not None else None,
            "contrafactual": round(contraf) if contraf is not None else None,
            # Se mira el MENOR de los dos extremos, no solo el nivel actual: una
            # base de 4 casos produce variaciones espectaculares aunque hoy haya
            # 32. Chepén (4 → 32) pasaba sin marca mostrando un +700%.
            "volumen_bajo": (
                min(round(recibido), round(actual)) < 12
                if recibido is not None and actual is not None else True
            ),
            "var_real": round((actual / recibido - 1) * 100) if recibido else None,
            "vs_contraf": round((actual / contraf - 1) * 100) if contraf else None,
            "tend1": t1, "tend2": t2,
            "veredicto": veredicto,
            "serie": [{"mes": m, "n": v} for m, v in sorted(d["meses"].items())],
        })

    salida.sort(key=lambda r: r["n2"], reverse=True)
    return {
        "nivel": nivel,
        "periodo1": {"desde": p1_desde, "hasta": p1_hasta},
        "periodo2": {"desde": p2_desde, "hasta": p2_hasta},
        "territorios": salida,
        "total": len(salida),
    }


@app.post("/api/nowcast")
def lanzar_nowcast(
    mes: Optional[str] = Query(None, description="Mes a cerrar 'YYYY-MM'; por defecto el último con datos"),
    tipo: Optional[str] = None,
    sexo: Optional[str] = None,
    departamento: Optional[str] = None,
    lugar: Optional[str] = None,
    ventana: int = Query(24, ge=6, le=60, description="Meses de entrenamiento"),
    origenes: int = Query(12, ge=4, le=24, description="Orígenes del backtest"),
):
    """
    Proyecta el cierre del mes en curso a partir de la serie diaria.

    A diferencia de /api/forecast (que trabaja con meses cerrados), aquí se
    combina lo ya observado del mes, el rezago de registro pendiente y una
    predicción de los días que faltan.
    """
    import jobs
    import nowcast as nw
    import store

    # El filtro se arma con la misma whitelist del resto de la API; los valores
    # se escapan porque aquí viajan dentro del SQL en lugar de como parámetros.
    def esc(v: str) -> str:
        return v.replace("'", "''")

    partes = []
    for valor, columna in ((tipo, "MUERTE_VIOLENTA"), (sexo, "SEXO"),
                           (departamento, "DEPARTAMENTO_FALLECIMIENTO"), (lugar, "TIPO_LUGAR")):
        if valor:
            ops = [f"'{esc(v.strip())}'" for v in valor.split(",") if v.strip()]
            if ops:
                partes.append(f"{columna} IN ({','.join(ops)})")
    filtro_sql = " AND ".join(partes) if partes else "1=1"

    base = q("SELECT MAX(FECHA) f, COUNT(*) n FROM tbl_SINADEF")[0]
    firma = store.firma_datos(base["f"], base["n"])
    params = {
        "modo": "nowcast", "mes": mes, "tipo": tipo, "sexo": sexo,
        "departamento": departamento, "lugar": lugar,
        "ventana": ventana, "origenes": origenes, "firma_datos": firma,
        # Versiona el método: al cambiar el algoritmo, lo guardado con la
        # versión anterior deja de reutilizarse.
        "algo_v": nw.VERSION,
    }

    guardada = store.buscar(store.clave(params, firma))
    if guardada:
        return jobs.estado(jobs.desde_repositorio(params, guardada)["id"])

    def trabajo(avance):
        import time as _t
        t0 = _t.time()
        res = nw.proyectar_mes(
            filtro_sql=filtro_sql, mes=mes, meses_ventana=ventana,
            n_origenes=origenes, progreso=avance,
        )
        res["filtros"] = {
            "tipo": tipo or "(todas)", "sexo": sexo or "(todos)",
            "departamento": departamento or "(todos)", "lugar": lugar or "(todos)",
        }
        res["firma_datos"] = firma
        store.guardar(store.clave(params, firma), firma, params, res, _t.time() - t0)
        return res

    return jobs.estado(jobs.lanzar(params, trabajo)["id"])


@app.get("/api/forecast-historial")
def historial_forecast(
    limite: int = Query(50, ge=1, le=200),
    solo_vigentes: bool = Query(False, description="Solo las hechas con la versión actual de la base"),
):
    """
    Proyecciones ya calculadas, compartidas entre todos los usuarios.

    Las que no son 'vigentes' se hicieron con una versión anterior de la base;
    se conservan para poder contrastar después qué se proyectó y cuándo contra
    lo que realmente ocurrió.
    """
    import store

    base = q("SELECT MAX(FECHA) f, COUNT(*) n FROM tbl_SINADEF")[0]
    firma = store.firma_datos(base["f"], base["n"])
    filas = store.historial(limite, firma if solo_vigentes else None)
    for f in filas:
        f["vigente"] = f["firma_datos"] == firma
    return {
        "firma_actual": firma,
        "datos_hasta": base["f"],
        "proyecciones": filas,
        **store.estadisticas(),
    }


@app.get("/api/forecast/{job_id}")
def estado_forecast(job_id: str):
    import jobs

    e = jobs.estado(job_id)
    if not e:
        raise HTTPException(404, "Trabajo no encontrado o expirado")
    return e


@app.get("/health")
def health():
    try:
        n = q("SELECT COUNT(*) n FROM tbl_SINADEF")[0]["n"]
        return {"status": "ok", "filas": n}
    except Exception as e:
        raise HTTPException(503, str(e))


# ── Frontend ────────────────────────────────────────────────────────────
# El build de React se sirve desde la misma app: un solo proceso que levantar.

if FRONTEND_DIST.exists():
    app.mount("/assets", StaticFiles(directory=FRONTEND_DIST / "assets"), name="assets")

    # Páginas de análisis estáticas (frontend/public/analisis → dist/analisis).
    # Tienen URL propia para que al compartirlas X muestre la tarjeta de vista
    # previa; html=True sirve el index.html de cada carpeta.
    if (FRONTEND_DIST / "analisis").exists():
        app.mount("/analisis", StaticFiles(directory=FRONTEND_DIST / "analisis", html=True),
                  name="analisis")

    @app.get("/")
    def index():
        return FileResponse(FRONTEND_DIST / "index.html")