"""MP1 · paso 10: construcción de la tabla de lotes (desenlaces a, b, c) con variables ex ante y ventanas de historial.

Pre-registro: prereg/DEFINICIONES_MP1.md (md5 3da2be07…). Imagen buscalicitaciones-etl:v2 (duckdb 1.1.3, sin numpy).
Usa, en solo lectura, el módulo de E1 (/app = <repo>/ @ main) para la población de lotes (x_lotes).

Salidas (en /w/interno, 700):
  lotes.csv        un lote por fila: claves, conjunto (hist / pred), atributos ex ante, filtros de población P,
                   desenlaces (y_ou, y_des, y_mod, y_mod10) y, por nivel y desenlace, n y k de su ventana de historial
  informe_10.json  recuentos, tiempos, versiones
Uso: python3 10_construir.py [--perturbar]   (--perturbar: test dinámico ex ante, §12 (6): altera todos los desenlaces y
     columnas posteriores al anuncio de los lotes con anuncio >= 2026-01-01 y escribe lotes_perturbado.csv)
"""
import json, sys, time, resource, datetime as dt
from pathlib import Path
sys.path.insert(0, "/app")
import duckdb
from etl.v2 import exante_l1 as e1

T0 = time.time()
PERT = "--perturbar" in sys.argv
PANIO = int(sys.argv[sys.argv.index("--perturbar") + 1]) if PERT else None   # 2026: prueba de (a)/(b); 2025: prueba de (c)
PVENT = {2026: ("2026-01-01", "2026-04-30"), 2025: ("2025-01-01", "2025-04-15")}
U = Path("/d/unified_20261025")
SAL = Path("/w/interno"); SAL.mkdir(exist_ok=True)
TMP = Path("/w/tmp/duck"); TMP.mkdir(parents=True, exist_ok=True)
FECHA_CORTE = dt.date(2026, 10, 25)
inf = {"duckdb": duckdb.__version__, "python": sys.version.split()[0], "perturbado": PERT, "unified": str(U)}

con = e1.conexion(memoria="9GB", hilos=6, temporal=TMP)
e1.paso_entrada(con, U, Path(f"{U}.identidad"), Path(f"{U}.criterios"), Path(f"{U}.documentos"), Path(f"{U}.cpvs"))
e1.paso_lotes(con)
inf["x_lotes"] = con.execute("SELECT count(*), count(*) FILTER (WHERE feed='propio') FROM x_lotes").fetchone()
print("x_lotes", inf["x_lotes"], round(time.time() - T0), "s", flush=True)

con.execute("CREATE OR REPLACE TEMP VIEW vf AS SELECT source, id_canonical, cuenta FROM read_parquet('/d/unified_20261025.vinculos/filas.parquet') WHERE source = 'PLACSP'")
con.execute("CREATE OR REPLACE TEMP VIEW x_mod AS SELECT * FROM read_parquet('/d/unified_20261025.modificaciones/*.parquet')")

# --- filas de cada lote: contratos adjudicados y copias --------------------------------------------------------------
con.execute(f"""CREATE OR REPLACE TEMP TABLE x_lr AS
    SELECT CASE WHEN u.tabla_origen = 'resultados_licitaciones' THEN 'propio' ELSE 'agregacion' END feed, u.licitacion_id,
        coalesce(u.lote_id, u.licitacion_id) lote_u,
        list(u.id_canonical) FILTER (WHERE u.unidad = 'adjudicacion') ids_adj,
        bool_and(v.cuenta IS NOT NULL AND NOT v.cuenta) todo_copia
    FROM x_u u LEFT JOIN vf v ON v.id_canonical = u.id_canonical
    WHERE {e1._filtro_poblacion('u')} AND u.tabla_origen IN ('resultados_licitaciones', 'resultados_agregacion')
        AND u.anio_ref BETWEEN 2012 AND 2026
    GROUP BY 1, 2, 3""")

