Validação de CPF para plataformas de venda de passagens aéreas e rodoviárias

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

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
    mensagens = {
    401: "Chave de API inválida",
    404: None, # CPF não encontrado
    }
    if status in mensagens:
    if mensagens[status] is None:
    return None
    raise Exception(mensagens[status])
    raise Exception(f"Erro HTTP {status}")
    except requests.exceptions.RequestException:
    raise Exception("Erro de conexão com a API")

def normalizar_nome(nome: str) -> str:
    """Remove acentos e normaliza o nome para comparação."""
    import unicodedata
    nome_normalizado = unicodedata.normalize("NFKD", nome)
    nome_sem_acento = nome_normalizado.encode("ASCII", "ignore").decode("ASCII")
    return nome_sem_acento.upper().strip()

def verificar_correspondencia_nome(
    nome_informado: str, nome_api: str, tolerancia: float = 0.8
) -> dict:
    """Verifica se o nome informado corresponde ao retornado pela API."""
    partes_informado = normalizar_nome(nome_informado).split()
    partes_api = normalizar_nome(nome_api).split()

    # Verifica primeiro e último nome
    primeiro_ok = partes_informado[0] == partes_api[0] if partes_informado and partes_api else False
    ultimo_ok = partes_informado[-1] == partes_api[-1] if partes_informado and partes_api else False

    # Calcula score de similaridade
    partes_em_comum = set(partes_informado) & set(partes_api)
    total_partes = max(len(partes_informado), len(partes_api))
    score = len(partes_em_comum) / total_partes if total_partes > 0 else 0

    return {
    "corresponde": primeiro_ok and ultimo_ok and score >= tolerancia,
    "score": round(score, 2),
    "primeiro_nome_ok": primeiro_ok,
    "ultimo_nome_ok": ultimo_ok,
    "nome_sugerido": nome_api,
    }

def validar_passageiro(
    cpf: str,
    nome_passageiro: str,
    data_viagem: str,
) -> dict:
    """Valida CPF e dados do passageiro para emissão de bilhete."""

    cpf_limpo = re.sub(r"\D", "", cpf)

    # Validação estrutural do CPF
    if len(cpf_limpo) != 11:
    return {"valido": False, "erro": "CPF deve ter 11 dígitos"}

    # Consulta à API
    try:
    dados = consultar_cpf(cpf_limpo)
    except Exception as e:
    return {"valido": False, "erro": str(e)}

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

    # Verificação de correspondência de nome
    verificacao_nome = verificar_correspondencia_nome(
    nome_passageiro, dados.get("name", "")
    )

    if not verificacao_nome["corresponde"]:
    return {
    "valido": False,
    "erro": "Nome do passageiro não corresponde ao CPF",
    "nome_no_cpf": dados.get("name"),
    "score_similaridade": verificacao_nome["score"],
    "sugestao": f"Verifique se o nome correto é: {dados.get('name')}",
    }

    # Verificação de idade (menores podem precisar de autorização)
    ano_nascimento = dados.get("year", 0)
    idade = datetime.now().year - ano_nascimento
    menor_de_idade = idade < 18

    return {
    "valido": True,
    "passageiro": {
    "cpf": cpf_limpo,
    "nome": dados.get("name"),
    "nomeUpper": dados.get("nameUpper"),
    "dataNascimento": dados.get("birthDate"),
    "genero": dados.get("gender"),
    "menorDeIdade": menor_de_idade,
    },
    "alertas": (
    ["Passageiro menor de idade -- autorização pode ser necessária"]
    if menor_de_idade
    else []
    ),
    }

def emitir_bilhete(
    cpf: str,
    nome_passageiro: str,
    origem: str,
    destino: str,
    data_viagem: str,
    tipo: str = "aereo",
) -> dict:
    """Emite bilhete após validação completa."""

    # Valida passageiro
    validacao = validar_passageiro(cpf, nome_passageiro, data_viagem)

    if not validacao["valido"]:
    return {"emitido": False, "erro": validacao["erro"]}

    passageiro = validacao["passageiro"]

    # Gera localizador
    import hashlib
    hash_base = f"{cpf}-{origem}-{destino}-{data_viagem}-{datetime.now().isoformat()}"
    localizador = hashlib.md5(hash_base.encode()).hexdigest()[:6].upper()

    bilhete = {
    "emitido": True,
    "localizador": localizador,
    "tipo": tipo,
    "passageiro": passageiro["nomeUpper"],
    "cpf": passageiro["cpf"][:3] + ".***.***-" + passageiro["cpf"][-2:],
    "origem": origem,
    "destino": destino,
    "dataViagem": data_viagem,
    "emitidoEm": datetime.now().strftime("%d/%m/%Y %H:%M"),
    "alertas": validacao.get("alertas", []),
    }

    return bilhete

# Exemplo de uso
if __name__ == "__main__":
    # Emitir passagem aérea
    resultado = emitir_bilhete(
    cpf="123.456.789-09",
    nome_passageiro="João da Silva",
    origem="GRU",
    destino="GIG",
    data_viagem="15/10/2025",
    tipo="aereo",
    )
    print(json.dumps(resultado, indent=2, ensure_ascii=False))

    # Emitir passagem rodoviária
    resultado_rod = emitir_bilhete(
    cpf="987.654.321-00",
    nome_passageiro="Maria Oliveira",
    origem="São Paulo - Tietê",
    destino="Rio de Janeiro - Novo Rio",
    data_viagem="20/10/2025",
    tipo="rodoviario",
    )
    print(json.dumps(resultado_rod, indent=2, ensure_ascii=False))

from collections import defaultdict
from datetime import datetime, timedelta

# Registro de compras por CPF
compras_por_cpf = defaultdict(list)

# Limites configuráveis
LIMITE_PASSAGENS_DIA = 5
LIMITE_PASSAGENS_MES = 20
LIMITE_MESMO_TRECHO_DIA = 2

def verificar_limites(cpf: str, origem: str, destino: str) -> dict:
    """Verifica se o CPF atingiu algum limite de compra."""
    cpf_limpo = re.sub(r"\D", "", cpf)
    agora = datetime.now()

    compras = compras_por_cpf.get(cpf_limpo, [])

    # Compras nas últimas 24 horas
    compras_24h = [c for c in compras if agora - c["data"] < timedelta(hours=24)]

    # Compras no mês atual
    compras_mes = [c for c in compras if c["data"].month == agora.month]

    # Mesmo trecho nas últimas 24 horas
    mesmo_trecho_24h = [
    c for c in compras_24h
    if c["origem"] == origem and c["destino"] == destino
    ]

    alertas = []

    if len(compras_24h) >= LIMITE_PASSAGENS_DIA:
    return {
    "permitido": False,
    "erro": f"Limite de {LIMITE_PASSAGENS_DIA} passagens por dia atingido",
    }

    if len(compras_mes) >= LIMITE_PASSAGENS_MES:
    return {
    "permitido": False,
    "erro": f"Limite de {LIMITE_PASSAGENS_MES} passagens por mês atingido",
    }

    if len(mesmo_trecho_24h) >= LIMITE_MESMO_TRECHO_DIA:
    return {
    "permitido": False,
    "erro": f"Limite de {LIMITE_MESMO_TRECHO_DIA} passagens para o mesmo trecho por dia",
    }

    if len(compras_24h) >= LIMITE_PASSAGENS_DIA - 1:
    alertas.append("Próximo do limite diário de compras")

    return {"permitido": True, "alertas": alertas}