"""MP1 · paso 12: población y etiqueta del desenlace (d), recurso estimado contra el pliego (pre-registro §8).
Imagen buscalicitaciones-etl:v2. Lee unified_20261025 (+ satélites, con el módulo de E1 para criterios, pliegos y CPV) y el
banco de tribunales (/banco = PLIEGOS/banco, /trib = TRIBUNALES), solo metadatos. Escribe /w/interno/recurso.csv y
/w/salidas/12_recurso.json.
"""
import sys, json, time, collections
from pathlib import Path
sys.path.insert(0, "/app")
from etl.v2 import exante_l1 as e1

T0 = time.time()
U = Path("/d/unified_20261025")
TMP = Path("/w/tmp/duck"); TMP.mkdir(parents=True, exist_ok=True)
con = e1.conexion(memoria="8GB", 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"))
inf = {}
procs = ", ".join(e1._lit(p) for p in e1.PROCEDIMIENTOS)
A = e1._attr
con.execute(f"""CREATE OR REPLACE TEMP TABLE r_l AS
    WITH f AS (SELECT * FROM x_u WHERE source = 'PLACSP' AND vigente AND es_contrato AND NOT es_menor AND procedimiento IN ({procs})
                 AND coalesce(rol_am, '') NOT IN ('basado', 'basado_sda') AND coalesce(tramitacion, '') <> 'emergencia'
                 AND tabla_origen IN ('resultados_licitaciones', 'resultados_agregacion', 'licitaciones_sin_resultado')
                 AND unidad IN ('adjudicacion', 'desierto', 'licitacion:sin_resultado'))
    SELECT licitacion_id, {A('tabla_origen')}, {A('estado')}, {A('fecha_pub')}, {A('fecha_fin_ofertas', 'ffo')}, {A('procedimiento_native', 'proc_nat')},
        {A('tramitacion', 'tram')}, {A('tipo_contrato')}, {A('cpv_division', 'cpv_div')}, {A('imp_lic_eur_expediente', 'pbl_exp')},
        {A('imp_lic_eur', 'pbl_lote')}, {A('valor_estimado_eur_expediente', 've_exp')}, {A('duracion_meses', 'dur')}, {A('sara')},
        {A('financiacion_ue', 'ue')}, {A('organo_nif')}, {A('organo_nombre')}, {A('dir3')}, {A('ccaa')},
        count(*) filas, bool_or(lote_id IS NOT NULL) con_lotes
    FROM f GROUP BY 1""")
con.execute("""CREATE OR REPLACE TEMP TABLE r_l2 AS
    SELECT l.*, o.organo_clave, t.tipo tipo_organo, cv.n_divisiones cpv_n_div, coalesce(na.n, 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
    FROM r_l l
    LEFT JOIN x_of o ON o.organo_nif IS NOT DISTINCT FROM l.organo_nif AND o.organo_nombre IS NOT DISTINCT FROM l.organo_nombre
        AND o.dir3 IS NOT DISTINCT FROM l.dir3
    LEFT JOIN x_ot t ON t.organo_clave = o.organo_clave
    LEFT JOIN x_cpv_exp cv ON cv.licitacion_id = l.licitacion_id
    LEFT JOIN (SELECT licitacion_id, count(DISTINCT lote) n FROM x_cpv_raw WHERE ambito = 'lote' AND lote IS NOT NULL GROUP BY 1) na
        ON na.licitacion_id = l.licitacion_id
    LEFT JOIN x_crit c ON c.licitacion_id = l.licitacion_id
    LEFT JOIN x_docs d ON d.licitacion_id = l.licitacion_id""")
inf["tipos_contrato"] = con.execute("SELECT tipo_contrato, count(*) FROM r_l2 WHERE ccaa IN ('AN','CT','AR','CN','CL') AND fecha_pub >= DATE '2025-01-01' GROUP BY 1 ORDER BY 2 DESC").fetchall()
# banco
con.execute("CREATE OR REPLACE TEMP TABLE cas AS SELECT * FROM read_json_auto('/banco/casado.jsonl')")
inf["banco_casa"] = dict(zip(["casadas", "en_unified_20261025", "en_r_l"], con.execute("""SELECT count(DISTINCT licitacion_id),
    count(DISTINCT licitacion_id) FILTER (WHERE licitacion_id IN (SELECT DISTINCT licitacion_id FROM x_u WHERE source = 'PLACSP')),
    count(DISTINCT licitacion_id) FILTER (WHERE licitacion_id IN (SELECT licitacion_id FROM r_l2))
    FROM cas WHERE estado = 'casada'""").fetchone()))
con.execute("""CREATE OR REPLACE TEMP TABLE pos AS
    SELECT licitacion_id, min(fecha::DATE) fecha_res, arg_min(tribunal, fecha) tribunal, arg_min(sentido, fecha) sentido, count(*) n_res
    FROM cas WHERE estado = 'casada' AND sentido IN ('estimacion', 'estimacion_parcial') GROUP BY 1""")
con.execute("""CREATE OR REPLACE TEMP TABLE neg_banco AS SELECT DISTINCT licitacion_id FROM cas WHERE estado = 'casada' AND sentido = 'desestimacion'""")
RECURRIBLE = """CASE WHEN tipo_contrato IN ('Servicios', 'Suministros') THEN coalesce(ve_exp, pbl_exp, pbl_lote) >= 100000
                     WHEN tipo_contrato IN ('Obras', 'Concesion_obras', 'Concesion_servicios', 'Gestion_servicios_publicos')
                          THEN coalesce(ve_exp, pbl_exp, pbl_lote) >= 3000000 ELSE FALSE END"""
con.execute(f"""CREATE OR REPLACE TEMP TABLE r_pob AS
    SELECT l.*, ({RECURRIBLE}) recurrible, (p.licitacion_id IS NOT NULL)::INTEGER y, p.tribunal, p.sentido, p.fecha_res,
        (nb.licitacion_id IS NOT NULL) desestimada_banco, year(l.fecha_pub)::INTEGER T,
        CASE WHEN l.ffo - l.fecha_pub > 0 AND l.ffo - l.fecha_pub <= 180 THEN l.ffo - l.fecha_pub END plazo
    FROM r_l2 l LEFT JOIN pos p USING (licitacion_id) LEFT JOIN neg_banco nb USING (licitacion_id)
    WHERE l.ccaa IN ('AN', 'CT', 'AR', 'CN', 'CL') AND coalesce(l.tipo_organo, '') <> 'age'
        AND l.fecha_pub BETWEEN DATE '2025-01-01' AND DATE '2026-04-30'""")
inf["embudo"] = con.execute("""SELECT count(*), count(*) FILTER (WHERE recurrible), count(*) FILTER (WHERE recurrible AND plazo IS NOT NULL),
    sum(y), sum(y) FILTER (WHERE recurrible), sum(y) FILTER (WHERE recurrible AND plazo IS NOT NULL) FROM r_pob""").fetchone()
inf["positivos_fuera_por_regla"] = con.execute("""SELECT tipo_contrato, recurrible, plazo IS NULL, count(*) FROM r_pob WHERE y = 1
    AND NOT (recurrible AND plazo IS NOT NULL) GROUP BY ALL ORDER BY 4 DESC""").fetchall()
con.execute("""COPY (SELECT * FROM r_pob WHERE recurrible AND plazo IS NOT NULL ORDER BY licitacion_id) TO '/w/interno/recurso.csv' (HEADER)""")
inf["por_ventana"] = con.execute("""SELECT CASE WHEN fecha_pub < DATE '2026-01-01' THEN 'train' ELSE 'test' END v, ccaa, count(*), sum(y)
    FROM r_pob WHERE recurrible AND plazo IS NOT NULL GROUP BY ALL ORDER BY 1, 2""").fetchall()
inf["coherencia_tribunal_ccaa"] = con.execute("""SELECT tribunal, ccaa, count(*) FROM r_pob WHERE y = 1 GROUP BY ALL ORDER BY 1, 2""").fetchall()
# fracciones de muestreo del banco: seleccionadas / indexadas por tribunal × sentido × año (acto pliegos)
idx = collections.Counter(); sel = collections.Counter()
for t in ["tarcja", "tccsp", "tacpa", "tacpc", "tarccyl"]:
    for line in open(f"/trib/{t}/indice.jsonl"):
        dd = json.loads(line)
        if dd.get("acto") == "pliegos":
            idx[(t, str(dd.get("sentido")), int(dd.get("anio") or 0))] += 1
for line in open("/trib/seleccion_pdfs.jsonl"):
    dd = json.loads(line)
    if dd.get("acto") == "pliegos":
        sel[(dd["tribunal"], str(dd.get("sentido")), int(dd.get("anio") or 0))] += 1
inf["fraccion_muestreo"] = {f"{k[0]}|{k[1]}|{k[2]}": [sel[k], v, (sel[k] / v if v else None)] for k, v in sorted(idx.items())
                            if k[1] in ("estimacion", "estimacion_parcial") and k[2] >= 2025}
# anuncios en plazo (para la predicción de producción y la maqueta): misma población, sin resultado, Publicada, fin de
# ofertas >= 25-oct-2026, cualquier fecha de publicación
con.execute(f"""COPY (
    SELECT l.*, ({RECURRIBLE}) recurrible, NULL::INTEGER y, NULL::VARCHAR tribunal, NULL::VARCHAR sentido, NULL::DATE fecha_res,
        FALSE desestimada_banco, year(l.fecha_pub)::INTEGER T,
        CASE WHEN l.ffo - l.fecha_pub > 0 AND l.ffo - l.fecha_pub <= 180 THEN l.ffo - l.fecha_pub END plazo
    FROM r_l2 l WHERE l.ccaa IN ('AN', 'CT', 'AR', 'CN', 'CL') AND coalesce(l.tipo_organo, '') <> 'age'
        AND l.tabla_origen = 'licitaciones_sin_resultado' AND l.estado = 'Publicada' AND l.ffo >= DATE '2026-10-25'
        AND ({RECURRIBLE}) AND l.ffo - l.fecha_pub > 0 AND l.ffo - l.fecha_pub <= 180 ORDER BY licitacion_id)
    TO '/w/interno/recurso_pred.csv' (HEADER)""")
inf["en_plazo"] = con.execute("SELECT count(*) FROM read_csv_auto('/w/interno/recurso_pred.csv')").fetchone()[0]
inf["segundos"] = round(time.time() - T0, 1)
json.dump(inf, open("/w/salidas/12_recurso.json", "w"), indent=1, default=str, ensure_ascii=False)
print(json.dumps(inf, default=str, ensure_ascii=False)[:4000])