Validação de CPF para programas de pontos e milhas aéreas

import requests
import re
from datetime import datetime, timedelta
from typing import Optional
import json

CPFHUB_API_URL = "https://api.cpfhub.io/cpf"
CPFHUB_API_KEY = "SUA_CHAVE_DE_API"
REQUEST_TIMEOUT = 10 # segundos

# Simulação de banco de dados
contas_fidelidade = {} # cpf -> conta
transacoes = []

def consultar_cpf(cpf: str) -> Optional[dict]:
    """Consulta CPF na API CPFHub.io."""
    cpf_limpo = re.sub(r"\D", "", cpf)

    try:
    response = requests.get(
    f"{CPFHUB_API_URL}/{cpf_limpo}",
    headers={
    "x-api-key": CPFHUB_API_KEY,
    "Accept": "application/json",
    },
    timeout=REQUEST_TIMEOUT,
    )
    response.raise_for_status()
    dados = response.json()

    if dados.get("success"):
    return dados["data"]
    return None

    except requests.exceptions.Timeout:
    raise Exception("Timeout na consulta de CPF")
    except requests.exceptions.HTTPError as e:
    status = e.response.status_code
    if status == 404:
    return None
    if status == 401:
    raise Exception("API key inválida")
    raise Exception(f"Erro HTTP {status}")
    except requests.exceptions.RequestException:
    raise Exception("Erro de conexão")

def cadastrar_membro(cpf: str, nome: str, email: str) -> dict:
    """Cadastra um novo membro no programa de fidelidade."""
    cpf_limpo = re.sub(r"\D", "", cpf)

    # Verifica duplicidade
    if cpf_limpo in contas_fidelidade:
    return {
    "sucesso": False,
    "erro": "Este CPF já possui uma conta no programa de fidelidade",
    }

    # Valida CPF na API
    try:
    dados = consultar_cpf(cpf_limpo)
    except Exception as e:
    return {"sucesso": False, "erro": str(e)}

    if not dados:
    return {"sucesso": False, "erro": "CPF não encontrado na base"}

    # Verifica correspondência de nome
    nome_api = dados.get("nameUpper", "").split()
    nome_informado = nome.upper().split()

    if not nome_api or not nome_informado:
    return {"sucesso": False, "erro": "Nomes não puderam ser comparados"}

    if nome_api[0] != nome_informado[0]:
    return {
    "sucesso": False,
    "erro": "O primeiro nome informado não corresponde ao CPF",
    }

    # Cria conta
    conta = {
    "cpf": cpf_limpo,
    "nome": dados.get("name"),
    "nomeUpper": dados.get("nameUpper"),
    "email": email,
    "dataNascimento": dados.get("birthDate"),
    "genero": dados.get("gender"),
    "saldoPontos": 500, # bônus de boas-vindas
    "nivel": "BASICO",
    "verificado": True,
    "criadoEm": datetime.now().isoformat(),
    }

    contas_fidelidade[cpf_limpo] = conta

    transacoes.append({
    "cpf": cpf_limpo,
    "tipo": "CREDITO",
    "descricao": "Bônus de boas-vindas",
    "pontos": 500,
    "data": datetime.now().isoformat(),
    })

    return {
    "sucesso": True,
    "conta": {
    "nome": conta["nome"],
    "saldo": conta["saldoPontos"],
    "nivel": conta["nivel"],
    },
    "mensagem": "Conta criada com 500 pontos de boas-vindas",
    }

def acumular_pontos(cpf: str, valor_compra: float, parceiro: str) -> dict:
    """Acumula pontos por compra em parceiro."""
    cpf_limpo = re.sub(r"\D", "", cpf)

    conta = contas_fidelidade.get(cpf_limpo)
    if not conta:
    return {"sucesso": False, "erro": "Conta não encontrada"}

    if not conta.get("verificado"):
    return {"sucesso": False, "erro": "Conta não verificada"}

    # Calcula pontos (1 ponto a cada R$ 1 gasto)
    pontos = int(valor_compra)

    # Multiplicador por nível
    multiplicadores = {"BASICO": 1, "PRATA": 1.5, "OURO": 2, "DIAMANTE": 3}
    multiplicador = multiplicadores.get(conta["nivel"], 1)
    pontos_final = int(pontos * multiplicador)

    conta["saldoPontos"] += pontos_final

    transacoes.append({
    "cpf": cpf_limpo,
    "tipo": "CREDITO",
    "descricao": f"Compra em {parceiro} - R$ {valor_compra:.2f}",
    "pontos": pontos_final,
    "multiplicador": multiplicador,
    "data": datetime.now().isoformat(),
    })

    return {
    "sucesso": True,
    "pontosAcumulados": pontos_final,
    "multiplicador": multiplicador,
    "saldoAtual": conta["saldoPontos"],
    }