if PERT:
    # Test dinámico: todo lo posterior al anuncio de los lotes anunciados desde el 1-ene-2026 se altera (la etiqueta incluida).
    # Las variables de esos lotes no pueden cambiar (pre-registro §12 (6)).
    # Pre-registro §12 (6), corregido el 26-oct (la primera versión perturbaba por fecha_pub >= 2026 y tocaba lotes resueltos en
    # 2022-2024 con la fecha de anuncio rehecha, que entran legítimamente en el historial): se alteran los desenlaces de las
    # unidades de la prueba del año PANIO y de todo lo resuelto desde PANIO, y se quitan las modificaciones publicadas desde PANIO.
    a0, a1 = PVENT[PANIO]
    con.execute(f"""UPDATE x_lotes SET num_ofertas = 1 + (hash(licitacion_id || lote_u) % 7)::INTEGER,
                       desierto = NOT coalesce(desierto, FALSE), adjudicado = NOT coalesce(adjudicado, FALSE),
                       fecha_adj = fecha_adj + 40, anio_ref = anio_ref + 1
                   WHERE anio_ref >= {PANIO} OR (fpub_ok AND fecha_pub BETWEEN DATE '{a0}' AND DATE '{a1}')""")
    con.execute(f"""CREATE OR REPLACE TEMP VIEW x_mod AS SELECT * FROM read_parquet('/d/unified_20261025.modificaciones/*.parquet')
                   WHERE fecha_primera_version < DATE '{PANIO}-01-01'""")

# --- base de lotes del feed propio con sus marcas ----------------------------------------------------------------------
con.execute("""CREATE OR REPLACE TEMP TABLE x_b AS
    SELECT l.*, coalesce(r.todo_copia, FALSE) todo_copia, r.ids_adj,
        (l.adjudicado OR l.desierto) resuelto, (l.desierto AND NOT l.adjudicado) des,
        year(l.fecha_adj) anio_adj,
        count(*) OVER (PARTITION BY l.licitacion_id) n_lotes_lic
    FROM x_lotes l LEFT JOIN x_lr r USING (feed, licitacion_id, lote_u)
    WHERE l.feed = 'propio'""")

# --- (c) modificación temprana del lote -----------------------------------------------------------------------------
con.execute("""CREATE OR REPLACE TEMP TABLE x_ma AS
    SELECT b.licitacion_id, b.lote_u, unnest(b.ids_adj) id_canonical, b.fecha_adj FROM x_b b WHERE b.adjudicado AND b.ids_adj IS NOT NULL""")
con.execute("""CREATE OR REPLACE TEMP TABLE x_ym AS
    WITH c AS (
        SELECT a.licitacion_id, a.lote_u, m.fecha_primera_version fpv, m.cambia_importe, m.importe_modificacion_sin_iva imod,
               m.importe_adjudicacion_sin_iva iadj, a.fecha_adj
        FROM x_ma a JOIN x_mod m ON m.id_canonical = a.id_canonical AND m.contrato_casado = 'uno' AND m.tipo = 'modificacion'
        UNION ALL
        SELECT b.licitacion_id, b.lote_u, m.fecha_primera_version, m.cambia_importe, m.importe_modificacion_sin_iva,
               m.importe_adjudicacion_sin_iva, b.fecha_adj
        FROM x_b b JOIN x_mod m ON m.licitacion_id = b.licitacion_id AND m.casa_con = 'licitacion' AND m.tipo = 'modificacion'
        WHERE b.adjudicado AND b.n_lotes_lic = 1 AND b.lote_u = b.licitacion_id)
    SELECT licitacion_id, lote_u,
        bool_or(fpv BETWEEN fecha_adj AND fecha_adj + 365) y_mod,
        bool_or(fpv BETWEEN fecha_adj AND fecha_adj + 365 AND coalesce(cambia_importe, FALSE) AND iadj > 0
                AND abs(imod) >= 0.10 * iadj) y_mod10
    FROM c GROUP BY 1, 2""")
con.execute("""CREATE OR REPLACE TEMP TABLE x_b2 AS
    SELECT b.*, CASE WHEN b.adjudicado THEN coalesce(m.y_mod, FALSE) END y_mod,
                CASE WHEN b.adjudicado THEN coalesce(m.y_mod10, FALSE) END y_mod10
    FROM x_b b LEFT JOIN x_ym m USING (licitacion_id, lote_u)""")

# --- historial: celdas-año (sin copias; todos los lotes resueltos del feed propio) -------------------------------------
# ou: n = resueltos con ofertas, k = una oferta; des: n = resueltos, k = desierto; por anio_ref.  mod: por año de adjudicación.
con.execute("""CREATE OR REPLACE TEMP TABLE x_h AS
    SELECT organo_clave, coalesce(cpv_div, '??') cpv_cel, proc_nat, coalesce(ccaa, '??') ccaa_c, coalesce(tipo_organo, '??') toc,
        anio_ref, anio_adj,
        (resuelto AND num_ofertas >= 1)::INTEGER ou_n, (resuelto AND num_ofertas = 1)::INTEGER ou_k,
        resuelto::INTEGER des_n, des::INTEGER des_k,
        adjudicado::INTEGER mod_n, coalesce(y_mod, FALSE)::INTEGER mod_k
    FROM x_b2 WHERE NOT todo_copia AND (resuelto) AND anio_ref BETWEEN 2012 AND 2026""")

