DEV Community

CPFHub
CPFHub

Posted on

Higienizando uma base de CPFs em Python com pandas: validação vetorizada, duplicados e consulta em lote

#python #pandas #dados #brasil

Transparência: sou fundador da CPFHub, citada no final como opção para a etapa de consulta. As etapas de limpeza não dependem de nenhum serviço.

Toda empresa tem aquela planilha de clientes exportada do sistema antigo: CPF com máscara, sem máscara, com zero à esquerda perdido pelo Excel, duplicado, ou simplesmente 000.000.000-00. Antes de migrar ou de usar essa base para qualquer coisa, vale limpá-la. Vamos fazer isso com pandas.

1. Carregue como texto

O erro número um é deixar o pandas inferir o tipo. 01234567890 vira 1234567890.

import pandas as pd

df = pd.read_csv("clientes.csv", dtype={"cpf": "string"}, encoding="utf-8")
Enter fullscreen mode Exit fullscreen mode

Se a coluna já veio numérica de um Excel, recupere o zero à esquerda depois de normalizar com zfill(11), mas só quando o tamanho for 10 (ou 9): mais que isso é dado corrompido.

2. Normalize 

df["cpf_limpo"] = df["cpf"].str.replace(r"\D", "", regex=True)
mask_curto = df["cpf_limpo"].str.len().between(9, 10)
df.loc[mask_curto, "cpf_limpo"] = df.loc[mask_curto, "cpf_limpo"].str.zfill(11)
Enter fullscreen mode Exit fullscreen mode

3. Valide de forma vetorizada

Aplicar uma função Python linha a linha funciona, mas com milhões de linhas é lento. Dá para calcular os dígitos verificadores com NumPy:

import numpy as np

def cpf_valido_vec(s: pd.Series) -> pd.Series:
    ok_len = s.str.len() == 11
    s11 = s.where(ok_len, "00000000000")
    d = np.array([list(map(int, x)) for x in s11], dtype=np.int64)
    p1 = np.arange(10, 1, -1)       # 10..2
    p2 = np.arange(11, 1, -1)       # 11..2
    dv1 = (d[:, :9] @ p1) * 10 % 11 % 10
    dv2 = (d[:, :10] @ p2) * 10 % 11 % 10
    repetido = (d == d[:, [0]]).all(axis=1)
    return ok_len & (d[:, 9] == dv1) & (d[:, 10] == dv2) & ~repetido

df["cpf_valido"] = cpf_valido_vec(df["cpf_limpo"].fillna(""))
Enter fullscreen mode Exit fullscreen mode

A multiplicação de matriz (@) calcula a soma ponderada de todas as linhas de uma vez. Em uma base de 1 milhão de linhas, a diferença em relação a apply é grande.

Confira com casos conhecidos:

teste = pd.Series(["52998224725", "11111111111", "52998224724", "123"])
print(cpf_valido_vec(teste).tolist())  # [True, False, False, False]
Enter fullscreen mode Exit fullscreen mode

4. Duplicados 

dups = df[df.duplicated("cpf_limpo", keep=False) & df["cpf_valido"]]
print(f"{dups['cpf_limpo'].nunique()} CPFs aparecem mais de uma vez")
Enter fullscreen mode Exit fullscreen mode

Para decidir qual registro manter, ordene por data de atualização e use drop_duplicates(keep="last"). Não apague nada da base original: gere um arquivo de saída e um relatório.

5. Relatório 

resumo = {
    "total": len(df),
    "validos": int(df["cpf_valido"].sum()),
    "invalidos": int((~df["cpf_valido"]).sum()),
    "duplicados": int(dups["cpf_limpo"].nunique()),
}
print(resumo)
df[~df["cpf_valido"]].to_csv("cpfs_invalidos.csv", index=False)
Enter fullscreen mode Exit fullscreen mode

O arquivo de inválidos vai para o time que pode corrigir na origem.

6. Confirmando existência em lote

O DV só garante formato. Se a base vai ser usada para crédito, faturamento ou comunicação, você provavelmente quer saber quais CPFs existem e se o nome bate. Fazer isso CPF a CPF em uma base grande esbarra em limite por minuto. O caminho mais limpo é um endpoint de lote.

Na CPFHub, por exemplo, existe POST /cpf/bulk, que aceita até 10.000 CPFs por requisição e processa em segundo plano. O fluxo fica assim (detalhes na documentação):

import os, requests

validos = df.loc[df["cpf_valido"], "cpf_limpo"].unique().tolist()
lotes = [validos[i:i + 10_000] for i in range(0, len(validos), 10_000)]

for lote in lotes:
    r = requests.post(
        "https://api.cpfhub.io/cpf/bulk",
        headers={"x-api-key": os.environ["CPFHUB_API_KEY"]},
        json={"cpfs": lote},
        timeout=60,
    )
    r.raise_for_status()
    job_id = r.json()["data"]["jobId"]   # resposta 202
    print("lote enfileirado:", job_id)
Enter fullscreen mode Exit fullscreen mode

Depois, faça polling em GET /cpf/bulk/{jobId} até data.status ser done ou failed. CPFs inválidos são descartados e repetidos são consultados uma vez, mas é melhor limpar antes. Um detalhe importante: o POST não é idempotente, então não repita o envio depois de um timeout sem conferir se o lote já foi criado.

Mande apenas os CPFs válidos e únicos: você não paga por erro de digitação nem por duplicado.

7. Comparando nomes

Com o resultado em mãos, faça o merge e compare nomes normalizados:

import unicodedata

def norm(n):
    n = unicodedata.normalize("NFKD", str(n)).encode("ascii", "ignore").decode()
    return " ".join(n.upper().split())

res["nome_confere"] = res["nome_base"].map(norm) == res["name"].map(norm)
Enter fullscreen mode Exit fullscreen mode

Divergências não são necessariamente fraude: nome de casada, abreviação e erro de digitação são comuns. Separe para revisão.

8. LGPD

Higienizar base é tratamento de dado pessoal. Tenha a finalidade clara, rode o processo em ambiente controlado, não espalhe cópias do CSV e apague os arquivos intermediários ao final.

Sobre a CPFHub

A CPFHub cobra 1 crédito por CPF encontrado no lote (não encontrado não consome) e tem 50 créditos grátis por mês para testar o fluxo com uma amostra antes de rodar a base inteira.

Top comments (0)