API de CPF para verificação de dados em processos de crédito consignado

import requests
from datetime import datetime

class VerificadorConsignado:
    def __init__(self, api_key):
    self.session = requests.Session()
    self.session.headers.update({
    'x-api-key': api_key,
    'Accept': 'application/json'
    })

    def verificar_solicitante(self, cpf, nome_informado, nascimento_informado):
    """
    Verifica os dados do solicitante contra a API.
    nascimento_informado: formato 'DD/MM/YYYY'
    """
    url = f'https://api.cpfhub.io/cpf/{cpf}'
    response = self.session.get(url, timeout=15)

    if response.status_code != 200:
    return {
    'aprovado': False,
    'motivo': f'Erro na consulta: HTTP {response.status_code}',
    'score_confianca': 0
    }

    data = response.json()

    if not data.get('success'):
    return {
    'aprovado': False,
    'motivo': 'CPF nao encontrado',
    'score_confianca': 0
    }

    pessoa = data['data']
    verificacoes = []
    pontos = 0

    # Verificacao 1: Nome
    nome_match = self._comparar_nomes(pessoa['name'], nome_informado)
    verificacoes.append({
    'campo': 'nome',
    'informado': nome_informado,
    'encontrado': pessoa['name'],
    'match': nome_match
    })
    if nome_match:
    pontos += 40

    # Verificacao 2: Data de nascimento
    nascimento_match = pessoa['birthDate'] == nascimento_informado
    verificacoes.append({
    'campo': 'nascimento',
    'informado': nascimento_informado,
    'encontrado': pessoa['birthDate'],
    'match': nascimento_match
    })
    if nascimento_match:
    pontos += 40

    # Verificacao 3: CPF valido e consultavel
    pontos += 20

    score = pontos
    aprovado = score >= 80

    return {
    'aprovado': aprovado,
    'score_confianca': score,
    'verificacoes': verificacoes,
    'dados_api': {
    'nome': pessoa['name'],
    'genero': pessoa['gender'],
    'nascimento': pessoa['birthDate']
    },
    'motivo': 'aprovado' if aprovado else 'dados_divergentes'
    }

    def _comparar_nomes(self, nome1, nome2):
    palavras1 = set(nome1.upper().split())
    palavras2 = set(nome2.upper().split())
    if not palavras1 or not palavras2:
    return False
    intersecao = palavras1 & palavras2
    return len(intersecao) / max(len(palavras1), len(palavras2)) >= 0.75

# Uso
verificador = VerificadorConsignado('SUA_CHAVE_DE_API')
resultado = verificador.verificar_solicitante(
    cpf='12345678900',
    nome_informado='Joao da Silva',
    nascimento_informado='15/06/1990'
)
print(resultado)

def processar_solicitacao_consignado(verificador, solicitacao):
    # Etapa 1: Verificar dados do solicitante
    verificacao = verificador.verificar_solicitante(
    cpf=solicitacao['cpf'],
    nome_informado=solicitacao['nome'],
    nascimento_informado=solicitacao['nascimento']
    )

    if not verificacao['aprovado']:
    return {
    'status': 'rejeitado',
    'etapa': 'verificacao_cadastral',
    'motivo': verificacao['motivo'],
    'score': verificacao['score_confianca']
    }

    # Etapa 2: Verificar idade minima (normalmente 18+)
    dados = verificacao['dados_api']
    # Extrair ano da data de nascimento
    partes = dados['nascimento'].split('/')
    ano_nascimento = int(partes[2])
    idade = datetime.now().year - ano_nascimento

    if idade < 18:
    return {
    'status': 'rejeitado',
    'etapa': 'verificacao_idade',
    'motivo': 'solicitante_menor_de_idade'
    }

    # Etapa 3: Aprovado para analise de credito
    return {
    'status': 'aprovado_para_analise',
    'score_cadastral': verificacao['score_confianca'],
    'dados_validados': verificacao['dados_api'],
    'proximo_passo': 'analise_credito'
    }

import csv
import time

def validar_base_beneficiarios(verificador, arquivo):
    resultados = {
    'total': 0,
    'validos': 0,
    'divergentes': 0,
    'nao_encontrados': 0,
    'erros': 0
    }

    with open(arquivo, 'r') as f:
    reader = csv.DictReader(f)
    for row in reader:
    resultado = verificador.verificar_solicitante(
    cpf=row['cpf'],
    nome_informado=row['nome'],
    nascimento_informado=row['nascimento']
    )

    resultados['total'] += 1

    if resultado['aprovado']:
    resultados['validos'] += 1
    elif resultado['motivo'] == 'CPF nao encontrado':
    resultados['nao_encontrados'] += 1
    elif resultado['motivo'] == 'dados_divergentes':
    resultados['divergentes'] += 1
    else:
    resultados['erros'] += 1

    time.sleep(2) # respeitar rate limit

    return resultados

async function pipelineConsignado(solicitacao) {
    const etapas = [
    { nome: 'verificacao_cpf', fn: verificarCPF },
    { nome: 'analise_margem', fn: analisarMargem },
    { nome: 'score_credito', fn: calcularScore },
    { nome: 'aprovacao', fn: aprovar }
    ];

    let contexto = { solicitacao };

    for (const etapa of etapas) {
    const resultado = await etapa.fn(contexto);

    if (resultado.status === 'rejeitado') {
    return { aprovado: false, etapa_rejeicao: etapa.nome, ...resultado };
    }

    contexto = { ...contexto, [etapa.nome]: resultado };
    }

    return { aprovado: true, contexto };
}

async function verificarCPF(ctx) {
    const { cpf, nome } = ctx.solicitacao;

    const response = await fetch(`https://api.cpfhub.io/cpf/${cpf}`, {
    method: 'GET',
    headers: {
    'x-api-key': process.env.CPFHUB_API_KEY,
    'Accept': 'application/json'
    },
    signal: AbortSignal.timeout(15000)
    });

    const data = await response.json();

    if (!data.success) {
    return { status: 'rejeitado', motivo: 'cpf_nao_encontrado' };
    }

    // Comparar nome
    const nomeAPI = data.data.name.toUpperCase();
    const nomeInformado = nome.toUpperCase();

    if (!nomeAPI.includes(nomeInformado.split(' ')[0])) {
    return { status: 'rejeitado', motivo: 'nome_divergente' };
    }

    return { status: 'aprovado', dados: data.data };
}