def transferir_pontos(
    cpf_origem: str, cpf_destino: str, quantidade: int
) -> dict:
    """Transfere pontos entre contas com validação de CPF."""
    cpf_orig = re.sub(r"\D", "", cpf_origem)
    cpf_dest = re.sub(r"\D", "", cpf_destino)

    # Mesma conta
    if cpf_orig == cpf_dest:
    return {"sucesso": False, "erro": "Não é possível transferir para si mesmo"}

    # Verifica contas
    conta_origem = contas_fidelidade.get(cpf_orig)
    conta_destino = contas_fidelidade.get(cpf_dest)

    if not conta_origem:
    return {"sucesso": False, "erro": "Conta de origem não encontrada"}
    if not conta_destino:
    return {"sucesso": False, "erro": "Conta de destino não encontrada"}

    # Revalida CPF de origem (proteção contra roubo de conta)
    try:
    dados_origem = consultar_cpf(cpf_orig)
    if not dados_origem:
    return {
    "sucesso": False,
    "erro": "Falha na revalidação do CPF de origem",
    }
    except Exception as e:
    return {"sucesso": False, "erro": f"Erro na validação: {str(e)}"}

    # Verifica saldo
    if conta_origem["saldoPontos"] < quantidade:
    return {"sucesso": False, "erro": "Saldo insuficiente"}

    # Limite de transferência por dia
    hoje = datetime.now().date()
    transferencias_hoje = [
    t for t in transacoes
    if t["cpf"] == cpf_orig
    and t["tipo"] == "TRANSFERENCIA_SAIDA"
    and datetime.fromisoformat(t["data"]).date() == hoje
    ]
    total_transferido_hoje = sum(t["pontos"] for t in transferencias_hoje)

    LIMITE_DIARIO = 10000
    if total_transferido_hoje + quantidade > LIMITE_DIARIO:
    return {
    "sucesso": False,
    "erro": f"Limite diário de transferência ({LIMITE_DIARIO} pontos) excedido",
    }

    # Executa transferência
    conta_origem["saldoPontos"] -= quantidade
    conta_destino["saldoPontos"] += quantidade

    transacoes.append({
    "cpf": cpf_orig,
    "tipo": "TRANSFERENCIA_SAIDA",
    "descricao": f"Transferência para {conta_destino['nome']}",
    "pontos": quantidade,
    "cpf_destino": cpf_dest,
    "data": datetime.now().isoformat(),
    })

    transacoes.append({
    "cpf": cpf_dest,
    "tipo": "TRANSFERENCIA_ENTRADA",
    "descricao": f"Transferência de {conta_origem['nome']}",
    "pontos": quantidade,
    "cpf_origem": cpf_orig,
    "data": datetime.now().isoformat(),
    })

    return {
    "sucesso": True,
    "pontosTransferidos": quantidade,
    "saldoOrigem": conta_origem["saldoPontos"],
    "destinatario": conta_destino["nome"],
    }

def resgatar_milhas(cpf: str, quantidade: int, destino: str) -> dict:
    """Resgata pontos como milhas aéreas."""
    cpf_limpo = re.sub(r"\D", "", cpf)

    conta = contas_fidelidade.get(cpf_limpo)
    if not conta:
    return {"sucesso": False, "erro": "Conta não encontrada"}

    # Revalida CPF antes do resgate
    try:
    dados = consultar_cpf(cpf_limpo)
    if not dados:
    return {"sucesso": False, "erro": "CPF não pôde ser revalidado"}
    except Exception as e:
    return {"sucesso": False, "erro": f"Erro na validação: {str(e)}"}

    if conta["saldoPontos"] < quantidade:
    return {"sucesso": False, "erro": "Saldo insuficiente"}

    # Taxa de conversão: 1 ponto = 1 milha
    milhas = quantidade

    conta["saldoPontos"] -= quantidade

    transacoes.append({
    "cpf": cpf_limpo,
    "tipo": "RESGATE",
    "descricao": f"Resgate de {milhas} milhas para {destino}",
    "pontos": quantidade,
    "data": datetime.now().isoformat(),
    })

    return {
    "sucesso": True,
    "milhasResgatadas": milhas,
    "programaDestino": destino,
    "saldoRestante": conta["saldoPontos"],
    "titular": dados.get("name"),
    }

# Exemplo de uso
if __name__ == "__main__":
    # Cadastro
    print("--- Cadastro ---")
    r1 = cadastrar_membro("123.456.789-09", "João da Silva", "joao@email.com")
    print(json.dumps(r1, indent=2, ensure_ascii=False))

    # Acúmulo
    print("\n--- Acúmulo ---")
    r2 = acumular_pontos("12345678909", 1500.00, "Loja Online ABC")
    print(json.dumps(r2, indent=2, ensure_ascii=False))

    # Cadastro de segundo membro para transferência
    r3 = cadastrar_membro("987.654.321-00", "Maria Oliveira", "maria@email.com")

    # Transferência
    print("\n--- Transferência ---")
    r4 = transferir_pontos("12345678909", "98765432100", 500)
    print(json.dumps(r4, indent=2, ensure_ascii=False))

    # Resgate
    print("\n--- Resgate ---")
    r5 = resgatar_milhas("12345678909", 1000, "Smiles")
    print(json.dumps(r5, indent=2, ensure_ascii=False))