pyjuris 1.0.0

Normalização de processos judiciais brasileiros — multi-fonte, multi-output

etl-lops

Pacote Python para normalização de processos judiciais brasileiros.

Recebe payloads brutos de fontes externas (Escavador, futuramente Solucionare) e entrega JSON padronizado para consumidores internos do ecossistema LOPS (lops_api, lops_agents).

• Stateless — funções puras, sem estado interno
• Idempotente — mesma entrada, mesma saída
• Sem descartes — dados incompletos entram com flags, nunca são perdidos
• Sem IA — transformações explícitas e auditáveis por regras

---

Instalação

# Core (sem dependências externas)
pip install -e .

# Com CLI (flask, pymongo, rich)
pip install -e ".[cli]"

# Com testes
pip install -e ".[dev]"

Requisito: Python 3.11+

---

Início rápido

from etl_lops import normalizar_processo

with open("processo.json") as f:
    payload = json.load(f)

result = normalizar_processo(payload)
# → dict padronizado pronto para persistir no lops_api

---

API

normalizar_processo

Ponto de entrada principal. Transforma payload bruto em JSON normalizado.

from etl_lops import normalizar_processo

result = normalizar_processo(payload)

# Explícito (equivalente ao padrão)
result = normalizar_processo(
    payload,
    input_template="escavador",   # fonte dos dados
    output_template="lops_api",   # formato de saída
    enrich_orgaos=True,           # estrutura orgao_julgador automaticamente
)

# Sem enriquecimento de órgão (bulk processing sem necessidade do campo)
result = normalizar_processo(payload, enrich_orgaos=False)

Saída:

{
  "numero_cnj": "1004370-10.2025.8.26.0590",
  "tipo_principal": "RECURSO",
  "fase_atual": "SEGUNDO_GRAU",
  "status_geral": "ATIVO",
  "total_instancias": 2,
  "segredo_justica": false,
  "capa_disponivel": true,
  "tem_audiencia": false,
  "tem_multiplas_instancias_ativas": true,
  "fenomenos_detectados": [
    "inversao_polo:08057867833",
    "multiplas_instancias_ativas"
  ],
  "ano_inicio": 2025,
  "data_inicio": "2025-04-11",
  "data_ultima_movimentacao": "2025-10-02",
  "data_ultima_verificacao": "2025-10-08T12:30:39Z",
  "polo_ativo": "Thial Felix da Silva",
  "polo_passivo": "Banco Agibank S.A",
  "instancias": [ ... ],
  "schema_version": "1.0",
  "_etl": {
    "processado_em": "2026-03-12T21:00:00Z",
    "versao_etl": "1.0.0",
    "fonte_origem": "escavador"
  }
}

---

resumo

Texto legível para humanos — útil em logs e CLIs.

from etl_lops import resumo

print(resumo(result))

────────────────────────────────────────────────────────
  Processo   1004370-10.2025.8.26.0590
  Tipo       RECURSO
  Fase       SEGUNDO_GRAU
  Status     ATIVO
  Polo ativo   Thial Felix da Silva
  Polo passivo Banco Agibank S.A
  Instâncias 2 ⚠ múltiplas ativas
  Distribuição        2025-04-11
  Última movimentação 2025-10-02

  Alertas:
    ⚠  Polo ativo e passivo invertidos entre graus do processo
    ⚠  Processo ativo simultaneamente em mais de uma instância
────────────────────────────────────────────────────────

---

should_reprocess

Decide se um processo deve ser reprocessado comparando datas de movimentação.

from etl_lops import should_reprocess

should_reprocess("2025-09-01", "2025-10-02")  # → True  (incoming mais recente)
should_reprocess("2025-10-02", "2025-10-02")  # → False (mesma data)
should_reprocess(None, "2025-10-02")          # → True  (sem dado armazenado)

---

descrever_fenomeno / descrever_fenomenos

Converte códigos de fenômenos em descrições legíveis em português.

from etl_lops import descrever_fenomeno, descrever_fenomenos

descrever_fenomeno("inversao_polo:12345678900")
# → "Polo ativo e passivo invertidos entre graus do processo"

descrever_fenomenos(result["fenomenos_detectados"])
# → [{"codigo": "inversao_polo:...", "descricao": "..."}, ...]

Fenômenos disponíveis:

Código	Descrição
cnj_invalido	Número CNJ fora do formato padrão
sem_fontes	Nenhuma fonte judicial encontrada
multiplas_instancias_ativas	Processo ativo em mais de uma instância
duplicata_ingestao	Fonte duplicada detectada na mesma instância
redistribuicao	Redistribuição identificada entre fontes do mesmo grau
crawl_duplicado	Coleta duplicada da mesma fonte
recursos_distintos	Recursos distintos no mesmo grau
inversao_polo:<doc>	Polo ativo/passivo invertidos entre graus

---

get_orgaos / parsear_orgao / get_tribunais

Referência de órgãos julgadores brasileiros — 32k+ órgãos de 64 tribunais, bundled com o pacote.

from etl_lops import get_tribunais, get_orgaos, parsear_orgao

# Lista todos os tribunais disponíveis
get_tribunais()
# → ["STJ", "TJSP", "TJRS", "TRF1", "TRT1", ...]  (64 tribunais)

# Órgãos estruturados de um tribunal
get_orgaos("TJRS")[9]
# → {
#     "nome_original": "10ª Vara Criminal do Foro Central - Porto Alegre",
#     "nome_normalizado": "10ª Vara Criminal do Foro Central - Porto Alegre",
#     "encoding_status": "ok",
#     "tipo": "VARA",
#     "numero": 10,
#     "especialidade": "CRIMINAL",
#     "localidade": "Porto Alegre",
#     "cargo": None,
#     "nome_pessoa": None
#   }

# Parsing avulso de qualquer string
parsear_orgao("3ª Vara da Fazenda Pública - Belo Horizonte")
# → {"tipo": "VARA", "numero": 3, "especialidade": "FAZENDA_PUBLICA",
#    "localidade": "Belo Horizonte", "cargo": None, "nome_pessoa": None}

parsear_orgao("GABINETE DA MINISTRA NANCY ANDRIGHI")
# → {"tipo": "GABINETE", "numero": None, "especialidade": None,
#    "localidade": None, "cargo": "Ministra", "nome_pessoa": "Nancy Andrighi"}

O campo orgao_julgador_info é adicionado automaticamente em cada instância pelo normalizar_processo (controlado por enrich_orgaos=True).

Tipos de órgão: VARA, CAMARA, TURMA, TURMA_RECURSAL, GABINETE, JUIZADO_ESPECIAL, JUIZADO_ESPECIAL_FEDERAL, JUIZADO, CEJUSC, UNIDADE_JURISDICIONAL, NUCLEO, GRUPO, SUBSECAO, SECAO, PRESIDENCIA, CORREGEDORIA, DIRETORIA, SECRETARIA, PLENARIO, OUTRO

Especialidades: CIVEL, CRIMINAL, FAZENDA_PUBLICA, EXECUCAO_FISCAL, PREVIDENCIARIO, FAMILIA, ORFAOS_SUCESSOES, INFANCIA_JUVENTUDE, CONSUMIDOR, TRABALHISTA, ACIDENTES_TRABALHO, ELEITORAL, MILITAR, PLANTAO, AMBIENTAL, EMPRESARIAL, PROPRIEDADE_INTELECTUAL, RECUPERACAO_JUDICIAL, AGRARIO, TRIBUTARIO

---

CLI

# Pipe: stdin → stdout
cat processo.json | etl-lops

# Resumo legível
cat processo.json | etl-lops --resumo

# Arquivo → arquivo
etl-lops -i processo.json -o resultado.json

# Batch: diretório inteiro
etl-lops -i ./jsons/ -o ./output/

# Templates explícitos
etl-lops --input-template escavador --output-template lops_api < processo.json

# Servidor HTTP local (porta 5045)
etl-lops --api

Servidor HTTP (--api):

Endpoint	Método	Descrição
POST /transform	JSON body	Normaliza um processo
POST /upload	multipart	Upload de arquivo JSON
GET /	—	Health check

---

Schema de saída

Processo (topo)

Campo	Tipo	Valores
numero_cnj	string | null	Formato NNNNNNN-DD.AAAA.J.TT.OOOO
tipo_principal	enum	ACAO_ORIGINAL, RECURSO, CUMPRIMENTO_SENTENCA, RECURSO_INTERMEDIARIO, DESCONHECIDO
fase_atual	enum	PRIMEIRO_GRAU, SEGUNDO_GRAU, TERCEIRO_GRAU, ENCERRADO, DESCONHECIDO
status_geral	enum	ATIVO, INATIVO
total_instancias	integer	—
segredo_justica	boolean	—
capa_disponivel	boolean	—
tem_audiencia	boolean	—
tem_multiplas_instancias_ativas	boolean	—
fenomenos_detectados	string[]	Ver tabela de fenômenos
ano_inicio	integer | null	—
data_inicio	date | null	YYYY-MM-DD
data_ultima_movimentacao	date | null	YYYY-MM-DD
data_ultima_verificacao	datetime | null	ISO 8601
polo_ativo	string | null	Nome da parte ativa principal
polo_passivo	string | null	Nome da parte passiva principal
instancias	Instancia[]	Ver abaixo
schema_version	string	"1.0"
_etl	object	processado_em, versao_etl, fonte_origem