NIVELES = {  # nombre: (claves, ventana)
    "glob": ([], "3y"), "mkt": (["cpv_cel", "proc_nat"], "3y"), "cpv": (["cpv_cel"], "3y"),
    "terr": (["ccaa_c", "cpv_cel"], "3y"), "toc": (["toc", "ccaa_c"], "3y"),
    "org": (["organo_clave"], "cum"), "orgl": (["organo_clave"], "last"), "orgcpv": (["organo_clave", "cpv_cel"], "cum"),
}

def ventana(v, col, s):
    if v == "cum":
        return f"{col} < d.T - {s}"
    if v == "last":
        return f"{col} = d.T - 1 - {s}"
    return f"{col} BETWEEN d.T - 3 - {s} AND d.T - 1 - {s}"

# --- objetos: lotes históricos (anuncio 2014-2026) y anuncios en plazo a predecir -------------------------------------
con.execute("""CREATE OR REPLACE TEMP TABLE x_nla AS SELECT licitacion_id, count(DISTINCT lote) n_lotes_an
               FROM x_cpv_raw WHERE ambito = 'lote' AND lote IS NOT NULL GROUP BY 1""")
procs = ", ".join(e1._lit(p) for p in e1.PROCEDIMIENTOS)
con.execute(f"""CREATE OR REPLACE TEMP TABLE x_o AS
    SELECT 'hist' conjunto, b.licitacion_id, b.lote_u, b.organo_clave, coalesce(b.cpv_div, '??') cpv_cel, b.cpv_div, b.proc_nat,
        b.tram, b.tipo_contrato, b.tipo_organo, b.ccaa, b.mes_pub, coalesce(b.pbl_lote, b.pbl_exp) pbl, b.ve_exp, b.dur,
        b.plazo_anuncio plazo, b.sara, b.ue, b.cpv_n_div, (b.lote_u <> b.licitacion_id) es_lote, b.fecha_pub, b.ffo,
        year(b.fecha_pub)::INTEGER T,
        -- filtros de población P (§2) y desenlaces (solo etiqueta)
        b.fpub_ok, b.todo_copia, b.resuelto, b.adjudicado, b.des, b.desistido, b.num_ofertas, b.fecha_adj,
        (b.fecha_adj - b.fecha_pub) dias_res, b.y_mod, b.y_mod10, b.organo_nom
    FROM x_b2 b WHERE b.fecha_pub BETWEEN DATE '2014-01-01' AND DATE '2026-10-25'
    UNION ALL
    SELECT 'pred', p.licitacion_id, coalesce(p.lote_id, p.licitacion_id), o.organo_clave, coalesce(p.cpv_division, '??'),
        p.cpv_division, p.procedimiento_native, p.tramitacion, p.tipo_contrato, t.tipo, p.ccaa, month(p.fecha_pub),
        coalesce(p.imp_lic_eur, p.imp_lic_eur_expediente), p.valor_estimado_eur_expediente, p.duracion_meses,
        CASE WHEN p.fecha_fin_ofertas - p.fecha_pub > 0 AND p.fecha_fin_ofertas - p.fecha_pub <= 180 THEN p.fecha_fin_ofertas - p.fecha_pub END,
        p.sara, p.financiacion_ue, coalesce(cl.n_divisiones, ce.n_divisiones), (p.lote_id IS NOT NULL), p.fecha_pub, p.fecha_fin_ofertas, 2026,
        TRUE, FALSE, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, {e1._nom('p.organo_nombre')}
    FROM x_u p
    LEFT JOIN x_of o ON o.organo_nif IS NOT DISTINCT FROM p.organo_nif AND o.organo_nombre IS NOT DISTINCT FROM p.organo_nombre
        AND o.dir3 IS NOT DISTINCT FROM p.dir3
    LEFT JOIN x_ot t ON t.organo_clave = o.organo_clave
    LEFT JOIN x_cpv_lote cl ON cl.lote_id = coalesce(p.lote_id, p.licitacion_id)
    LEFT JOIN x_cpv_exp ce ON ce.licitacion_id = p.licitacion_id
    WHERE p.source = 'PLACSP' AND p.vigente AND p.es_contrato AND NOT p.es_menor AND p.url_origen LIKE '%contrataciondelestado%'
        AND p.tabla_origen = 'licitaciones_sin_resultado' AND p.unidad = 'licitacion:sin_resultado' AND p.estado = 'Publicada'
        AND p.fecha_fin_ofertas >= DATE '{FECHA_CORTE.isoformat()}' AND p.procedimiento IN ({procs})
        AND coalesce(p.rol_am, '') NOT IN ('basado', 'basado_sda') AND coalesce(p.tramitacion, '') <> 'emergencia'""")
