"""LT1 · 10 · universo (§1), firma N2, marcas por menor (§3) y celdas mensuales por órgano. DuckDB.
Escribe interno/cel_men.parquet, interno/cel_nm.parquet, interno/lotes_sara.parquet, interno/org.parquet y salidas/10_base.json."""
import duckdb, json, time
T0 = time.time()
c = duckdb.connect(); c.execute("SET threads=4; SET memory_limit='6GB'; SET temp_directory='/w/tmp/duck'; SET max_temp_directory_size='5GB'; SET preserve_insertion_order=false")
x = c.execute
def log(*a): print(f"[{time.time()-T0:6.1f}s]", *a, flush=True)
R = {}
x("CREATE VIEW u AS SELECT * FROM read_parquet('/d/unified_20261018/**/*.parquet', hive_partitioning=1, union_by_name=1)")
x("CREATE VIEW vf AS SELECT source, id_canonical, cuenta FROM '/d/unified_20261018.vinculos/filas.parquet'")
x("CREATE VIEW of AS SELECT * FROM '/d/unified_20261018.identidad/organos_filas.parquet'")
x("CREATE VIEW ot AS SELECT * FROM '/d/unified_20261018.identidad/organos_tipo.parquet'")
assert x("SELECT count(*) FROM (SELECT organo_nif, organo_nombre, dir3 FROM of GROUP BY ALL HAVING count(*)>1)").fetchone()[0] == 0
assert x("SELECT count(*) FROM (SELECT source, id_canonical FROM vf GROUP BY ALL HAVING count(*)>1)").fetchone()[0] == 0
BASE = """FROM u LEFT JOIN vf ON vf.source=u.source AND vf.id_canonical=u.id_canonical
 LEFT JOIN of o ON o.organo_nif IS NOT DISTINCT FROM u.organo_nif AND o.organo_nombre IS NOT DISTINCT FROM u.organo_nombre AND o.dir3 IS NOT DISTINCT FROM u.dir3
 LEFT JOIN ot t ON t.organo_clave=o.organo_clave
 WHERE u.unidad='adjudicacion' AND u.vigente AND u.es_contrato AND u.copia_de IS NULL AND coalesce(u.correccion_imp_adj,'') <> 'copia_otra_fuente'
  AND coalesce(vf.cuenta, TRUE) AND u.tipo_contrato IN ('Servicios','Suministros','Obras')
  AND u.fecha_adj BETWEEN DATE '2012-01-01' AND DATE '2026-09-30' AND o.organo_clave IS NOT NULL"""
COLS = """u.id_canonical id, u.source, o.organo_clave org,
  CASE WHEN u.tipo_contrato='Obras' THEN 'O' WHEN coalesce(t.agente_secti,FALSE) THEN 'SSI' ELSE 'SS' END grupo,
  u.fecha_adj, date_trunc('month', u.fecha_adj)::DATE mes,
  CASE WHEN u.imp_adj_eur_comparable > 0 AND u.iva_base IN ('sin','sin_medido') AND NOT (u.source='ARAGON' AND u.iva_base='sin_medido')
       THEN u.imp_adj_eur_comparable::DECIMAL(18,2) END imp,
  CASE WHEN u.adjudicatario_nif_valid THEN u.adjudicatario_nif END nif"""
# ---------- órganos
x("""CREATE TABLE org AS SELECT organo_clave org, coalesce(agente_secti,FALSE) secti, coalesce(tipo='salud',FALSE) salud, coalesce(tipo='age',FALSE) age, tipo FROM ot""")
# ---------- menores
x(f"CREATE TABLE m0 AS SELECT {COLS}, u.objeto {BASE} AND u.es_menor")
R['menores'] = x("SELECT count(*) FROM m0").fetchone()[0]; log('menores', R['menores'])
S = "del las los con para por una unos unas sus que sin sobre entre segun desde hasta mediante durante tras ante bajo cada este esta estos estas dicho dicha dels les per amb als pel pels els das dos nas nos cos coa aos eta".split()
G = "servicio suministro contrato contratacion adquisicion compra obra prestacion menor expediente ejecucion realizacion trabajo vario diverso otro general municipal ayuntamiento concello ajuntament servei subministrament contracte adquisicio servizo subministracion zerbitzu hornidura kontratu lote anualidad ano mes enero febrero marzo abril mayo junio julio agosto septiembre setiembre octubre noviembre diciembre".split()
sl = "[" + ",".join("'%s'" % w for w in S) + "]"; gl = "[" + ",".join("'%s'" % w for w in G) + "]"
x(f"""CREATE MACRO sing(w) AS CASE WHEN length(w)>=5 AND suffix(w,'es') AND substr(w, length(w)-2, 1) IN ('r','n','l','d','j','y') THEN left(w, length(w)-2)
   WHEN length(w)>=4 AND suffix(w,'s') AND NOT suffix(w,'ss') THEN left(w, length(w)-1) ELSE w END""")
