Como consumir API de CPF em Jupyter Notebook para análise exploratória

# Célula 1: Importações e configuração
import requests
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import re
import time
import os
from tqdm.notebook import tqdm

# Configuração da API
API_KEY = os.getenv("CPFHUB_API_KEY", "SUA_CHAVE_DE_API")
BASE_URL = "https://api.cpfhub.io"
TIMEOUT = 5 # segundos

# Configuração visual
sns.set_theme(style="whitegrid")
pd.set_option("display.max_columns", None)

print("Ambiente configurado com sucesso.")

# Célula 2: Função de consulta
def consultar_cpf(cpf: str) -> dict:
    """
    Consulta dados de um CPF na API da CPFHub.io.
    Retorna dict com dados ou dict com erro.
    """
    cpf_limpo = re.sub(r"\D", "", str(cpf))

    if len(cpf_limpo) != 11:
    return {"error": "CPF inválido", "cpf": cpf_limpo}

    url = f"{BASE_URL}/cpf/{cpf_limpo}"
    headers = {
    "x-api-key": API_KEY,
    "Accept": "application/json",
    }

    try:
    response = requests.get(url, headers=headers, timeout=TIMEOUT)

    if response.status_code == 200:
    data = response.json()
    if data.get("success"):
    return {**data["data"], "error": None}

    error_messages = {
    400: "Formato inválido",
    401: "API key inválida",
    404: "Não encontrado",
    }
    return {
    "error": error_messages.get(response.status_code, f"HTTP {response.status_code}"),
    "cpf": cpf_limpo,
    }

    except requests.exceptions.Timeout:
    return {"error": "Timeout", "cpf": cpf_limpo}
    except requests.exceptions.RequestException as e:
    return {"error": str(e), "cpf": cpf_limpo}

# Teste rápido
resultado = consultar_cpf("12345678900")
print(resultado)

# Célula 3: Consulta individual
cpf_teste = "12345678900"
dados = consultar_cpf(cpf_teste)

if dados.get("error") is None:
    print(f"Nome: {dados['name']}")
    print(f"CPF: {dados['cpf']}")
    print(f"Gênero: {dados['gender']}")
    print(f"Nascimento: {dados['birthDate']}")
else:
    print(f"Erro: {dados['error']}")

# Célula 4: Carregar dataset de exemplo
df = pd.DataFrame({
    "cpf": ["12345678900", "98765432100", "11122233344"],
    "valor_compra": [150.00, 320.50, 89.90],
    "data_compra": ["2026-01-15", "2026-02-20", "2026-03-10"],
})

print(f"Dataset original: {len(df)} registros")
df.head()

# Célula 5: Enriquecer dataset com dados da CPFHub
resultados = []

for idx, row in tqdm(df.iterrows(), total=len(df), desc="Consultando CPFs"):
    dados = consultar_cpf(row["cpf"])
    resultados.append(dados)
    time.sleep(0.5) # Respeitar rate limit

df_enriquecido = df.copy()
df_api = pd.DataFrame(resultados)

# Mesclar dados
if "name" in df_api.columns:
    df_enriquecido["nome"] = df_api["name"]
    df_enriquecido["genero"] = df_api["gender"]
    df_enriquecido["nascimento"] = df_api["birthDate"]
    df_enriquecido["ano_nasc"] = df_api["year"]
    df_enriquecido["erro_api"] = df_api["error"]

print(f"Dataset enriquecido: {len(df_enriquecido)} registros")
df_enriquecido.head()

# Célula 6: Análise de gênero
if "genero" in df_enriquecido.columns:
    fig, axes = plt.subplots(1, 2, figsize=(12, 5))

    # Distribuição de gênero
    genero_counts = df_enriquecido["genero"].value_counts()
    axes[0].bar(genero_counts.index, genero_counts.values, color=["#3498db", "#e74c3c"])
    axes[0].set_title("Distribuicao por Genero")
    axes[0].set_xlabel("Genero")
    axes[0].set_ylabel("Quantidade")

    # Valor médio por gênero
    media_genero = df_enriquecido.groupby("genero")["valor_compra"].mean()
    axes[1].bar(media_genero.index, media_genero.values, color=["#3498db", "#e74c3c"])
    axes[1].set_title("Valor Medio de Compra por Genero")
    axes[1].set_xlabel("Genero")
    axes[1].set_ylabel("Valor Medio (R$)")

    plt.tight_layout()
    plt.show()

# Célula 7: Análise por faixa etária
if "ano_nasc" in df_enriquecido.columns:
    df_enriquecido["idade"] = 2026 - df_enriquecido["ano_nasc"]

    bins = [0, 25, 35, 45, 55, 65, 100]
    labels = ["18-25", "26-35", "36-45", "46-55", "56-65", "65+"]
    df_enriquecido["faixa_etaria"] = pd.cut(
    df_enriquecido["idade"], bins=bins, labels=labels
    )

    fig, ax = plt.subplots(figsize=(10, 5))
    faixa_counts = df_enriquecido["faixa_etaria"].value_counts().sort_index()
    faixa_counts.plot(kind="bar", ax=ax, color="#2ecc71")
    ax.set_title("Distribuicao por Faixa Etaria")
    ax.set_xlabel("Faixa Etaria")
    ax.set_ylabel("Quantidade")
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.show()

# Célula 8: Métricas de qualidade
total = len(df_enriquecido)
sucesso = df_enriquecido["erro_api"].isna().sum()
falha = total - sucesso

print(f"Total de CPFs: {total}")
print(f"Consultas com sucesso: {sucesso} ({sucesso/total*100:.1f}%)")
print(f"Consultas com erro: {falha} ({falha/total*100:.1f}%)")

if falha > 0:
    print("\nDetalhamento dos erros:")
    print(df_enriquecido[df_enriquecido["erro_api"].notna()]["erro_api"].value_counts())

# Célula 9: Exportar resultado
output_path = "dataset_enriquecido.csv"
df_enriquecido.to_csv(output_path, index=False, encoding="utf-8-sig")
print(f"Dataset exportado para: {output_path}")

# Também pode exportar para Excel
df_enriquecido.to_excel("dataset_enriquecido.xlsx", index=False)
print("Exportado também para Excel.")