import os, json
from pathlib import Path
import duckdb, pandas as pd
import numpy as np
from transbordo_utils import resolver_cache, carregar_pares_oficiais, carregar_paradas, adicionar_baldeacao_real

os.makedirs("/tmp/duckdb_spill", exist_ok=True)
MATRIZ = resolver_cache("matriz_integracao.json")
PARES = carregar_pares_oficiais(MATRIZ)
STOPS = carregar_paradas(resolver_cache("itinerarios/pontos"))

# Meses a comparar (parquet de usuarios por mês). Ajuste se faltar algum.
MESES = {
    "jan": "usuarios_limpo_2026-01-01_2026-01-31.parquet",
    "fev": "usuarios_limpo_2026-02-01_2026-02-28.parquet",
}
JANELA_MIN, JANELA_MAX = 3, 60

try:
    PARES_MESMA = set(tuple(sorted(map(str, x)))
                      for x in json.load(open(resolver_cache("pares_mesma_linha.json"))))
except FileNotFoundError:
    PARES_MESMA = set()

def analisar(parquet_nome):
    pq = resolver_cache(parquet_nome)
    con = duckdb.connect(config={"memory_limit":"500MB","temp_directory":"/tmp/duckdb_spill","threads":"2"})
    od = con.execute(f"""
    WITH seq AS (
      SELECT NUMEROCARTAO, CODLINHA, DATA_HORA_UTILIZACAO dh, LATITUDE, LONGITUDE,
             LEAD(CODLINHA) OVER w cd, LEAD(DATA_HORA_UTILIZACAO) OVER w dhd,
             LEAD(LATITUDE) OVER w lat_d, LEAD(LONGITUDE) OVER w lon_d
      FROM read_parquet('{pq}')
      WINDOW w AS (PARTITION BY NUMEROCARTAO ORDER BY DATA_HORA_UTILIZACAO))
    SELECT CODLINHA CODLINHA_ORIGEM, cd CODLINHA_DESTINO,
           LATITUDE LATITUDE_ORIGEM, LONGITUDE LONGITUDE_ORIGEM, lat_d LATITUDE_DESTINO, lon_d LONGITUDE_DESTINO,
           date_diff('second', dh, dhd)/60.0 espera_min
    FROM seq WHERE dhd IS NOT NULL
      AND date_diff('second', dh, dhd)/60.0 BETWEEN {JANELA_MIN} AND {JANELA_MAX}
    """).df()
    n_user = con.execute(f"SELECT count(*) FROM read_parquet('{pq}')").fetchone()[0]
    con.close()
    od["CODLINHA_ORIGEM"]=od["CODLINHA_ORIGEM"].astype(str); od["CODLINHA_DESTINO"]=od["CODLINHA_DESTINO"].astype(str)
    # coordenadas fora de Curitiba -> NaN (mesmo tratamento do 02)
    bb = lambda la, lo: la.between(-25.75, -25.25) & lo.between(-49.55, -49.10)
    for p in ("ORIGEM", "DESTINO"):
        ok = bb(od[f"LATITUDE_{p}"], od[f"LONGITUDE_{p}"])
        od.loc[~ok, [f"LATITUDE_{p}", f"LONGITUDE_{p}"]] = np.nan
    mesma = [tuple(sorted([o, d])) in PARES_MESMA for o, d in zip(od["CODLINHA_ORIGEM"], od["CODLINHA_DESTINO"])]
    od = od[[not x for x in mesma]].reset_index(drop=True)
    od = adicionar_baldeacao_real(od, STOPS)   # desembarque estimado (REGRAS §3)
    bald_temporal = od[od["CODLINHA_ORIGEM"] != od["CODLINHA_DESTINO"]]
    bald = od[od["baldeacao_real"]].copy()
    bald["o"], bald["d"] = bald["CODLINHA_ORIGEM"], bald["CODLINHA_DESTINO"]
    bald["par"]=list(zip(bald["o"],bald["d"]))
    com = bald["par"].isin(PARES)
    return {
        "registros_usuario": n_user,
        "transbordos": len(od),
        "baldeacoes_temporais": len(bald_temporal),
        "baldeacoes": len(bald),
        "com_integracao": int(com.sum()),
        "sem_integracao": int((~com).sum()),
        "_bald": bald, "_com": com,
    }