x(f"""CREATE TABLE fir AS WITH d AS (SELECT DISTINCT objeto FROM m0 WHERE objeto IS NOT NULL),
  t AS (SELECT objeto, list_filter(list_transform(list_filter(string_split(trim(regexp_replace(strip_accents(lower(objeto)), '[^a-z0-9]+', ' ', 'g')), ' '),
        w -> length(w) >= 3 AND NOT list_contains({sl}, w)), w -> sing(w)), w -> NOT list_contains({gl}, w)) tk FROM d)
  SELECT objeto, CASE WHEN len(tk) >= 1 AND len(list_filter(tk[2:], w -> w <> tk[1])) >= 1 THEN tk[1] || ' ' || list_filter(tk[2:], w -> w <> tk[1])[1] END firma FROM t""")
log('firmas', x("SELECT count(*), count(firma) FROM fir").fetchone())
x("""CREATE TABLE m1 AS SELECT m.* EXCLUDE (objeto), f.firma, year(m.fecha_adj) anio FROM m0 m LEFT JOIN fir f USING (objeto)""")
x("DROP TABLE m0"); x("DROP TABLE fir")
x("""CREATE TABLE m2 AS SELECT *,
  CASE WHEN firma IS NOT NULL THEN count(*) OVER (PARTITION BY source, org, grupo, firma ORDER BY fecha_adj RANGE BETWEEN INTERVAL 182 DAY PRECEDING AND INTERVAL 182 DAY FOLLOWING) >= 2 END en_serie,
  CASE WHEN nif IS NOT NULL THEN row_number() OVER (PARTITION BY source, org, grupo, nif, anio ORDER BY fecha_adj, id) > 1 END rep,
  CASE WHEN nif IS NOT NULL AND imp IS NOT NULL THEN sum(imp) OVER (PARTITION BY source, org, grupo, nif, anio ORDER BY fecha_adj, id ROWS UNBOUNDED PRECEDING)
       > CASE WHEN grupo='O' THEN 40000 ELSE 15000 END END sobre_tope,
  CASE WHEN grupo='O' THEN (CASE WHEN fecha_adj < DATE '2018-03-09' THEN 50000 ELSE 40000 END) ELSE (CASE WHEN fecha_adj < DATE '2018-03-09' THEN 18000 ELSE 15000 END) END u_vig
 FROM m1""")
x("DROP TABLE m1")
assert x("SELECT count(*) FROM m2").fetchone()[0] == R['menores']
x("""COPY (SELECT org, grupo, source, mes, count(*) n, count(imp) n_imp, coalesce(sum(imp),0)::DOUBLE eur,
   count(*) FILTER (WHERE imp >= 0.95*u_vig AND imp < u_vig AND imp % 500 <> 0) fr_num,
   count(*) FILTER (WHERE imp >= 0.70*u_vig AND imp < u_vig AND imp % 500 <> 0) fr_den,
   count(*) FILTER (WHERE grupo<>'O' AND imp >= 15000 AND imp <= 50000) alto_num,
   count(*) FILTER (WHERE grupo<>'O' AND imp >= 5000 AND imp <= 50000) alto_den,
   count(*) FILTER (WHERE grupo<>'O' AND imp >= 47500 AND imp <= 50000 AND imp % 500 <> 0) f50_num,
   count(*) FILTER (WHERE grupo<>'O' AND imp >= 35000 AND imp <= 50000 AND imp % 500 <> 0) f50_den,
   count(*) FILTER (WHERE en_serie) n2_num, count(firma) n2_den,
   count(*) FILTER (WHERE rep) rep_num, count(nif) rep_den,
   count(*) FILTER (WHERE sobre_tope) t_num, count(sobre_tope) t_den,
   count(*) FILTER (WHERE sobre_tope AND imp < CASE WHEN grupo='O' THEN 40000 ELSE 15000 END) t2_num,
   count(sobre_tope) FILTER (WHERE imp < CASE WHEN grupo='O' THEN 40000 ELSE 15000 END) t2_den
 FROM m2 GROUP BY ALL ORDER BY ALL) TO '/w/interno/cel_men.parquet' (FORMAT parquet)""")
R['men_por_grupo'] = x("SELECT grupo, count(*), count(imp), count(nif), count(firma) FROM m2 GROUP BY 1 ORDER BY 1").fetchall()
log('cel_men'); x("DROP TABLE m2")
# ---------- no menores PLACSP
x(f"""CREATE TABLE n0 AS SELECT {COLS}, u.fecha_pub, date_trunc('month', CASE WHEN u.fecha_pub <= u.fecha_adj THEN u.fecha_pub ELSE u.fecha_adj END)::DATE mes_pub, u.tramitacion tram, u.procedimiento proc,
   u.procedimiento_native nat, u.num_ofertas nof, coalesce(u.valor_estimado_eur_expediente, u.valor_estimado_eur) ve, coalesce(u.financiacion_ue, FALSE) ue,
   u.sara, u.ccaa, u.tabla_origen {BASE} AND u.source='PLACSP' AND NOT coalesce(u.es_menor, FALSE)""")
