"""SOB1 · 10 · Universos de los expedientes CNMC abiertos (sin casar empresas).
Unidad = lote adjudicado (source, licitacion_id, coalesce(lote_id, licitacion_id)), universo NOT es_copia
(copia_de OR copia_otra_fuente OR NOT cuenta en vínculos), unified_20261018. Hechos ya medidos unidos por id_canonical:
OU1 (oferta_unica/interno/lotes_p), VC1 (versiones/interno/lotes_p), RS1 (lotes hermanos + celda evaluable),
D1 capa 1 (colusion/interno/sc_impl1: baja, baja_cond, pct_s3, pnc...), B3 (piloto A de actas: pertenencia y screens).
Salidas: /s/salidas/universos/<exp>.parquet (ids md5 y hechos, sin empresas) · /s/interno/universos_mapa.parquet (ids reales)
· /s/salidas/10_universos.json (recuentos)."""
import duckdb, json, time
t0 = time.time()
con = duckdb.connect()
con.execute("SET memory_limit='6GB'; SET threads=4; SET temp_directory='/s/tmp/duck'; SET preserve_insertion_order=false")
D = '/d/unified_20261018'; I = D + '.identidad/'; A = '/a'
con.execute(f"CREATE VIEW u AS SELECT * FROM read_parquet('{D}/*/*/*.parquet', hive_partitioning=1, union_by_name=1)")
log = lambda *a: print(f'[{time.time()-t0:6.0f}s]', *a, flush=True)

# --- definiciones por expediente (mercado, territorio, órganos citados, periodo supuesto) ---------------------------
CANT = "(ccaa='CB' OR nuts LIKE 'ES13%' OR lower(strip_accents(coalesce(provincia,''))) LIKE '%cantabria%')"
CLM = ("(ccaa='CM' OR nuts LIKE 'ES42%' OR lower(strip_accents(coalesce(provincia,''))) IN "
       "('albacete','ciudad real','cuenca','guadalajara','toledo'))")
