"""LT1 · 15 · series mensuales 2012-01 a 2026-09 (DEFINICIONES_PREVIAS §9). Menores: una fuente por (órgano, grupo, año natural).
Escribe salidas/series_mensuales.csv."""
import duckdb
c = duckdb.connect(); c.execute("SET threads=4; SET memory_limit='6GB'; SET temp_directory='/w/tmp/duck'; SET max_temp_directory_size='5GB'")
x = c.execute
x("CREATE TABLE cm AS SELECT *, year(mes) anio FROM '/w/interno/cel_men.parquet'")
x("""CREATE TABLE el AS SELECT org, grupo, anio, arg_min(source, (-n, CASE WHEN source='PLACSP' THEN 0 ELSE 1 END, source)) fuente
     FROM (SELECT org, grupo, anio, source, sum(n) n FROM cm GROUP BY ALL) GROUP BY ALL""")
x("CREATE TABLE m AS SELECT cm.* FROM cm JOIN el ON el.org=cm.org AND el.grupo=cm.grupo AND el.anio=cm.anio AND el.fuente=cm.source")
x("CREATE TABLE n AS SELECT * FROM '/w/interno/cel_nm.parquet'")
MS = {'M_N': ('n', None), 'M_EUR': ('eur', 'n_imp'), 'M_FRANJA': ('fr_num', 'fr_den'), 'M_ALTO': ('alto_num', 'alto_den'), 'M_N2': ('n2_num', 'n2_den'),
      'M_REP': ('rep_num', 'rep_den'), 'M_1183': ('t_num', 't_den'), 'M_1183r': ('t2_num', 't2_den')}
NS = {'NM_N': ('n', None, 'adj'), 'NM_OU': ('ou_num', 'ou_den', 'pub'), 'NM_URG': ('urg_num', 'urg_den', 'pub'), 'NM_EMER': ('emer_num', 'emer_den', 'adj'),
      'NM_NSP': ('nsp_num', 'nsp_den', 'adj'), 'NM_SIMPL': ('simpl_num', 'simpl_den', 'pub'), 'NM_MOD': ('mod_num', 'mod_den', 'adj')}
parts = []
for s, (num, den) in MS.items():
    unit = den or num
    parts.append(f"""SELECT '{s}' serie, grupo, mes, sum({num})::DOUBLE num, {('sum(' + den + ')::DOUBLE') if den else 'NULL::DOUBLE'} den,
        count(DISTINCT org) FILTER (WHERE {unit} > 0) org_mes, '{'recuento' if s in ('M_N', 'M_EUR') else 'razon'}' tipo FROM m GROUP BY ALL""")
for s, (num, den, rel) in NS.items():
    unit = den or num
    parts.append(f"""SELECT '{s}' serie, grupo, mes, sum({num})::DOUBLE num, {('sum(' + den + ')::DOUBLE') if den else 'NULL::DOUBLE'} den,
        count(DISTINCT org) FILTER (WHERE {unit} > 0) org_mes, '{'recuento' if s == 'NM_N' else 'razon'}' tipo FROM n WHERE reloj='{rel}' GROUP BY ALL""")
x("CREATE TABLE s AS " + " UNION ALL ".join(parts))
# órganos activos en los 12 meses previos (incluido el mes): por serie y grupo
x("""CREATE TABLE act AS SELECT serie, grupo, mes, org FROM (""" + " UNION ALL ".join(
    [f"SELECT DISTINCT '{s}' serie, grupo, mes, org FROM m WHERE {den or num} > 0" for s, (num, den) in MS.items()] +
    [f"SELECT DISTINCT '{s}' serie, grupo, mes, org FROM n WHERE reloj='{rel}' AND {den or num} > 0" for s, (num, den, rel) in NS.items()]) + ")")
x("""CREATE TABLE meses AS SELECT range::DATE mes FROM range(DATE '2012-01-01', DATE '2026-10-01', INTERVAL 1 MONTH)""")
x("""CREATE TABLE a12 AS SELECT k.serie, k.grupo, mm.mes, count(DISTINCT a.org) org_12m FROM (SELECT DISTINCT serie, grupo FROM s) k CROSS JOIN meses mm
     LEFT JOIN act a ON a.serie=k.serie AND a.grupo=k.grupo AND a.mes > mm.mes - INTERVAL 12 MONTH AND a.mes <= mm.mes GROUP BY ALL""")
x("""COPY (SELECT s.serie, s.grupo, strftime(s.mes, '%Y-%m') mes, s.tipo, s.num, s.den,
      CASE WHEN s.tipo='razon' THEN s.num / nullif(s.den, 0) ELSE s.num END valor, s.org_mes, a.org_12m,
      CASE WHEN s.tipo='recuento' THEN s.num / nullif(a.org_12m, 0) END valor_por_organo_activo
    FROM s JOIN a12 a USING (serie, grupo, mes) WHERE s.mes BETWEEN DATE '2012-01-01' AND DATE '2026-09-01' ORDER BY 1, 2, 3) TO '/w/salidas/series_mensuales.csv' (HEADER)""")
print(x("SELECT serie, grupo, count(*), min(mes), max(mes) FROM s GROUP BY ALL ORDER BY ALL").fetchall())