n_dup = con.execute("SELECT count(*) - count(DISTINCT (conjunto, licitacion_id, lote_u)) FROM x_o").fetchone()[0]
inf["x_o"] = con.execute("SELECT conjunto, count(*) FROM x_o GROUP BY 1 ORDER BY 1").fetchall()
inf["x_o_duplicados_pred"] = n_dup
# Los anuncios de la predicción pueden repetir fila por adjudicatario/versión: una por (licitación, lote), la de menor id
con.execute("""CREATE OR REPLACE TEMP TABLE x_o2 AS SELECT * FROM (
    SELECT *, row_number() OVER (PARTITION BY conjunto, licitacion_id, lote_u ORDER BY fecha_pub, pbl) rn FROM x_o) WHERE rn = 1""")
con.execute("ALTER TABLE x_o2 DROP COLUMN rn")
con.execute("CREATE OR REPLACE TEMP TABLE x_o3 AS SELECT row_number() OVER (ORDER BY conjunto, licitacion_id, lote_u)::BIGINT oid, * FROM x_o2")
print("objetos", inf["x_o"], round(time.time() - T0), "s", flush=True)

# --- ventanas por nivel: una tabla por nivel con (claves, T) y n/k por desenlace --------------------------------------
sel_cols = ["x_o3.oid"]
for nom, (claves, v) in NIVELES.items():
    k_o = ", ".join(f"{c}" for c in claves)
    # claves de los objetos (mismas transformaciones que x_h)
    mapa = {"cpv_cel": "cpv_cel", "proc_nat": "proc_nat", "ccaa_c": "coalesce(ccaa, '??')", "toc": "coalesce(tipo_organo, '??')",
            "organo_clave": "organo_clave"}
    sel = ", ".join(f"{mapa[c]} AS {c}" for c in claves)
    distintos = f"SELECT DISTINCT {sel + ', ' if sel else ''}T FROM x_o3" + (" WHERE organo_clave IS NOT NULL" if "organo_clave" in claves else "")
    join_k = " AND ".join([f"h.{c} IS NOT DISTINCT FROM d.{c}" for c in claves] or ["TRUE"])
    aggs = []
    for h, col, s in (("ou", "anio_ref", 0), ("des", "anio_ref", 0), ("mod", "anio_adj", 1)):
        w = ventana(v, f"h.{col}", s)
        aggs.append(f"coalesce(sum(h.{h}_n) FILTER (WHERE {w}), 0)::BIGINT AS {nom}_{h}_n")
        aggs.append(f"coalesce(sum(h.{h}_k) FILTER (WHERE {w}), 0)::BIGINT AS {nom}_{h}_k")
    gb = ", ".join([f"d.{c}" for c in claves] + ["d.T"])
    con.execute(f"""CREATE OR REPLACE TEMP TABLE w_{nom} AS
        SELECT {gb.replace('d.', 'd.')}, {', '.join(aggs)}
        FROM ({distintos}) d
        LEFT JOIN (SELECT {(k_o + ', ') if k_o else ''}anio_ref, anio_adj, sum(ou_n) ou_n, sum(ou_k) ou_k, sum(des_n) des_n,
                          sum(des_k) des_k, sum(mod_n) mod_n, sum(mod_k) mod_k FROM x_h GROUP BY ALL) h
            ON {join_k} AND h.anio_ref <= d.T
        GROUP BY {gb}""")
    print("ventana", nom, con.execute(f"SELECT count(*) FROM w_{nom}").fetchone()[0], round(time.time() - T0), "s", flush=True)

