Pipeline de dados jurídicos para Fintechs: Arquitetura, código e boas práticas

from dataclasses import dataclass
from enum import Enum
import asyncio
import aiohttp

class RiscoJuridico(Enum):
    BAIXO = "low"
    MEDIO = "medium"
    ALTO = "high"
    CRITICO = "critical"

@dataclass
class PerfilRiscoJuridico:
    cpf_cnpj: str
    total_processos: int
    execucoes_fiscais: int
    execucoes_trabalhistas: int
    valor_total_causas: float
    tem_recuperacao_judicial: bool
    tem_mandado_prisao: bool
    nivel_risco: RiscoJuridico
    recomendacao: str


class PipelineKYCJuridico:

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.endpoint = "https://requests.prod.judit.io/requests"

    async def consultar_paralelo(self, cpf_cnpj: str) -> dict:
        """Faz todas as consultas necessárias em paralelo."""

        async with aiohttp.ClientSession() as session:
            # Executa 3 consultas simultaneamente (mais rápido)
            tarefas = [
                self._consultar(session, cpf_cnpj, "entity"),      # dados cadastrais
                self._consultar(session, cpf_cnpj, "lawsuits"),    # processos
                self._consultar(session, cpf_cnpj, "warrant"),     # mandados de prisão
            ]
            resultados = await asyncio.gather(*tarefas, return_exceptions=True)

        return {
            "entity": resultados[0],
            "lawsuits": resultados[1],
            "warrants": resultados[2]
        }

    async def _consultar(self, session, cpf_cnpj, response_type):
        payload = {"search_type": "cpf_cnpj", "search_key": cpf_cnpj, "response_type": response_type}
        async with session.post(self.endpoint, headers=self.headers, json=payload) as r:
            return await r.json()

    def calcular_risco(self, dados: dict) -> PerfilRiscoJuridico:
        """Calcula o perfil de risco jurídico a partir dos dados coletados."""

        processos = dados.get("lawsuits", {}).get("lawsuits", [])
        mandados = dados.get("warrants", {}).get("warrants", [])

        # Filtra por tipo de processo
        execucoes_fiscais = [p for p in processos if "Execução Fiscal" in p.get("subject", "")]
        execucoes_trab = [p for p in processos if p.get("court", "").startswith("TRT")]
        rec_judicial = any("Recuperação Judicial" in p.get("subject", "") for p in processos)
        valor_total = sum(p.get("value", 0) for p in processos)

        # Determina nível de risco
        if mandados or rec_judicial:
            risco = RiscoJuridico.CRITICO
            recomendacao = "BLOQUEAR - Mandado de prisão ou recuperação judicial ativa"
        elif len(execucoes_fiscais) > 5 or valor_total > 500_000:
            risco = RiscoJuridico.ALTO
            recomendacao = "REVISAR - Alto volume de execuções ou valor elevado"
        elif len(processos) > 10 or execucoes_trab:
            risco = RiscoJuridico.MEDIO
            recomendacao = "APROVAR COM LIMITE REDUZIDO"
        else:
            risco = RiscoJuridico.BAIXO
            recomendacao = "APROVAR"

        return PerfilRiscoJuridico(
            cpf_cnpj=dados["lawsuits"].get("search_key", ""),
            total_processos=len(processos),
            execucoes_fiscais=len(execucoes_fiscais),
            execucoes_trabalhistas=len(execucoes_trab),
            valor_total_causas=valor_total,
            tem_recuperacao_judicial=rec_judicial,
            tem_mandado_prisao=bool(mandados),
            nivel_risco=risco,
            recomendacao=recomendacao
        )

import requests
from typing import List

def ativar_monitoramento_carteira(cnpjs: List[str], webhook_url: str) -> dict:
    """
    Ativa monitoramento para toda a carteira de clientes.
    Cada movimentação relevante será notificada via webhook.
    """
    resultados = []

    for cnpj in cnpjs:
        payload = {
            "search_type": "cpf_cnpj",
            "search_key": cnpj,
            "webhook_url": webhook_url,
            "notification_config": {
                "on_new_lawsuit": True,
                "on_movement": True,
                "on_decision": True,
                # Filtra por tipos críticos para reduzir ruído
                "filter_subjects": [
                    "Execução Fiscal",
                    "Recuperação Judicial",
                    "Falência",
                    "Execução de Título Extrajudicial"
                ]
            }
        }
        response = requests.post(
            "https://tracking.prod.judit.io/tracking",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json=payload
        )
        resultados.append({"cnpj": cnpj, "status": response.status_code})

    ativados = sum(1 for r in resultados if r["status"] == 201)
    print(f"✅ {ativados}/{len(cnpjs)} clientes em monitoramento ativo")
    return {"ativados": ativados, "detalhes": resultados}


# Webhook que recebe alertas e reavalia risco do cliente
def reavaliar_risco_cliente(evento_webhook: dict):
    cnpj = evento_webhook["search_key"]
    novo_processo = evento_webhook.get("lawsuit", {})

    # Verifica se é processo crítico
    if "Recuperação Judicial" in novo_processo.get("subject", ""):
        bloquear_cliente(cnpj, motivo="Recuperação judicial detectada")
        notificar_time_risco(cnpj, urgencia="CRITICA")
    elif novo_processo.get("value", 0) > 100_000:
        reduzir_limite(cnpj, percentual=50)
        notificar_time_risco(cnpj, urgencia="ALTA")