Boletins de Urna

bu_imgbu_logjez_rdv_vscmr_2022_1t_SP.zip  (1o turno - SP)
bu_imgbu_logjez_rdv_vscmr_2022_2t_SP.zip  (2o turno - SP)

https://cdn.tse.jus.br/estatistica/sead/odsele/boletim_urna/bu_imgbu_logjez_rdv_vscmr_2022_1t_SP.zip
https://cdn.tse.jus.br/estatistica/sead/odsele/boletim_urna/bu_imgbu_logjez_rdv_vscmr_2022_1t_RJ.zip
https://cdn.tse.jus.br/estatistica/sead/odsele/boletim_urna/bu_imgbu_logjez_rdv_vscmr_2022_1t_MG.zip

import requests
import zipfile
import pandas as pd
from io import BytesIO
from pathlib import Path
 
 
def baixar_boletins_urna(
    ano: int,
    turno: int,
    uf: str,
    destino: Path = Path("./dados_tse"),
) -> Path:
    """
    Baixa os boletins de urna de um estado em um turno.
 
    ATENÇÃO: Os arquivos são muito grandes (vários GB).
    Certifique-se de ter espaço em disco suficiente.
 
    Args:
        ano: Ano da eleição
        turno: Turno (1 ou 2)
        uf: Sigla da UF
        destino: Diretório para salvar os arquivos
 
    Returns:
        Path do diretório com os arquivos extraídos
    """
    url = (
        f"https://cdn.tse.jus.br/estatistica/sead/odsele/boletim_urna/"
        f"bu_imgbu_logjez_rdv_vscmr_{ano}_{turno}t_{uf.upper()}.zip"
    )
    print(f"Baixando boletins de urna {ano} - {turno}o turno - {uf.upper()}...")
    print("ATENÇÃO: Download pode demorar vários minutos.")
 
    destino_uf = destino / f"bu_{ano}_{turno}t_{uf.upper()}"
    destino_uf.mkdir(parents=True, exist_ok=True)
 
    # Download em streaming para não estourar memória
    response = requests.get(url, stream=True)
    response.raise_for_status()
 
    arquivo_zip = destino_uf / "bu.zip"
    with open(arquivo_zip, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
 
    print("Download concluído. Extraindo...")
    with zipfile.ZipFile(arquivo_zip) as zf:
        # Extrair apenas os CSVs de boletins (ignorar imagens e logs para economia)
        bu_files = [f for f in zf.namelist() if f.endswith(".csv")]
        for bf in bu_files:
            zf.extract(bf, destino_uf)
 
    print(f"Arquivos extraídos em: {destino_uf}")
    return destino_uf
 
 
# Baixar boletins do 1o turno de SP em 2022
diretorio = baixar_boletins_urna(2022, 1, "SP")

def carregar_boletins_csv(
    diretorio: Path,
    cargo: str = None,
) -> pd.DataFrame:
    """
    Carrega CSVs de boletins de urna de um diretório.
 
    Args:
        diretorio: Diretório contendo os CSVs extraídos
        cargo: Filtrar por cargo (opcional)
 
    Returns:
        DataFrame com dados dos boletins
    """
    csvs = list(diretorio.glob("**/*.csv"))
    if not csvs:
        raise FileNotFoundError(f"Nenhum CSV encontrado em {diretorio}")
 
    dfs = []
    for csv_file in csvs:
        try:
            df = pd.read_csv(
                csv_file,
                sep=";",
                encoding="latin-1",
                dtype=str,
            )
            dfs.append(df)
        except Exception as e:
            print(f"Erro ao ler {csv_file}: {e}")
 
    df_completo = pd.concat(dfs, ignore_index=True)
 
    if cargo:
        df_completo = df_completo[
            df_completo["DS_CARGO_PERGUNTA"].str.upper() == cargo.upper()
        ]
 
    return df_completo
 
 
# Carregar boletins e analisar
df_bu = carregar_boletins_csv(Path("./dados_tse/bu_2022_1t_SP"))
print(f"Registros nos boletins: {len(df_bu):,}")

def auditar_resultados(
    df_bu: pd.DataFrame,
    df_resultados: pd.DataFrame,
    cargo: str,
    municipio: str = None,
) -> pd.DataFrame:
    """
    Compara a soma dos boletins de urna com os resultados oficiais.
 
    Args:
        df_bu: DataFrame com dados dos boletins de urna
        df_resultados: DataFrame com resultados oficiais
        cargo: Cargo a auditar
        municipio: Código do município (opcional)
 
    Returns:
        DataFrame com comparativo
    """
    # Agregar votos dos boletins
    df_bu["QT_VOTOS"] = pd.to_numeric(df_bu["QT_VOTOS"], errors="coerce")
    bu_agregado = (
        df_bu.groupby(["NR_VOTAVEL"])
        .agg(votos_bu=("QT_VOTOS", "sum"))
        .reset_index()
    )
 
    # Agregar votos dos resultados oficiais
    df_resultados["QT_VOTOS_NOMINAIS"] = pd.to_numeric(
        df_resultados["QT_VOTOS_NOMINAIS"], errors="coerce"
    )
    res_agregado = (
        df_resultados.groupby(["NR_CANDIDATO"])
        .agg(votos_oficiais=("QT_VOTOS_NOMINAIS", "sum"))
        .reset_index()
    )
 
    # Comparar
    comparativo = pd.merge(
        bu_agregado,
        res_agregado,
        left_on="NR_VOTAVEL",
        right_on="NR_CANDIDATO",
        how="outer",
    )
    comparativo["diferenca"] = (
        comparativo["votos_bu"].fillna(0) - comparativo["votos_oficiais"].fillna(0)
    )
 
    divergencias = comparativo[comparativo["diferenca"] != 0]
    if divergencias.empty:
        print("AUDITORIA OK: Nenhuma divergência encontrada.")
    else:
        print(f"ALERTA: {len(divergencias)} divergências encontradas!")
 
    return comparativo

import pandas as pd
 
# 1. Carregar boletins de urna de uma zona específica
df_bu = pd.read_csv(
    "dados_tse/bu_2022_1t_SP/boletim_urna_SP.csv",
    sep=";",
    encoding="latin-1",
    dtype=str,
)
 
# 2. Carregar resultados oficiais
df_resultados = pd.read_csv(
    "dados_tse/votacao_secao_2022_SP.csv",
    sep=";",
    encoding="latin-1",
    dtype=str,
)
 
# 3. Filtrar por cargo (ex: Presidente) e turno
df_bu_pres = df_bu[
    (df_bu["DS_CARGO_PERGUNTA"].str.upper() == "PRESIDENTE")
    & (df_bu["NR_TURNO"] == "1")
]
df_res_pres = df_resultados[
    (df_resultados["DS_CARGO"].str.upper() == "PRESIDENTE")
    & (df_resultados["NR_TURNO"] == "1")
]
 
# 4. Somar votos dos BUs por candidato
df_bu_pres["QT_VOTOS"] = pd.to_numeric(df_bu_pres["QT_VOTOS"], errors="coerce")
soma_bu = df_bu_pres.groupby("NR_VOTAVEL")["QT_VOTOS"].sum().reset_index()
soma_bu.columns = ["nr_candidato", "votos_bu"]
 
# 5. Somar votos dos resultados oficiais por candidato
df_res_pres["QT_VOTOS_NOMINAIS"] = pd.to_numeric(
    df_res_pres["QT_VOTOS_NOMINAIS"], errors="coerce"
)
soma_oficial = (
    df_res_pres.groupby("NR_CANDIDATO")["QT_VOTOS_NOMINAIS"].sum().reset_index()
)
soma_oficial.columns = ["nr_candidato", "votos_oficiais"]
 
# 6. Comparar
comparativo = pd.merge(soma_bu, soma_oficial, on="nr_candidato", how="outer")
comparativo["diferenca"] = comparativo["votos_bu"] - comparativo["votos_oficiais"]
 
print("Auditoria — Presidente 1o turno — SP:")
print(comparativo)
print(f"\nDivergências: {(comparativo['diferenca'] != 0).sum()}")