# NSP del órgano (filas de adjudicación del feed propio, todos los procedimientos; T-3…T-1), como E1
con.execute("""CREATE OR REPLACE TEMP TABLE x_nsp AS
    SELECT o.organo_clave, u.anio_ref, count(*) n_adj, count(*) FILTER (WHERE u.procedimiento = 'Negociado_sin_publicidad') n_nsp
    FROM x_u u JOIN x_of o ON o.organo_nif IS NOT DISTINCT FROM u.organo_nif AND o.organo_nombre IS NOT DISTINCT FROM u.organo_nombre
        AND o.dir3 IS NOT DISTINCT FROM u.dir3
    WHERE u.source = 'PLACSP' AND u.vigente AND u.es_contrato AND NOT u.es_menor AND u.tabla_origen = 'resultados_licitaciones'
        AND u.unidad = 'adjudicacion' AND u.anio_ref BETWEEN 2012 AND 2026 AND o.organo_clave IS NOT NULL
    GROUP BY 1, 2""")
con.execute("""CREATE OR REPLACE TEMP TABLE w_nsp AS
    SELECT d.organo_clave, d.T, coalesce(sum(h.n_adj), 0)::BIGINT nsp_adj, coalesce(sum(h.n_nsp), 0)::BIGINT nsp_n
    FROM (SELECT DISTINCT organo_clave, T FROM x_o3 WHERE organo_clave IS NOT NULL) d
    LEFT JOIN x_nsp h ON h.organo_clave = d.organo_clave AND h.anio_ref BETWEEN d.T - 3 AND d.T - 1 GROUP BY 1, 2""")

# --- tabla final ---------------------------------------------------------------------------------------------------
joins = []
cols_w = []
for nom, (claves, v) in NIVELES.items():
    mapa = {"cpv_cel": "o.cpv_cel", "proc_nat": "o.proc_nat", "ccaa_c": "coalesce(o.ccaa, '??')",
            "toc": "coalesce(o.tipo_organo, '??')", "organo_clave": "o.organo_clave"}
    cond = " AND ".join([f"w_{nom}.{c} IS NOT DISTINCT FROM {mapa[c]}" for c in claves] + [f"w_{nom}.T = o.T"])
    joins.append(f"LEFT JOIN w_{nom} ON {cond}")
    for h in ("ou", "des", "mod"):
        cols_w += [f"coalesce(w_{nom}.{nom}_{h}_n, 0) {nom}_{h}_n", f"coalesce(w_{nom}.{nom}_{h}_k, 0) {nom}_{h}_k"]
con.execute(f"""CREATE OR REPLACE TEMP TABLE x_final AS
    SELECT o.*, coalesce(na.n_lotes_an, 1) n_lotes_an, c.n_crit, c.suman100, c.n_sin_tipo, c.peso_jv, c.peso_suma, c.peso_precio,
        d.pcap, d.ppt, coalesce(ns.nsp_adj, 0) nsp_adj, coalesce(ns.nsp_n, 0) nsp_n, {', '.join(cols_w)}
    FROM x_o3 o
    LEFT JOIN x_nla na ON na.licitacion_id = o.licitacion_id
    LEFT JOIN x_crit c ON c.licitacion_id = o.licitacion_id
    LEFT JOIN x_docs d ON d.licitacion_id = o.licitacion_id
    LEFT JOIN w_nsp ns ON ns.organo_clave = o.organo_clave AND ns.T = o.T
    {' '.join(joins)}""")
n1, n2 = con.execute("SELECT count(*), count(DISTINCT oid) FROM x_final").fetchone()
assert n1 == n2, f"x_final multiplica filas: {n1} {n2}"
# solo lo que usa MP1: anuncios 2018-01-01…2026-04-30 (históricos) y la predicción
con.execute("""CREATE OR REPLACE TEMP TABLE x_sal AS SELECT * FROM x_final
               WHERE conjunto = 'pred' OR fecha_pub BETWEEN DATE '2018-01-01' AND DATE '2026-04-30'""")
inf["salida"] = con.execute("SELECT conjunto, count(*) FROM x_sal GROUP BY 1 ORDER BY 1").fetchall()
nombre = f"lotes_perturbado_{PANIO}.csv" if PERT else "lotes.csv"
con.execute(f"COPY (SELECT * FROM x_sal ORDER BY oid) TO '{SAL / nombre}' (HEADER, DELIMITER ',')")
# historial por celda para el empirical Bayes (los m se estiman sobre las celdas-año de entrenamiento): se reconstruye en el
# worker desde las ventanas de cada lote (las mismas), así que no hace falta otra tabla.
inf["segundos"] = round(time.time() - T0, 1)
inf["ru_maxrss_mb"] = round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024)
json.dump(inf, open(SAL / (f"informe_10_perturbado_{PANIO}.json" if PERT else "informe_10.json"), "w"), indent=1, default=str)
print(json.dumps(inf, default=str), flush=True)