"""A1 · SEGUNDA IMPLEMENTACIÓN (pyarrow + Python puro, sin DuckDB) del cálculo sellado, escrita aparte de e2_hechos.py.
Recalcula desde las entradas crudas el universo, la cantidad, el estado, la referencia, la razón, válido, el intervalo
dejando fuera al órgano, K de N, Holm y el hecho; compara con priv/30_contratos_calculo.csv y salidas/31_hechos_organo.csv.
Criterio sellado: diferencias ≤ 0,5 % de los contratos (estado, válido, razón a 1e-9 relativo) y ≤ 0,5 % de las unidades
órgano × producto (n, K, hecho). Salida: salidas/34_impl2.json."""
import pyarrow.parquet as pq, csv, json, math, os, re, datetime as dt
P = '/p'
T = pq.read_table(f'{P}/priv/a1_base.parquet').to_pylist()
ref = {(r['mes'], r['producto']): float(r['ref_nacional_sin_iva_eur_l']) for r in csv.DictReader(open(f'{P}/salidas/15_ref_mensual.csv'))}
fac = {(r['prov_ine'], r['producto'], int(r['anio']), int(r['trimestre'])): float(r['factor']) for r in csv.DictReader(open(f'{P}/salidas/15_factor_trimestral.csv'))}
lec = {}
LF = f'{P}/priv/lectura_final.csv'
if os.path.exists(LF) and os.path.getsize(LF) > 0:
    for r in csv.DictReader(open(LF)): lec[(r['source'], r['id_canonical'])] = r
def fnum(x):
    try: return float(x)
    except (TypeError, ValueError): return None