OBJ = "lower(strip_accents(coalesce(objeto,'')))"
ORG = "lower(strip_accents(coalesce(organo_nombre,'')))"
AENA = f"(regexp_matches({ORG}, '(^|[^a-z])aena([^a-z]|$)') OR {ORG} LIKE '%aeropuertos espanoles%')"
EXP = {
 'S0004-25': dict(
    nombre='Obra civil, edificación y servicios de construcción · Cantabria',
    mercado=f"(tipo_contrato='Obras' OR cpv_division='45')", nucleo="TRUE",
    ampliado=f"(cpv_division IN ('45','71','50') OR tipo_contrato='Obras')",
    territorio=CANT, periodo=('2019-01-01', '2025-03-31'),
    organos=f"(tipo_organo IN ('autonomica','ayuntamiento','local_supramunicipal') OR {ORG} LIKE '%santander%')"),
 'S0015-25': dict(
    nombre='Consultoría y asistencia técnica de ingeniería · España',
    mercado="(cpv_division='71')", nucleo="(cpv_code LIKE '713%' OR cpv_code LIKE '7152%' OR cpv_code LIKE '7124%')",
    ampliado="(cpv_division='71')", territorio="TRUE", periodo=('2018-01-01', '2024-11-28'),
    organos=f"(tipo_organo='age' AND {ORG} LIKE '%general de carreteras%')"),
 'S0017-25': dict(
    nombre='Transporte escolar y discrecional de viajeros · Castilla-La Mancha',
    mercado=f"(cpv_code LIKE '6013%' OR cpv_code LIKE '6011%' OR cpv_code LIKE '6014%' OR cpv_code LIKE '6017%' OR cpv_code LIKE '6010%' OR {OBJ} LIKE '%transporte escolar%')",
    nucleo=f"(cpv_code LIKE '6013%' OR {OBJ} LIKE '%escolar%')", ampliado=f"(cpv_division='60' OR {OBJ} LIKE '%transporte escolar%')",
    territorio=CLM, periodo=('2018-01-01', '2025-10-08'),
    organos=f"(tipo_organo='autonomica' AND {ORG} LIKE '%educacion%')"),
 'S0014-25': dict(
    nombre='Instalaciones FBO (aviación general y ejecutiva) en aeropuertos · Aena',
    mercado=f"({AENA} AND ({OBJ} LIKE '%fbo%' OR {OBJ} LIKE '%base fija%' OR {OBJ} LIKE '%bases fijas%' OR {OBJ} LIKE '%aviacion general%' OR {OBJ} LIKE '%aviacion ejecutiva%') AND ({OBJ} LIKE '%arrend%' OR {OBJ} LIKE '%licencia%' OR {OBJ} LIKE '%explotacion%' OR {OBJ} LIKE '%concesion%' OR {OBJ} LIKE '%gestion de%' OR {OBJ} LIKE '%prestacion de servicios aeroportuarios%' OR {OBJ} LIKE '%handling%'))",
    nucleo="TRUE", ampliado=f"({AENA})", territorio="TRUE",
    periodo=('2014-01-01', '2025-06-13'), organos="TRUE"),
 'S0010-23': dict(
    nombre='Tracción ferroviaria de mercancías (licitación de ene-2022) · Pecovasa',
    mercado=f"({ORG} LIKE '%pecovasa%' AND {OBJ} LIKE '%traccion%')", nucleo="TRUE",
    ampliado=f"({ORG} LIKE '%pecovasa%' OR {OBJ} LIKE '%traccion%ferrovi%')", territorio="TRUE",
    periodo=('2021-07-01', '2022-12-31'), organos=f"({ORG} LIKE '%pecovasa%')"),
 'V-AEREOS-26': dict(
    nombre='[vigilancia, sin incoar] Trabajos aéreos: incendios, transporte sanitario, salvamento',
    mercado=f"(cpv_code LIKE '6044%' OR cpv_code LIKE '6042%' OR cpv_code LIKE '6045%' OR (({OBJ} LIKE '%aere%' OR {OBJ} LIKE '%helicoptero%' OR {OBJ} LIKE '%aeronave%' OR {OBJ} LIKE '%avion%') AND ({OBJ} LIKE '%incendio%' OR {OBJ} LIKE '%salvamento%' OR {OBJ} LIKE '%sanitari%' OR {OBJ} LIKE '%rescate%' OR {OBJ} LIKE '%emergencia%')))",
    nucleo=f"({OBJ} LIKE '%incendio%' OR {OBJ} LIKE '%salvamento%' OR {OBJ} LIKE '%sanitari%' OR {OBJ} LIKE '%rescate%')",
    ampliado="(cpv_division='60' AND cpv_code LIKE '604%')", territorio="TRUE", periodo=('2018-01-01', '2026-01-30'),
    organos="(tipo_organo IN ('autonomica','age','salud'))"),
 'V-SUEROS-26': dict(
    nombre='[vigilancia, sin incoar] Sueros hospitalarios y nutrición parenteral',
    mercado=f"(cpv_code LIKE '33692%' OR {OBJ} LIKE '%suero%' OR {OBJ} LIKE '%parenteral%' OR {OBJ} LIKE '%fluidoterapia%')",
    nucleo=f"({OBJ} LIKE '%suero%' OR {OBJ} LIKE '%parenteral%' OR {OBJ} LIKE '%fluidoterapia%' OR cpv_code LIKE '336925%' OR cpv_code LIKE '336922%')",
    ampliado="(cpv_code LIKE '3369%')", territorio="TRUE", periodo=('2018-01-01', '2026-06-12'),
    organos="(tipo_organo IN ('salud','autonomica'))"),
}

