Como Fintechs de Remessas Internacionais Podem Validar CPFs de Remetentes

import requests
from datetime import datetime
from typing import Optional

class ValidacaoRemessa:
    def __init__(self, api_key: str):
    self.api_key = api_key

    def validar_remetente(self, cpf: str, nome: str, nascimento: str) -> dict:
    """Valida o CPF do remetente brasileiro."""
    return self._validar_pessoa(cpf, nome, nascimento, "remetente")

    def validar_destinatario(self, cpf: str, nome: str, nascimento: str) -> dict:
    """Valida o CPF do destinatário brasileiro."""
    return self._validar_pessoa(cpf, nome, nascimento, "destinatario")

    def _validar_pessoa(self, cpf: str, nome: str, nascimento: str, papel: str) -> dict:
    cpf_limpo = cpf.replace(".", "").replace("-", "")
    response = requests.get(
    f"https://api.cpfhub.io/cpf/{cpf_limpo}",
    headers={"x-api-key": self.api_key},
    timeout=10,
    )

    if not response.json().get("success"):
    return {
    "papel": papel,
    "valido": False,
    "motivo": "CPF não encontrado na base oficial",
    }

    data = response.json()["data"]
    inconsistencias = []

    if data["nameUpper"] != nome.upper().strip():
    inconsistencias.append({
    "campo": "nome",
    "informado": nome.upper().strip(),
    "oficial": data["nameUpper"],
    })

    if data["birthDate"] != nascimento:
    inconsistencias.append({
    "campo": "data_nascimento",
    "informado": nascimento,
    "oficial": data["birthDate"],
    })

    return {
    "papel": papel,
    "valido": len(inconsistencias) == 0,
    "cpf": data["cpf"],
    "nome_oficial": data["name"],
    "inconsistencias": inconsistencias,
    "dados_para_registro": {
    "cpf": data["cpf"],
    "nome": data["name"],
    "nascimento": data["birthDate"],
    "genero": data["gender"],
    "validado_em": datetime.utcnow().isoformat(),
    },
    }

    def criar_ordem_remessa(
    self,
    remetente: dict,
    destinatario: dict,
    valor_usd: float,
    pais_destino: str,
    ) -> dict:
    """Cria uma ordem de remessa após validação."""
    validacoes = []

    # Validar remetente se for brasileiro
    if remetente.get("cpf"):
    val_rem = self.validar_remetente(
    remetente["cpf"], remetente["nome"], remetente["nascimento"]
    )
    validacoes.append(val_rem)
    if not val_rem["valido"]:
    return {"status": "BLOQUEADO", "motivo": "Remetente com dados inconsistentes"}

    # Validar destinatário se for brasileiro
    if destinatario.get("cpf"):
    val_dest = self.validar_destinatario(
    destinatario["cpf"], destinatario["nome"], destinatario["nascimento"]
    )
    validacoes.append(val_dest)
    if not val_dest["valido"]:
    return {"status": "BLOQUEADO", "motivo": "Destinatário com dados inconsistentes"}

    return {
    "status": "APROVADO",
    "ordem": {
    "valor_usd": valor_usd,
    "pais_destino": pais_destino,
    "validacoes": validacoes,
    "criada_em": datetime.utcnow().isoformat(),
    },
    }

def analisar_padroes_remessa(cpf: str, historico: list) -> dict:
    """
    Analisa padrões de remessas para um CPF específico.
    """
    alertas = []
    total_30_dias = sum(r["valor_usd"] for r in historico if r["dias_atras"] <= 30)
    total_90_dias = sum(r["valor_usd"] for r in historico if r["dias_atras"] <= 90)

    # Alerta: volume alto em 30 dias
    if total_30_dias > 10000:
    alertas.append({
    "tipo": "VOLUME_ALTO_30D",
    "valor": total_30_dias,
    "threshold": 10000,
    })

    # Alerta: muitos destinatários diferentes
    destinatarios = set(r["pais_destino"] for r in historico if r["dias_atras"] <= 90)
    if len(destinatarios) > 3:
    alertas.append({
    "tipo": "MULTIPLOS_DESTINOS",
    "quantidade": len(destinatarios),
    "paises": list(destinatarios),
    })

    # Alerta: fracionamento (structuring)
    remessas_pequenas = [r for r in historico if r["valor_usd"] < 500 and r["dias_atras"] <= 7]
    if len(remessas_pequenas) > 5:
    alertas.append({
    "tipo": "POSSIVEL_FRACIONAMENTO",
    "quantidade": len(remessas_pequenas),
    "total": sum(r["valor_usd"] for r in remessas_pequenas),
    })

    return {
    "cpf": cpf,
    "total_30d": total_30_dias,
    "total_90d": total_90_dias,
    "alertas": alertas,
    "risco": "ALTO" if len(alertas) >= 2 else "MEDIO" if alertas else "BAIXO",
    }