Candidaturas

https://cdn.tse.jus.br/estatistica/sead/odsele/consulta_cand/consulta_cand_2022.zip
https://cdn.tse.jus.br/estatistica/sead/odsele/consulta_cand/consulta_cand_2020.zip
https://cdn.tse.jus.br/estatistica/sead/odsele/consulta_cand/consulta_cand_2018.zip

import requests
import zipfile
import pandas as pd
from io import BytesIO
from pathlib import Path
 
 
def baixar_candidatos(ano: int, destino: Path = Path("./dados_tse")) -> pd.DataFrame:
    """
    Baixa e carrega dados de candidatos de um ano eleitoral.
 
    Args:
        ano: Ano da eleição (ex: 2022, 2020, 2018)
        destino: Diretório para salvar os arquivos
 
    Returns:
        DataFrame com todos os candidatos do ano
    """
    url = f"https://cdn.tse.jus.br/estatistica/sead/odsele/consulta_cand/consulta_cand_{ano}.zip"
    print(f"Baixando candidatos de {ano}...")
 
    response = requests.get(url)
    response.raise_for_status()
 
    destino.mkdir(parents=True, exist_ok=True)
 
    with zipfile.ZipFile(BytesIO(response.content)) as zf:
        # Extrair o arquivo consolidado nacional
        arquivo_brasil = [
            f for f in zf.namelist()
            if f.endswith(f"_BRASIL.csv") or f.endswith(f"BRASIL.csv")
        ]
        if arquivo_brasil:
            zf.extract(arquivo_brasil[0], destino)
            caminho_csv = destino / arquivo_brasil[0]
        else:
            # Extrair todos os arquivos
            zf.extractall(destino)
            csvs = [f for f in zf.namelist() if f.endswith(".csv")]
            caminho_csv = destino / csvs[0]
 
    df = pd.read_csv(
        caminho_csv,
        sep=";",
        encoding="latin-1",
        dtype=str,
    )
 
    print(f"Candidatos carregados: {len(df):,}")
    return df
 
 
# Baixar candidatos das eleições gerais de 2022
df_cand = baixar_candidatos(2022)
print(df_cand.head())

def filtrar_candidatos(
    df: pd.DataFrame,
    cargo: str = None,
    partido: str = None,
    uf: str = None,
) -> pd.DataFrame:
    """
    Filtra candidatos por cargo, partido e/ou UF.
 
    Args:
        df: DataFrame com dados de candidatos
        cargo: Descrição do cargo (ex: 'DEPUTADO FEDERAL')
        partido: Sigla do partido (ex: 'PT', 'PL')
        uf: Sigla da UF (ex: 'SP', 'RJ')
 
    Returns:
        DataFrame filtrado
    """
    resultado = df.copy()
 
    if cargo:
        resultado = resultado[resultado["DS_CARGO"].str.upper() == cargo.upper()]
    if partido:
        resultado = resultado[resultado["SG_PARTIDO"].str.upper() == partido.upper()]
    if uf:
        resultado = resultado[resultado["SG_UF"].str.upper() == uf.upper()]
 
    return resultado
 
 
# Exemplo: deputados federais do PT em São Paulo
deputados_pt_sp = filtrar_candidatos(df_cand, cargo="DEPUTADO FEDERAL", partido="PT", uf="SP")
print(f"Deputados federais PT-SP: {len(deputados_pt_sp)}")
print(deputados_pt_sp[["NM_CANDIDATO", "NR_CPF_CANDIDATO", "NR_CANDIDATO"]].head(10))

# 1. Carregar candidatos
df_cand = baixar_candidatos(2022)
 
# 2. Carregar QSA da Receita Federal (previamente baixado)
df_qsa = pd.read_csv(
    "dados_rfb/Socios0.csv",
    sep=";",
    header=None,
    names=[
        "cnpj_basico", "identificador_socio", "nome_socio",
        "cpf_cnpj_socio", "qualificacao_socio", "data_entrada",
        "pais", "representante_legal", "nome_representante",
        "qualificacao_representante", "faixa_etaria",
    ],
    dtype=str,
    encoding="latin-1",
)
 
# 3. Cruzar candidatos com sócios de empresas pelo CPF
df_cand["CPF_LIMPO"] = df_cand["NR_CPF_CANDIDATO"].str.zfill(11)
df_qsa["CPF_LIMPO"] = df_qsa["cpf_cnpj_socio"].str.zfill(11)
 
candidatos_socios = pd.merge(
    df_cand[["NM_CANDIDATO", "CPF_LIMPO", "DS_CARGO", "SG_PARTIDO", "SG_UF"]],
    df_qsa[["CPF_LIMPO", "cnpj_basico", "nome_socio", "qualificacao_socio"]],
    on="CPF_LIMPO",
    how="inner",
)
 
print(f"Candidatos com participação societária: {candidatos_socios['CPF_LIMPO'].nunique()}")
print(candidatos_socios.head(10))

import requests
import pandas as pd
 
API_KEY = "SEU_TOKEN_AQUI"
BASE_URL = "https://api.portaldatransparencia.gov.br/api-de-dados"
headers = {"chave-api-dados": API_KEY, "Accept": "application/json"}
 
# 1. Carregar candidatos (previamente baixados)
df_cand = pd.read_csv(
    "dados_tse/consulta_cand_2022_BRASIL.csv",
    sep=";",
    encoding="latin-1",
    dtype=str,
)
 
# 2. Para cada candidato, consultar se é servidor federal
# (ATENÇÃO: respeitar rate limit de 30 req/min)
import time
 
candidatos_servidores = []
for _, row in df_cand.head(50).iterrows():
    cpf = row["NR_CPF_CANDIDATO"]
    if pd.isna(cpf) or cpf == "":
        continue
 
    resp = requests.get(
        f"{BASE_URL}/servidores/por-cpf",
        headers=headers,
        params={"cpf": cpf},
    )
    if resp.status_code == 200 and resp.json():
        candidatos_servidores.append({
            "candidato": row["NM_CANDIDATO"],
            "cpf": cpf,
            "cargo_eletivo": row["DS_CARGO"],
            "partido": row["SG_PARTIDO"],
            "servidor_info": resp.json(),
        })
    time.sleep(2)  # Respeitar rate limit
 
print(f"Candidatos que são servidores federais: {len(candidatos_servidores)}")