# Universo del sobre S0004-25-C2 (capa 2, ofertas reales de Cantabria) según config/S0004-25-C2.md. Sin red. No lee ninguna lista
# de empresas ni el primer sobre. Entradas (solo lectura): colusion/cant1/interno/{screens_impl1,contratos_cob,casado}.parquet.
# Salidas: salidas/universos/S0004-25-C2.parquet (ids md5 + hechos) e interno/universos_mapa_S0004-25-C2.parquet (vuelta a los
# contratos reales; sin empresas).
import hashlib, os, collections, json
import pyarrow as pa, pyarrow.parquet as pq
CANT = '/a/colusion/cant1/interno'
md5 = lambda s: hashlib.md5(s.encode('utf-8')).hexdigest()
P = [r for r in pq.read_table(f'{CANT}/screens_impl1.parquet').to_pylist() if r['conjunto'] == 'P' and r['mercado'].startswith('Obras|')]
C = {r['id_contrato']: r for r in pq.read_table(f'{CANT}/contratos_cob.parquet').to_pylist()}
CAS = collections.defaultdict(set)
for r in pq.read_table(f'{CANT}/casado.parquet').to_pylist():
    if r['nivel'] in ('E1', 'E2') and r['source'] == 'PLACSP': CAS[r['id_contrato']].add(r['licitacion_id'])
# percentil de S dentro de su mercado (todos los contratos de P del mercado), rango medio, empates a 9 decimales
pct = {}
por_m = collections.defaultdict(list)
for r in P:
    if r['S'] is not None: por_m[r['mercado']].append(r)
for m, rs in por_m.items():
    vals = sorted((round(r['S'], 9), r['id_contrato']) for r in rs); N = len(vals); i = 0
    while i < N:
        j = i
        while j + 1 < N and vals[j + 1][0] == vals[i][0]: j += 1
        for k in range(i, j + 1): pct[vals[k][1]] = ((i + 1 + j + 1) / 2 - 0.5) / N
        i = j + 1
filas, mapa = [], []
for r in P:
    c = C[r['id_contrato']]
    ff = c['fecha_form']; anio = ff.year if ff else c['anio']
    lic = min(CAS[r['id_contrato']]) if CAS[r['id_contrato']] else None
    id_lote = md5('CANTABRIA_API|' + r['id_contrato'])
    id_lic = md5('PLACSP|' + lic) if lic else md5('CANTABRIA_API|' + r['id_contrato'])
    en_periodo = bool(ff and str(ff) <= '2025-03-31' and str(ff) >= '2019-01-01')
    filas.append(dict(id_lote=id_lote, id_lic=id_lic, id_organo=md5(c['organo'] or ''), anio=anio,
                      en_ambito=bool(r['ambito_sobre']), en_periodo=en_periodo, es_menor=(c['bloque'] == 'menor'),
                      c2_pct_s=pct.get(r['id_contrato']), c2_s=r['S'],
                      c2_pct_cv=r['pct_CV'], c2_pct_diffp=r['pct_DIFFP'], c2_pct_rd=r['pct_RD'], c2_pct_skew=r['pct_SKEW'],
                      c2_pct_kurto=r['pct_KURTO'], c2_cv=r['CV'], c2_n=r['n'],
                      c2_en_periodo=int(en_periodo), c2_es_menor=int(c['bloque'] == 'menor'), c2_casado=int(bool(lic))))
    mapa.append(dict(exp='S0004-25-C2', id_lote=id_lote, id_lic=id_lic, id_contrato=r['id_contrato'], expediente=c['expediente'],
                     licitacion_id_placsp=lic, licitaciones_placsp=sorted(CAS[r['id_contrato']]), organo=c['organo'],
                     consejeria=c['consejeria'], fecha_form=str(ff) if ff else None, anio=anio, bloque=c['bloque']))
filas.sort(key=lambda x: x['id_lote']); mapa.sort(key=lambda x: x['id_lote'])
pu = '/s/salidas/universos/S0004-25-C2.parquet'; pm = '/s/interno/universos_mapa_S0004-25-C2.parquet'
for p in (pu, pm):
    if os.path.exists(p): raise SystemExit(f'ya existe {p}')
pq.write_table(pa.Table.from_pylist(filas), pu); pq.write_table(pa.Table.from_pylist(mapa), pm)
amb = [f for f in filas if f['en_ambito']]
print(json.dumps(dict(filas=len(filas), en_ambito=len(amb), licitaciones_ambito=len({f['id_lic'] for f in amb}),
                      en_periodo=sum(f['en_periodo'] for f in amb), menores=sum(f['es_menor'] for f in amb),
                      casados=sum(f['c2_casado'] for f in amb), organos=len({f['id_organo'] for f in amb}),
                      sin_pct=sum(1 for f in amb if f['c2_pct_s'] is None),
                      mercados={m: len(v) for m, v in por_m.items()}), ensure_ascii=False))