# --- base: filas adjudicadas que caen en algún mercado ampliado ----------------------------------------------------
amp = " OR ".join(f"(({e['ampliado']}) AND ({e['territorio']}))" for e in EXP.values())
# tipo_organo no existe aún en la base; los filtros de mercado/territorio no lo usan
con.execute(f"""CREATE TEMP TABLE b0 AS SELECT source, id_canonical, coalesce(licitacion_id, source||':'||id_canonical) licitacion_id,
  coalesce(lote_id, licitacion_id, id_canonical) lote_u, coalesce(regexp_extract(lote_id, '\\|([^|]*)$', 1), '') lote_k, unidad, anio_ref, cpv_code, cpv_division, tipo_contrato, procedimiento, es_menor,
  num_ofertas, imp_adj_eur_comparable imp, imp_lic_eur, organo_nif, organo_nombre, dir3, ccaa, nuts, provincia, objeto,
  coalesce(fecha_adj, fecha_formalizacion, fecha_pub, make_date(anio_ref::INT, 7, 1)) fref, copia_de, correccion_imp_adj
FROM u WHERE unidad='adjudicacion' AND anio_ref BETWEEN 2014 AND 2026 AND ({amp})""")
log('b0', con.execute("SELECT count(*) FROM b0").fetchone())
con.execute(f"""CREATE TEMP TABLE b AS SELECT b0.*, (b0.copia_de IS NOT NULL OR coalesce(b0.correccion_imp_adj='copia_otra_fuente', FALSE)
   OR coalesce(NOT v.cuenta, FALSE)) es_copia, o.organo_clave, coalesce(t.tipo,'sin_tipo') tipo_organo
FROM b0 LEFT JOIN read_parquet('{D}.vinculos/filas.parquet') v ON v.source=b0.source AND v.id_canonical=b0.id_canonical
LEFT JOIN (SELECT organo_nif, organo_nombre, dir3, any_value(organo_clave) organo_clave FROM read_parquet('{I}organos_filas.parquet') GROUP BY ALL) o
  ON o.organo_nif IS NOT DISTINCT FROM b0.organo_nif AND o.organo_nombre IS NOT DISTINCT FROM b0.organo_nombre AND o.dir3 IS NOT DISTINCT FROM b0.dir3
LEFT JOIN read_parquet('{I}organos_tipo.parquet') t ON t.organo_clave=o.organo_clave""")
log('b', con.execute("SELECT count(*), sum(es_copia::INT) FROM b").fetchone())

# --- hechos por fila (id_canonical) -------------------------------------------------------------------------------
con.execute(f"""CREATE TEMP TABLE h_ou AS SELECT id_min id_canonical, unica ou1_unica, p ou1_p FROM read_parquet('{A}/oferta_unica/interno/lotes_p.parquet')""")
con.execute(f"""CREATE TEMP TABLE h_vc AS SELECT licitacion_id, lote_k, bool_or(fondo) vc1_fondo, max(tipo_cambio) vc1_tipo,
  max(subtipo) vc1_subtipo, max(p) vc1_p FROM read_parquet('{A}/versiones/interno/lotes_p.parquet') GROUP BY 1, 2""")
con.execute(f"""CREATE TEMP TABLE h_rs AS SELECT DISTINCT source, id_canonical, TRUE rs1_hermana FROM read_parquet('{A}/redes_societarias/interno/40_lotes_hermanos.parquet')""")
con.execute(f"""CREATE TEMP TABLE h_rc AS SELECT DISTINCT organo_clave, cpv2, anio FROM read_parquet('{A}/redes_societarias/interno/40_celda.parquet')""")
con.execute(f"""CREATE TEMP TABLE h_d1 AS SELECT id_canonical, en_c d1_en_c, pnc d1_pnc, rango d1_rango, sep d1_sep, baja d1_baja,
  baja_cond d1_baja_cond, pct_s3 d1_pct_s3, excluido d1_excluido FROM read_parquet('{A}/colusion/interno/sc_impl1.parquet')""")
con.execute(f"""CREATE TEMP TABLE h_b3 AS SELECT licitacion_id, max(CV) b3_cv, max(DIFFP) b3_diffp, max(n_validas) b3_n_validas
  FROM read_csv('{A}/actas/piloto_a/privado/screens_v1/lotes_validos.tsv', delim='\t', header=true, all_varchar=true,
       types={{'CV':'DOUBLE','DIFFP':'DOUBLE','n_validas':'INTEGER'}}) GROUP BY 1""")
con.execute(f"""CREATE TEMP TABLE h_b3l AS SELECT DISTINCT licitacion_id FROM read_csv('{A}/actas/piloto_a/privado/clave/lotes.tsv', delim='\t', header=true, all_varchar=true)""")

