Como Integrar uma API Gratuita de CPF a Sistemas de Gestão

import requests
import os
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class CPFService:
    def __init__(self):
    self.api_key = os.environ.get("CPFHUB_API_KEY", "")
    self.base_url = "https://api.cpfhub.io/cpf"
    self.session = requests.Session()
    self.session.headers["x-api-key"] = self.api_key

    def validar_e_enriquecer(self, cpf: str) -> dict:
    cpf_limpo = "".join(c for c in cpf if c.isdigit())

    if len(cpf_limpo) != 11:
    return {"valido": False, "erro": "CPF deve ter 11 dígitos"}

    try:
    response = self.session.get(
    f"{self.base_url}/{cpf_limpo}",
    timeout=10
    )

    if response.status_code == 200:
    dados = response.json()
    if dados.get("success"):
    return {
    "valido": True,
    "cpf": dados["data"]["cpf"],
    "nome": dados["data"]["name"],
    "nascimento": dados["data"]["birthDate"],
    "genero": dados["data"]["gender"],
    "consultado_em": datetime.now().isoformat(),
    }

    return {"valido": False, "erro": "CPF não encontrado"}

    except requests.exceptions.Timeout:
    logger.warning(f"Timeout na consulta do CPF {cpf_limpo[:3]}***")
    return {"valido": None, "erro": "Timeout na consulta"}
    except Exception as e:
    logger.error(f"Erro na consulta: {e}")
    return {"valido": None, "erro": "Erro de conexão"}

# Integração com cadastro de clientes
class ClienteService:
    def __init__(self):
    self.cpf_service = CPFService()

    def cadastrar_cliente(self, dados_formulario: dict) -> dict:
    cpf = dados_formulario.get("cpf", "")

    # Validar CPF via API
    validacao = self.cpf_service.validar_e_enriquecer(cpf)

    if validacao["valido"] is False:
    return {"sucesso": False, "erro": validacao["erro"]}

    if validacao["valido"] is True:
    # Enriquecer dados do formulário com dados da API
    dados_formulario["nome_completo"] = validacao["nome"]
    dados_formulario["data_nascimento"] = validacao["nascimento"]
    dados_formulario["cpf_validado"] = True

    # Se valido is None (timeout), aceitar mas marcar para revisão
    if validacao["valido"] is None:
    dados_formulario["cpf_validado"] = False
    dados_formulario["pendente_revisao"] = True

    # Gravar no banco de dados
    cliente_id = self._salvar_no_banco(dados_formulario)
    return {"sucesso": True, "cliente_id": cliente_id}

    def _salvar_no_banco(self, dados: dict) -> int:
    # Implementação de persistência
    pass

import csv
import time

class ImportadorClientes:
    def __init__(self):
    self.cpf_service = CPFService()

    def importar_csv(self, caminho_arquivo: str) -> dict:
    resultados = {
    "total": 0, "validos": 0,
    "invalidos": 0, "erros": 0, "detalhes": []
    }

    with open(caminho_arquivo, "r", encoding="utf-8") as f:
    leitor = csv.DictReader(f)
    for linha in leitor:
    resultados["total"] += 1
    cpf = linha.get("cpf", "")

    validacao = self.cpf_service.validar_e_enriquecer(cpf)

    if validacao["valido"] is True:
    resultados["validos"] += 1
    resultados["detalhes"].append({
    "cpf": cpf, "status": "valido",
    "nome_api": validacao["nome"]
    })
    elif validacao["valido"] is False:
    resultados["invalidos"] += 1
    resultados["detalhes"].append({
    "cpf": cpf, "status": "invalido",
    "erro": validacao["erro"]
    })
    else:
    resultados["erros"] += 1

    # Intervalo entre requisições para evitar sobrecarga
    time.sleep(0.1)

    return resultados

from functools import lru_cache
from datetime import datetime, timedelta

class CPFServiceComFallback:
    def __init__(self):
    self.cpf_service = CPFService()
    self._cache = {}
    self._cache_ttl = timedelta(hours=24)

    def validar(self, cpf: str) -> dict:
    cpf_limpo = "".join(c for c in cpf if c.isdigit())

    # Verificar cache primeiro
    if cpf_limpo in self._cache:
    entrada = self._cache[cpf_limpo]
    if datetime.now() - entrada["timestamp"] < self._cache_ttl:
    return entrada["resultado"]

    # Tentar API
    resultado = self.cpf_service.validar_e_enriquecer(cpf_limpo)

    # Cachear resultados positivos
    if resultado["valido"] is True:
    self._cache[cpf_limpo] = {
    "resultado": resultado,
    "timestamp": datetime.now()
    }

    # Fallback: validação local apenas
    if resultado["valido"] is None:
    resultado = self._validacao_local(cpf_limpo)

    return resultado

    def _validacao_local(self, cpf: str) -> dict:
    digits = [int(d) for d in cpf]
    if len(set(digits)) == 1:
    return {"valido": False, "erro": "CPF com dígitos repetidos"}

    soma1 = sum(digits[i] * (10 - i) for i in range(9))
    check1 = 0 if soma1 % 11 < 2 else 11 - (soma1 % 11)
    if digits[9] != check1:
    return {"valido": False, "erro": "Dígito verificador inválido"}

    soma2 = sum(digits[i] * (11 - i) for i in range(10))
    check2 = 0 if soma2 % 11 < 2 else 11 - (soma2 % 11)
    if digits[10] != check2:
    return {"valido": False, "erro": "Dígito verificador inválido"}

    return {
    "valido": None,
    "cpf": cpf,
    "nota": "Formato válido, pendente confirmação via API",
    }