R['no_menores'] = x("SELECT count(*) FROM n0").fetchone()[0]; log('no menores', R['no_menores'])
x("CREATE TABLE tp AS SELECT DISTINCT id_b id FROM '/d/unified_20261018.vinculos/parejas.parquet' WHERE par IN ('ted__placsp_lic','ted__placsp_agr')")
x("CREATE TABLE md AS SELECT id_canonical id, min(fecha_version) f FROM '/d/unified_20261018.modificaciones/modificaciones.parquet' WHERE id_canonical IS NOT NULL GROUP BY 1")
R['mod_ids'] = x("SELECT count(*) FROM md").fetchone()[0]
x("""CREATE TABLE n1 AS SELECT n.*, tp.id IS NOT NULL ted, md.f IS NOT NULL AND md.f <= n.fecha_adj + INTERVAL 548 DAY modif, md.f IS NOT NULL modif_alguna
  FROM n0 n LEFT JOIN tp ON tp.id=n.id LEFT JOIN md ON md.id=n.id""")
R['ted_casadas'] = x("SELECT count(*) FILTER (WHERE ted), count(*) FROM n1").fetchone()
R['mod_casadas_feed_propio'] = x("SELECT year(fecha_adj), count(*) FILTER (WHERE modif_alguna), count(*) FROM n1 WHERE tabla_origen='resultados_licitaciones' AND year(fecha_adj)>=2019 GROUP BY 1 ORDER BY 1").fetchall()
R['fecha_pub_nula_por_anio'] = x("SELECT year(fecha_adj), count(*) FILTER (WHERE fecha_pub IS NULL), count(*) FROM n1 GROUP BY 1 ORDER BY 1").fetchall()
R['ccaa_valores'] = x("SELECT ccaa, count(*) FROM n1 GROUP BY 1 ORDER BY 2 DESC LIMIT 25").fetchall()
# Cataluña: mayoría de filas 2014-2019 con ccaa catalana
x("""CREATE TABLE cat AS SELECT org, mode(ccaa) cc FROM n1 WHERE year(fecha_adj) BETWEEN 2014 AND 2019 AND ccaa IS NOT NULL GROUP BY 1""")
x("ALTER TABLE org ADD COLUMN cat BOOLEAN DEFAULT FALSE")
x("UPDATE org SET cat = TRUE FROM cat WHERE cat.org=org.org AND cat.cc = 'CT'")
R['org_cat'] = x("SELECT count(*) FILTER (WHERE cat), count(*) FROM org").fetchone()
SER = """count(*) n,
  count(*) FILTER (WHERE proc IN ('Abierto','Restringido','Negociado_publicidad','Dialogo_competitivo','Asociacion_innovacion') AND coalesce(nat,'')<>'Derivado de acuerdo marco' AND coalesce(tram,'')<>'emergencia' AND nof=1) ou_num,
  count(*) FILTER (WHERE proc IN ('Abierto','Restringido','Negociado_publicidad','Dialogo_competitivo','Asociacion_innovacion') AND coalesce(nat,'')<>'Derivado de acuerdo marco' AND coalesce(tram,'')<>'emergencia' AND nof>=1) ou_den,
  count(*) FILTER (WHERE tram='urgente') urg_num, count(*) FILTER (WHERE tram IN ('ordinaria','urgente')) urg_den,
  count(*) FILTER (WHERE tram='emergencia') emer_num, count(tram) emer_den,
  count(*) FILTER (WHERE proc='Negociado_sin_publicidad') nsp_num,
  count(*) FILTER (WHERE proc IN ('Abierto','Restringido','Negociado_publicidad','Negociado_sin_publicidad','Dialogo_competitivo','Asociacion_innovacion')) nsp_den,
  count(*) FILTER (WHERE proc='Abierto' AND nat='Abierto simplificado') simpl_num, count(*) FILTER (WHERE proc='Abierto') simpl_den,
  count(*) FILTER (WHERE tabla_origen='resultados_licitaciones' AND modif) mod_num, count(*) FILTER (WHERE tabla_origen='resultados_licitaciones') mod_den,
  count(fecha_pub) n_con_pub"""
x(f"""COPY (SELECT org, grupo, ue, 'adj' reloj, mes, {SER} FROM n1 GROUP BY ALL
   UNION ALL SELECT org, grupo, ue, 'pub' reloj, mes_pub mes, {SER} FROM n1 WHERE mes_pub IS NOT NULL GROUP BY ALL ORDER BY ALL) TO '/w/interno/cel_nm.parquet' (FORMAT parquet)""")
x("""COPY (SELECT n.org, o.age, n.mes, n.mes_pub, n.ve::DOUBLE ve, n.ted, n.sara FROM n1 n JOIN org o USING (org)
   WHERE n.grupo IN ('SS','SSI') AND n.proc IN ('Abierto','Restringido','Negociado_publicidad','Dialogo_competitivo','Asociacion_innovacion')
   AND n.ve >= 100000 AND n.ve < 400000 ORDER BY ALL) TO '/w/interno/lotes_sara.parquet' (FORMAT parquet)""")
x("COPY (SELECT * FROM org ORDER BY org) TO '/w/interno/org.parquet' (FORMAT parquet)")
json.dump(R, open('/w/salidas/10_base.json', 'w'), ensure_ascii=False, indent=1, default=str); log(json.dumps(R, default=str)[:3000])