res = {}
for nome, pq in MESES.items():
    try:
        res[nome] = analisar(pq)
        print(f"{nome}: ok")
    except FileNotFoundError:
        print(f"{nome}: parquet ainda não existe ({pq}) — pule até a coleta terminar")

jan: ok

fev: ok

# --- Tabela comparativa ---
rows=[]
for nome, r in res.items():
    b=r["baldeacoes"]
    rows.append({"mes":nome, "reg_usuario":r["registros_usuario"], "transbordos":r["transbordos"],
                 "baldeacoes":b, "com_integ":r["com_integracao"], "sem_integ":r["sem_integracao"],
                 "%_sem_integ": round(100*r["sem_integracao"]/max(b,1),1)})
comp = pd.DataFrame(rows).set_index("mes")
display(comp)
if len(comp)==2:
    a,b = comp.index[0], comp.index[1]
    print(f"\nVariação {b} vs {a}:")
    for col in ["reg_usuario","transbordos","baldeacoes","sem_integ"]:
        va,vb=comp.loc[a,col],comp.loc[b,col]
        print(f"  {col}: {va} -> {vb}  ({100*(vb-va)/max(va,1):+.0f}%)")

Variação fev vs jan:
  reg_usuario: 1102970 -> 1396670  (+27%)
  transbordos: 23562 -> 26323  (+12%)
  baldeacoes: 5084 -> 6234  (+23%)
  sem_integ: 3895 -> 5409  (+39%)

# --- Top pares SEM integração: jan vs fev (estabilidade dos candidatos) ---
def ranking(nome):
    r=res[nome]; sem=r["_bald"][~r["_com"]]
    g=sem.groupby(["o","d"]).size().reset_index(name=f"qtd_{nome}")
    g["par"]=g["o"]+"->"+g["d"]
    return g[["par",f"qtd_{nome}"]]
if len(res)==2:
    ms=list(res.keys())
    m=ranking(ms[0]).merge(ranking(ms[1]), on="par", how="outer").fillna(0)
    m["total"]=m[f"qtd_{ms[0]}"]+m[f"qtd_{ms[1]}"]
    m=m.sort_values("total",ascending=False).head(20)
    print("Top 20 pares sem integração (presença nos 2 meses):")
    display(m)
else:
    print("Comparação de pares requer os 2 meses.")

Top 20 pares sem integração (presença nos 2 meses):

	par	qtd_jan	qtd_fev	total
3233	967->205	10.0	40.0	50.0
3099	876->205	17.0	29.0	46.0
3042	870->040	13.0	32.0	45.0
3090	876->040	33.0	11.0	44.0
421	171->170	35.0	8.0	43.0
1268	380->801	15.0	28.0	43.0
2406	701->703	18.0	23.0	41.0
2445	703->701	24.0	13.0	37.0
2828	822->040	12.0	22.0	34.0
1843	609->679	8.0	25.0	33.0
149	040->876	21.0	8.0	29.0
412	170->901	22.0	6.0	28.0
2415	703->040	7.0	21.0	28.0
1067	366->380	0.0	26.0	26.0
299	166->170	24.0	2.0	26.0
3257	967->965	3.0	21.0	24.0
2804	812->040	15.0	8.0	23.0
3217	965->967	2.0	21.0	23.0
148	040->870	7.0	16.0	23.0
961	345->030	10.0	13.0	23.0

	reg_usuario	transbordos	baldeacoes	com_integ	sem_integ	%_sem_integ
mes
jan	1102970	23562	5084	1189	3895	76.6
fev	1396670	26323	6234	825	5409	86.8

08 — Comparativo entre meses (sazonalidade)¶