OK_P = {'gasoleo_a', 'gasoleo_b', 'gasoleo_c', 'gasolina95'}
C = {}
for a in T:
    if a['mercado'] != 'carburantes' or a['es_copia'] or a['excluido_v2'] or not (a['imp'] or 0) > 0: continue
    if (a['rol_am'] or '') in ('acuerdo', 'sda') or (a['prov_ine'] or '') in ('35', '38', '51', '52'): continue
    if (a['nuts'] or '')[:4] in ('ES70', 'ES63', 'ES64'): continue
    f0 = a['fecha_adj'] or a['fecha_formalizacion'] or a['fecha_pub']
    mes = f0.strftime('%Y-%m') if f0 and dt.date(2022, 1, 1) <= f0 <= dt.date(2025, 12, 31) else f"{a['anio_ref']}-07"
    l = lec.get((a['source'], a['id_canonical'])); lc = fnum(l['cantidad']) if l else None
    via = 'L' if lc is not None else ('R' if a['cant_v2'] is not None else None)
    prod = l['producto'] if via == 'L' else a['producto_v2']
    dur = a['duracion_meses'] if a['duracion_meses'] is not None else (fnum(l['duracion_leida']) if l else None)
    tot = None
    if via == 'R': tot = a['cant_v2']
    elif via == 'L' and l['unidad'] == 'litros':
        per = l['periodo']
        if per == 'total_contrato': tot = lc
        elif per == 'anual' and dur and dur > 0: tot = lc * dur / 12
        elif per == 'mensual' and dur and dur > 0: tot = lc * dur
    if via is None: est = 'sin_cantidad'
    elif prod not in OK_P: est = 'sin_producto_comparable'
    elif via == 'L' and l['unidad'] != 'litros': est = 'unidad_no_litros'
    elif via == 'L' and l['importe_tipo'] == 'descuento_surtidor': est = 'descuento_surtidor'
    elif via == 'L' and l['periodo'] not in ('total_contrato', 'anual', 'mensual'): est = 'periodo_desconocido'
    elif via == 'L' and l['periodo'] in ('anual', 'mensual') and not (dur and dur > 0): est = 'sin_duracion'
    else: est = 'comparable'
    razon = None
    if tot and prod in OK_P and (mes, prod) in ref:
        y, mm = int(mes[:4]), int(mes[5:7]); fa = fac.get((a['prov_ine'], prod, y, (mm - 1) // 3 + 1)) if prod in ('gasoleo_a', 'gasolina95') else None
        razon = a['imp'] / tot / (ref[(mes, prod)] * (fa if fa is not None else 1.0))
    nuc = est == 'comparable' and tot is not None and tot >= 200 and (via == 'L' or a['n_cant_v2'] == 1)
    C[(a['source'], a['id_canonical'])] = dict(org=a['organo_clave'], prod=prod, est=est, razon=razon, nuc=nuc,
                                                val=bool(nuc and razon is not None and 0.5 <= razon <= 2.0))
# comparar contratos
dif = 0; n1 = 0; vistos = set()
for r in csv.DictReader(open(f'{P}/priv/30_contratos_calculo.csv')):
    k = (r['source'], r['id_canonical']); n1 += 1; vistos.add(k); c = C.get(k)
    if c is None: dif += 1; continue
    r1 = fnum(r['razon']); v1 = r['valido'] == 'true'
    if r['estado'] != c['est'] or v1 != c['val'] or ((r1 is None) != (c['razon'] is None)) or (r1 is not None and abs(r1 - c['razon']) > 1e-9 * abs(r1)): dif += 1
dif += len(set(C) - vistos)
# unidades
V = {}
for c in C.values():
    if c['val'] and c['org'] is not None: V.setdefault(c['prod'], {}).setdefault(c['org'], []).append(c['razon'])
def q7(xs, p):
    xs = sorted(xs); h = (len(xs) - 1) * p; lo = math.floor(h); hi = min(lo + 1, len(xs) - 1); return xs[lo] + (h - lo) * (xs[hi] - xs[lo])
def cola(n, k, p=0.10):
    return sum(math.exp(math.lgamma(n + 1) - math.lgamma(i + 1) - math.lgamma(n - i + 1) + i * math.log(p) + (n - i) * math.log(1 - p)) for i in range(k, n + 1))
U = []
for prod, por in V.items():
    for o, xs in por.items():
        otros = [x for oo, ys in por.items() if oo != o for x in ys]
        if not otros: continue
        a10, a90 = q7(otros, .1), q7(otros, .9)
        u = dict(o=o, p=prod, n=len(xs), ks=sum(x > a90 for x in xs), ki=sum(x < a10 for x in xs))
        if u['n'] >= 10: u['ps'], u['pi'] = cola(u['n'], u['ks']), cola(u['n'], u['ki'])
        U.append(u)
T_ = sorted([(u['ps'], i, 'sup') for i, u in enumerate(U) if 'ps' in u] + [(u['pi'], i, 'inf') for i, u in enumerate(U) if 'pi' in u])
R = set()
for j, (p, i, lado) in enumerate(T_):
    if p > 0.05 / (len(T_) - j): break
    R.add((i, lado))
for i, u in enumerate(U):
    u['hecho'] = 'no_evaluable' if u['n'] < 10 else 'por_encima' if (i, 'sup') in R else 'por_debajo' if (i, 'inf') in R else 'ninguno'
mapa = {(u['o'], u['p']): u for u in U}; du = 0; nu = 0
for r in csv.DictReader(open(f'{P}/salidas/31_hechos_organo.csv')):
    nu += 1; u = mapa.pop((r['organo_clave'], r['producto']), None)
    if u is None or int(r['n']) != u['n'] or int(r['k_sup']) != u['ks'] or int(r['k_inf']) != u['ki'] or r['hecho'] != u['hecho']: du += 1
du += len(mapa)
res = dict(contratos_impl1=n1, contratos_impl2=len(C), contratos_distintos=dif, pct_contratos=round(100 * dif / max(1, n1), 4),
           unidades_impl1=nu, unidades_impl2=len(U), unidades_distintas=du, pct_unidades=round(100 * du / max(1, nu), 4),
           hechos_impl2={h: sum(u['hecho'] == h for u in U) for h in ('por_encima', 'por_debajo', 'ninguno', 'no_evaluable')})
res['pasa'] = res['pct_contratos'] <= 0.5 and res['pct_unidades'] <= 0.5
json.dump(res, open(f'{P}/salidas/34_impl2.json', 'w'), indent=1); print(res)