Como integrar validação de CPF em Streamlit para dashboards de dados

[cpfhub]
api_key = "SUA_CHAVE_DE_API"
base_url = "https://api.cpfhub.io"
timeout = 5

# cpf_service.py
import re
import requests
import streamlit as st

def consultar_cpf(cpf: str) -> dict:
    """
    Consulta dados de um CPF na API da CPFHub.io.
    Retorna dict com dados ou lança exceção.
    """
    cpf_limpo = re.sub(r"\D", "", cpf)

    if len(cpf_limpo) != 11:
    raise ValueError("CPF deve conter exatamente 11 dígitos.")

    url = f"{st.secrets.cpfhub.base_url}/cpf/{cpf_limpo}"
    headers = {
    "x-api-key": st.secrets.cpfhub.api_key,
    "Accept": "application/json",
    }

    try:
    response = requests.get(
    url,
    headers=headers,
    timeout=st.secrets.cpfhub.timeout,
    )
    except requests.exceptions.Timeout:
    raise ConnectionError("Timeout ao consultar a API. Tente novamente.")
    except requests.exceptions.RequestException as e:
    raise ConnectionError(f"Erro de conexão: {str(e)}")

    if response.status_code == 200:
    data = response.json()
    if data.get("success"):
    return data["data"]
    raise ValueError("Resposta inesperada da API.")
    elif response.status_code == 401:
    raise PermissionError("Chave de API inválida.")
    elif response.status_code == 404:
    raise LookupError("CPF não encontrado na base de dados.")
    else:
    raise RuntimeError(f"Erro HTTP {response.status_code}")

# app.py
import streamlit as st
import pandas as pd
from cpf_service import consultar_cpf

st.set_page_config(
    page_title="Dashboard de Validação de CPF",
    page_icon="",
    layout="wide",
)

st.title("Dashboard de Validacao de CPF")
st.markdown("Consulta de dados cadastrais via API da **CPFHub.io**")

st.markdown("---")

# -- Seção 1: Consulta individual --
st.header("Consulta Individual")

col1, col2 = st.columns([1, 2])

with col1:
    cpf_input = st.text_input(
    "Digite o CPF",
    placeholder="000.000.000-00",
    max_chars=14,
    )
    consultar_btn = st.button("Consultar CPF", type="primary")

with col2:
    if consultar_btn and cpf_input:
    with st.spinner("Consultando API da CPFHub.io..."):
    try:
    dados = consultar_cpf(cpf_input)
    st.success("CPF encontrado com sucesso!")

    metric_col1, metric_col2, metric_col3 = st.columns(3)
    metric_col1.metric("Nome", dados.get("name", "N/A"))
    metric_col2.metric("Genero", dados.get("gender", "N/A"))
    metric_col3.metric("Data de Nascimento", dados.get("birthDate", "N/A"))

    with st.expander("Ver dados completos"):
    st.json(dados)

    except ValueError as e:
    st.error(f"Erro de validacao: {str(e)}")
    except LookupError as e:
    st.warning(str(e))
    except (ConnectionError, RuntimeError, PermissionError) as e:
    st.error(str(e))

    elif consultar_btn:
    st.warning("Por favor, digite um CPF.")

# Continuação do app.py

st.markdown("---")

# -- Seção 2: Consulta em lote --
st.header("Consulta em Lote")

uploaded_file = st.file_uploader(
    "Envie um CSV com coluna 'cpf'",
    type=["csv"],
)

if uploaded_file is not None:
    df = pd.read_csv(uploaded_file, dtype=str)

    if "cpf" not in df.columns:
    st.error("O arquivo CSV deve conter uma coluna chamada 'cpf'.")
    else:
    st.dataframe(df.head(), use_container_width=True)
    total = len(df)
    st.info(f"Total de CPFs no arquivo: {total}")

    if st.button("Iniciar consulta em lote"):
    resultados = []
    progress_bar = st.progress(0)
    status_text = st.empty()

    for idx, row in df.iterrows():
    cpf = str(row["cpf"])
    status_text.text(f"Consultando {idx + 1}/{total}: {cpf}")

    try:
    dados = consultar_cpf(cpf)
    resultados.append({
    "cpf": cpf,
    "nome": dados.get("name"),
    "genero": dados.get("gender"),
    "nascimento": dados.get("birthDate"),
    "status": "Encontrado",
    })
    except Exception as e:
    resultados.append({
    "cpf": cpf,
    "nome": None,
    "genero": None,
    "nascimento": None,
    "status": str(e),
    })

    progress_bar.progress((idx + 1) / total)

    status_text.text("Consulta finalizada!")
    df_resultado = pd.DataFrame(resultados)
    st.dataframe(df_resultado, use_container_width=True)

    # Métricas de resumo
    encontrados = len(df_resultado[df_resultado["status"] == "Encontrado"])
    erros = total - encontrados

    col_m1, col_m2, col_m3 = st.columns(3)
    col_m1.metric("Total consultado", total)
    col_m2.metric("Encontrados", encontrados)
    col_m3.metric("Erros", erros)

    # Download do resultado
    csv_result = df_resultado.to_csv(index=False)
    st.download_button(
    "Baixar resultado (CSV)",
    csv_result,
    "resultado_cpf.csv",
    "text/csv",
    )

# cpf_service.py (versão com cache)
import re
import requests
import streamlit as st

@st.cache_data(ttl=3600, show_spinner=False)
def consultar_cpf_cached(cpf: str) -> dict:
    """
    Consulta CPF com cache de 1 hora para evitar chamadas duplicadas.
    """
    cpf_limpo = re.sub(r"\D", "", cpf)

    if len(cpf_limpo) != 11:
    raise ValueError("CPF deve conter exatamente 11 dígitos.")

    url = f"{st.secrets.cpfhub.base_url}/cpf/{cpf_limpo}"
    headers = {
    "x-api-key": st.secrets.cpfhub.api_key,
    "Accept": "application/json",
    }

    response = requests.get(url, headers=headers, timeout=st.secrets.cpfhub.timeout)

    if response.status_code == 200:
    data = response.json()
    if data.get("success"):
    return data["data"]
    raise RuntimeError(f"Erro: HTTP {response.status_code}")

streamlit run app.py --server.port 8501