"""E8 · base (implementación 1, DuckDB + reglas_e8): filas candidatas de los tres mercados, 2015-2026, con TODAS las columnas
crudas que necesitan las dos implementaciones (la 2.ª no usa las columnas extraídas aquí) y la extracción por reglas.
Construcción unified_20261025 (solo lectura). Unidad = adjudicación (contrato o lote). Mercado = reglas v1 de A1 (CPV u
objeto con exclusiones). Las copias se marcan (definición de la web), no se quitan: el universo lo fija el cálculo.
Salida: priv/e8_base.parquet (objetos y órganos: carpeta 700)."""
import sys, time
sys.path.insert(0, '/p/scripts'); sys.path.insert(0, '/a1/scripts')
import duckdb, pyarrow as pa, pyarrow.parquet as pq
from reglas import MERCADOS
import reglas_e8 as R
t0 = time.time()
con = duckdb.connect(); con.execute("SET memory_limit='8GB'; SET threads=6; SET temp_directory='/p/tmp/duck'; SET preserve_insertion_order=false")
D = '/d/unified_20261025'; I = D + '.identidad/'
con.execute(f"CREATE VIEW u AS SELECT * FROM read_parquet('{D}/*/*/*.parquet', hive_partitioning=1, union_by_name=1)")
con.execute(f"CREATE VIEW vf AS SELECT source, id_canonical, cuenta FROM read_parquet('{D}.vinculos/filas.parquet')")
con.execute("""CREATE TEMP TABLE b AS SELECT u.source, u.id_canonical, licitacion_id, lote_id, coalesce(es_menor, FALSE) menor, anio_ref,
  cpv_code cpv, objeto, lower(strip_accents(coalesce(objeto,''))) o, imp_adj_eur_comparable imp, imp_adj_eur, iva_base, rol_am, nuts,
  provincia, ccaa, organo_nif, organo_nombre, dir3, fecha_adj, fecha_formalizacion, fecha_pub, duracion_meses, procedimiento,
  num_ofertas, url_origen, copia_de, correccion_imp_adj, v.cuenta
 FROM u LEFT JOIN vf v ON v.source = u.source AND v.id_canonical = u.id_canonical
 WHERE anio_ref BETWEEN 2015 AND 2026 AND unidad = 'adjudicacion'""")
partes = []
for m in ('carburantes', 'electricidad', 'papel'):
    r = MERCADOS[m]
    partes.append(f"""SELECT '{m}' mercado, CASE WHEN {r['cpv']} THEN 'cpv' ELSE 'texto' END via_mercado, * FROM b
      WHERE {r['cpv']} OR (regexp_matches(o, '{r['txt']}') AND NOT regexp_matches(o, '{r['excl']}'))""")
con.execute("CREATE TEMP TABLE m AS " + " UNION ALL ".join(partes))
print('mercados', con.execute("SELECT mercado, count(*) FROM m GROUP BY 1 ORDER BY 1").fetchall(), round(time.time() - t0), flush=True)
con.execute(f"""CREATE TEMP TABLE m2 AS SELECT m.* EXCLUDE (o), o.organo_clave, coalesce(t.tipo, 'sin_tipo') tipo_organo
 FROM m 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 m.organo_nif AND o.organo_nombre IS NOT DISTINCT FROM m.organo_nombre AND o.dir3 IS NOT DISTINCT FROM m.dir3
 LEFT JOIN read_parquet('{I}organos_tipo.parquet') t ON t.organo_clave = o.organo_clave""")
rows = con.execute("SELECT * FROM m2").fetch_arrow_table()
print('filas', rows.num_rows, round(time.time() - t0), flush=True)
obj = rows.column('objeto').to_pylist(); mk = rows.column('mercado').to_pylist()
cols = {k: [] for k in ('x_motivo', 'x_cantidad', 'x_unidad', 'x_literal', 'x_n_distintas', 'x_producto', 'x_anual', 'x_gramaje',
                        'x_formato_explicito', 'x_excl_v2')}
for o, m in zip(obj, mk):
    if m == 'carburantes':
        c = R.cantidad_carb(o); p = R.producto_carb(o); ex = R.excluido_v2(o, 'carburantes')
        vals = ('ok' if c else 'sin_cantidad', c[0] if c else None, 'litros', c[1] if c else None, c[2] if c else 0, p, None, None, None, ex)
    elif m == 'electricidad':
        e = R.energia_elec(o)
        vals = (e['motivo'], e.get('kwh'), 'kwh', e.get('literal'), e.get('n_distintas', 0), 'electricidad', e.get('anual'), None, None, False)
    else:
        e = R.papel(o)
        vals = (e['motivo'], e.get('cajas_eq'), 'caja_2500', e.get('literal'), 1 if e['motivo'] == 'ok' else 0, e.get('fibra'), None,
                e.get('gramaje'), e.get('formato_explicito'), False)
    for k, v in zip(cols, vals):
        cols[k].append(v)
for k, v in cols.items():
    rows = rows.append_column(k, pa.array(v))
pq.write_table(rows, '/p/priv/e8_base.parquet')
con.execute("CREATE VIEW a AS SELECT * FROM read_parquet('/p/priv/e8_base.parquet')")
print(con.execute("SELECT mercado, x_motivo, count(*) FROM a GROUP BY ALL ORDER BY 1, 3 DESC").fetchall())
print(con.execute("SELECT mercado, x_producto, count(*) FROM a WHERE x_motivo='ok' GROUP BY ALL ORDER BY 1, 3 DESC").fetchall())
print('fin', round(time.time() - t0))