Instância

Campo	Tipo	Valores
fonte_id	integer | null	—
grau	enum	1, 2, 3
grau_formatado	enum	Primeiro Grau, Segundo Grau, Terceiro Grau
tipo	enum	Mesmo que tipo_principal
tribunal	object	sigla, nome, sistema
classe	string | null	Classe normalizada pelo ETL
classe_raw	string | null	Classe original da fonte
assunto_principal	string | null	—
assunto_path	string | null	Ex: DIREITO DO CONSUMIDOR > Bancários
assuntos	Assunto[]	{id, nome, path}
orgao_julgador	string | null	String original
orgao_julgador_info	OrgaoJulgador | null	Parsing estruturado (ver acima)
valor_causa	number | null	Em BRL
moeda	enum	BRL, null
data_inicio	date | null	—
data_ultima_movimentacao	date | null	—
status	enum	ATIVO, INATIVO
arquivado	boolean | null	—
capa_disponivel	boolean	—
partes	Parte[]	Ver abaixo
audiencias	array	—

Parte

Campo	Tipo	Valores
nome	string | null	—
polo	enum	ATIVO, PASSIVO, ADVOGADO, DESCONHECIDO
tipo	enum	FISICA, JURIDICA, DESCONHECIDO, null
tipo_pessoa	enum	FISICA, JURIDICA, DESCONHECIDO, null
documento	string | null	CPF ou CNPJ sem formatação
advogados	array	—

---

Arquitetura

payload bruto (Escavador | Solucionare)
    ↓ etl_lops/inputs/<template>.parse()
canonical dict          ← formato interno, nunca exposto diretamente
    ↓ etl_lops/outputs/<template>.serialize()
output final (lops_api | solucionare)
    ↓ pipeline.py (enrich_orgaos=True)
+ orgao_julgador_info em cada instância

O pipeline tem 6 camadas internas (aplicadas em inputs/escavador.py):

1. Sanitização — normaliza strings e datas antes de qualquer lógica
2. Classificação — mapeia classe processual para tipos canônicos
3. Deduplicação — remove fontes duplicadas, detecta redistribuições
4. Normalização — constrói objetos por instância judicial
5. Detecção de fenômenos — inversão de polo, múltiplas instâncias ativas
6. Transform — orquestra o pipeline e monta o canonical

Adicionar nova fonte de dados (input)

# 1. Criar etl_lops/inputs/nova_fonte.py
def parse(payload: dict) -> dict:
    """Converte payload bruto para canonical dict."""
    ...

# 2. Registrar em etl_lops/inputs/__init__.py
# 3. Usar: normalizar_processo(payload, input_template="nova_fonte")

Adicionar novo consumidor (output)

# 1. Criar etl_lops/outputs/novo_consumidor.py
def serialize(canonical: dict, etl_version: str) -> dict:
    """Converte canonical para formato do consumidor."""
    ...

# 2. Registrar em etl_lops/outputs/__init__.py
# 3. Usar: normalizar_processo(payload, output_template="novo_consumidor")

---

Testes

# Suite completa (117 testes)
pytest tests/ -v

# Smoke test com uso real
python scripts/testar_lib.py
python scripts/testar_lib.py --verboso

---

Scripts utilitários

Script	Descrição
scripts/testar_lib.py	Smoke test de uso real (não pytest)
scripts/sanear_orgaos.py	Gera data/orgaos_julgadores_v2.json para inspeção
scripts/eda_escavador.py	Análise exploratória de payloads no MongoDB
scripts/analyze_classe_score.py	Análise de cobertura de mapeamento de classes

---

Migrações de banco

Scripts em migrations/ para evolução do schema MongoDB/PostgreSQL:

python migrations/migration_0001_enums_e_dominios.py
python migrations/migration_0002_rename_e_tipos.py
python migrations/migration_0003_indices.py

Schema SQL completo em docs/schema_banco.sql.