res = {'generado': time.strftime('%Y-%m-%d %H:%M:%S'), 'expedientes': {}}
mapas = []
for exp, e in EXP.items():
    vi, vf = e['periodo']
    con.execute(f"""CREATE OR REPLACE TEMP TABLE r AS SELECT *, ({e['mercado']}) en_mercado, ({e['nucleo']}) AND ({e['mercado']}) en_nucleo,
        ({e['organos']}) en_organos, fref BETWEEN DATE '{vi}' AND DATE '{vf}' en_periodo
      FROM b WHERE NOT es_copia AND ({e['ampliado']}) AND ({e['territorio']})""")
    con.execute(f"""CREATE OR REPLACE TEMP TABLE lot AS
      SELECT '{exp}' exp, md5(r.source||'|'||r.licitacion_id||'|'||r.lote_u) id_lote, md5(r.source||'|'||r.licitacion_id) id_lic,
        r.source, any_value(r.anio_ref)::INT anio, min(r.fref) fref, any_value(r.cpv_division) cpv_div, any_value(substr(r.cpv_code,1,4)) cpv4,
        any_value(r.tipo_contrato) tipo_contrato, any_value(r.procedimiento) procedimiento, bool_or(r.es_menor) es_menor,
        any_value(r.tipo_organo) tipo_organo, md5(coalesce(any_value(r.organo_clave),'')) id_organo,
        bool_or(r.en_mercado) en_mercado, bool_or(r.en_nucleo) en_nucleo, bool_or(r.en_organos) en_organos, bool_or(r.en_periodo) en_periodo,
        max(r.num_ofertas) num_ofertas, sum(r.imp) imp_adj, max(r.imp_lic_eur) imp_lic, count(*) filas,
        bool_or(ou.id_canonical IS NOT NULL) ou1_evaluable, bool_or(ou.ou1_unica) ou1_unica, max(ou.ou1_p) ou1_p,
        bool_or(vc.licitacion_id IS NOT NULL) vc1_evaluable, bool_or(vc.vc1_fondo) vc1_fondo, max(vc.vc1_tipo) vc1_tipo, max(vc.vc1_subtipo) vc1_subtipo, max(vc.vc1_p) vc1_p,
        bool_or(rc.organo_clave IS NOT NULL) rs1_celda_evaluable,
        CASE WHEN bool_or(rc.organo_clave IS NOT NULL OR rs.rs1_hermana IS NOT NULL) THEN bool_or(coalesce(rs.rs1_hermana, FALSE)) END rs1_hermana,
        bool_or(coalesce(d1.d1_en_c, FALSE)) d1_en_c, max(d1.d1_pnc) d1_pnc, max(d1.d1_rango) d1_rango, max(d1.d1_sep) d1_sep,
        max(d1.d1_baja) d1_baja, max(d1.d1_baja_cond) d1_baja_cond, max(d1.d1_pct_s3) d1_pct_s3, bool_or(d1.id_canonical IS NOT NULL) d1_en_p,
        bool_or(b3l.licitacion_id IS NOT NULL) b3_en_piloto, max(b3.b3_cv) b3_cv, max(b3.b3_diffp) b3_diffp,
        min(r.id_canonical) id_min, any_value(r.licitacion_id) licitacion_id, any_value(r.lote_u) lote_u
      FROM r LEFT JOIN h_ou ou ON ou.id_canonical=r.id_canonical
        LEFT JOIN h_vc vc ON vc.licitacion_id=r.licitacion_id AND vc.lote_k=r.lote_k
        LEFT JOIN h_rs rs ON rs.source=r.source AND rs.id_canonical=r.id_canonical
        LEFT JOIN h_rc rc ON rc.organo_clave=r.organo_clave AND rc.cpv2=r.cpv_division AND rc.anio=r.anio_ref
        LEFT JOIN h_d1 d1 ON d1.id_canonical=r.id_canonical
        LEFT JOIN h_b3l b3l ON b3l.licitacion_id=r.licitacion_id LEFT JOIN h_b3 b3 ON b3.licitacion_id=r.licitacion_id
      GROUP BY r.source, r.licitacion_id, r.lote_u""")
    con.execute(f"COPY (SELECT * EXCLUDE (id_min, licitacion_id, lote_u) FROM lot ORDER BY id_lote) TO '/s/salidas/universos/{exp}.parquet' (FORMAT parquet)")
    con.execute(f"INSERT INTO mapa SELECT exp, id_lote, id_lic, source, licitacion_id, lote_u, id_min FROM lot" if mapas else
                f"CREATE TEMP TABLE mapa AS SELECT exp, id_lote, id_lic, source, licitacion_id, lote_u, id_min FROM lot")
    mapas.append(exp)
    q = lambda s: con.execute(s).fetchall()
    P = "en_mercado AND en_periodo"
    R = dict(nombre=e['nombre'], periodo=e['periodo'],
      lotes_ampliado_2014_2026=q("SELECT count(*) FROM lot")[0][0],
      principal=q(f"""SELECT count(*) lotes, count(DISTINCT id_lic) licitaciones, count(*) FILTER (WHERE NOT es_menor) no_menores,
          count(*) FILTER (WHERE en_nucleo) nucleo, count(*) FILTER (WHERE en_organos) organos_citados,
          count(*) FILTER (WHERE en_organos AND NOT es_menor) organos_citados_no_menores, count(DISTINCT id_organo) organos,
          round(sum(imp_adj)/1e6,1) imp_adj_meur,
          count(*) FILTER (WHERE source='PLACSP') placsp, count(*) FILTER (WHERE num_ofertas IS NOT NULL) con_nof,
          count(*) FILTER (WHERE ou1_evaluable) ou1_ev, count(*) FILTER (WHERE ou1_unica) ou1_unica,
          count(*) FILTER (WHERE vc1_evaluable) vc1_ev, count(*) FILTER (WHERE vc1_fondo) vc1_fondo,
          count(*) FILTER (WHERE rs1_celda_evaluable) rs1_ev, count(*) FILTER (WHERE rs1_hermana) rs1_hermana,
          count(*) FILTER (WHERE d1_en_p) d1_en_p, count(*) FILTER (WHERE d1_en_c) d1_en_c, count(*) FILTER (WHERE d1_baja IS NOT NULL) d1_baja,
          count(*) FILTER (WHERE d1_en_c AND d1_pct_s3>=0.8) d1_s3_ge80, count(*) FILTER (WHERE b3_en_piloto) b3_piloto
        FROM lot WHERE {P}""")[0],
      por_anio=q(f"SELECT anio, count(*), count(*) FILTER (WHERE en_organos), count(*) FILTER (WHERE ou1_evaluable), count(*) FILTER (WHERE d1_en_c) FROM lot WHERE {P} GROUP BY 1 ORDER BY 1"),
      por_fuente=q(f"SELECT source, count(*) FROM lot WHERE {P} GROUP BY 1 ORDER BY 2 DESC"),
      por_tipo_organo=q(f"SELECT tipo_organo, count(*) FROM lot WHERE {P} GROUP BY 1 ORDER BY 2 DESC"),
      por_cpv4=q(f"SELECT cpv4, count(*) FROM lot WHERE {P} GROUP BY 1 ORDER BY 2 DESC LIMIT 8"))
    cols = ['lotes','licitaciones','no_menores','nucleo','organos_citados','organos_citados_no_menores','organos','imp_adj_meur','placsp','con_nof',
            'ou1_ev','ou1_unica','vc1_ev','vc1_fondo','rs1_ev','rs1_hermana','d1_en_p','d1_en_c','d1_baja','d1_s3_ge80','b3_piloto']
    R['principal'] = dict(zip(cols, R['principal']))
    res['expedientes'][exp] = R
    log(exp, json.dumps(R['principal']))
con.execute("COPY mapa TO '/s/interno/universos_mapa.parquet' (FORMAT parquet)")
# nombres de órganos citados (son órganos públicos, no empresas): para cotejar los filtros
for exp in ['S0015-25', 'S0017-25', 'S0014-25', 'S0010-23']:
    e = EXP[exp]
    res['expedientes'][exp]['organos_top'] = con.execute(f"""SELECT organo_nombre, count(*) FROM b WHERE NOT es_copia AND ({e['ampliado']}) AND ({e['territorio']})
       AND ({e['organos']}) AND ({e['mercado']}) GROUP BY 1 ORDER BY 2 DESC LIMIT 8""").fetchall()
json.dump(res, open('/s/salidas/10_universos.json', 'w'), indent=1, ensure_ascii=False, default=str)
log('fin')