#!
/usr/bin/env python3
import os
import sys
import json
from datetime import datetime
# Bibliotecas de IA e Agentes
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
from [Link] import BaseTool
from langchain_openai import ChatOpenAI
# Estruturas de saída (Pydantic)
from pydantic import BaseModel, Field
from typing import List, Optional
# Bibliotecas de Arquivo, OCR e Excel
import pytesseract
from pdf2image import convert_from_path
from docxtpl import DocxTemplate
import openpyxl
from [Link] import Font, Alignment, PatternFill
from PIL import Image # para abrir imagens para o Tesseract
# --- CONFIGURAÇÃO ---
# NÃO deixe chaves hardcoded em produção
OPENAI_API_KEY = [Link]("OPENAI_API_KEY") or "sk-proj-
KAYfg_XD-0inKGUriqSHToEpIcwOE4M6MINhIT3YZtrmxBZrsv6pRscmpRp9QXDa__UJ
81pfbST3BlbkFJ8cOWsjtwFKbWDLccZ9KV7UMR42NW_h9ZiBSRG34HrTtkTs_X5xHcCK
GqPLqx8mMvApfFtqAaIA"
SERPER_API_KEY = [Link]("SERPER_API_KEY") or
"673cfd2e614b677a3330d1875fadf9baab5db042"
if not OPENAI_API_KEY or not SERPER_API_KEY:
print("ERRO: Variáveis de ambiente OPENAI_API_KEY e
SERPER_API_KEY são obrigatórias.")
[Link](1)
[Link]["OPENAI_API_KEY"] = OPENAI_API_KEY
[Link]["SERPER_API_KEY"] = SERPER_API_KEY
# Caminhos do Sistema
BASE_DIR = [Link]("~/Advocacia_AI")
TEMPLATE_PATH = [Link](BASE_DIR, "Templates",
"template_master.docx")
OUTPUT_DIR = [Link](BASE_DIR, "Outputs")
[Link](OUTPUT_DIR, exist_ok=True)
if not [Link](TEMPLATE_PATH):
print(f"ERRO: Template Word não encontrado em: {TEMPLATE_PATH}")
[Link](1)
# Modelo Cérebro (GPT-4o)
llm_gpt4 = ChatOpenAI(model="gpt-4o", temperature=0.2)
search_tool = SerperDevTool()
# --- MODELOS Pydantic PARA SAÍDAS ESTRUTURADAS ---
class DiagnosticoModelo(BaseModel):
tipo_acao: str = Field(
...,
description="Tipo de ação: ex: 'Reclamação Trabalhista',
'Rescisão Indireta', 'Acidente de Trabalho'.",
)
tem_cct: bool = Field(..., description="Se há menção a CCT.")
multiplas_empresas: bool = Field(..., description="Se há
indícios de grupo econômico/terceirização.")
tem_doenca_insalubridade: bool = Field(..., description="Se há
doença, acidente ou insalubridade.")
resumo_fatos: str = Field(..., description="Resumo cronológico,
apenas com fatos constantes dos documentos.")
dados_contrato: str = Field(
...,
description="Admissão, Demissão, Salário, Função e outros
dados contratuais relevantes.",
)
class VerbaModelo(BaseModel):
nome: str = Field(..., description="Nome da verba, ex: 'Férias
vencidas', 'Horas extras'.")
parametro: str = Field(
...,
description="Base de cálculo, período, quantidade de meses,
percentuais, etc.",
)
obs: Optional[str] = Field(
default="",
description="Observações adicionais sobre a verba, se
houver.",
)
class CalculoModelo(BaseModel):
dados_contrato: str = Field(
...,
description="Descrição textual dos dados de contrato
(admissão, demissão, salário, função).",
)
verbas: List[VerbaModelo] = Field(
...,
description="Lista de verbas a calcular, com parâmetros e
observações.",
)
# --- FERRAMENTAS ---
class UniversalReaderTool(BaseTool):
name: str = "Leitor Universal de Pasta"
description: str = "Lê TODO o conteúdo da pasta do caso (PDFs,
Imagens, TXT). Não inventa conteúdo; só transcreve."
def _run(self, folder_path: str) -> str:
texto_total = ""
if not [Link](folder_path):
return "ERRO: Pasta inexistente."
print(f"\n>>> Lendo arquivos em: {folder_path}")
# Percorre subpastas também
for root, dirs, files in [Link](folder_path):
for file in files:
if [Link]().endswith((".pdf", ".jpg", ".png",
".jpeg", ".txt")):
caminho = [Link](root, file)
try:
print(f" - Processando: {file}")
texto_total += f"\n--- ARQUIVO: {file} ---
\n"
if [Link]().endswith(".pdf"):
images = convert_from_path(caminho)
for img in images:
texto_total +=
pytesseract.image_to_string(
img, lang="por"
) + "\n"
elif [Link]().endswith(".txt"):
with open(caminho, "r",
encoding="utf-8", errors="ignore") as f:
texto_total += [Link]()
else:
# Abre explicitamente a imagem para OCR
img = [Link](caminho)
texto_total +=
pytesseract.image_to_string(
img, lang="por"
) + "\n"
except Exception as e:
print(f" [!] Erro lendo {file}: {e}")
if not texto_total:
return "Nenhum texto legível encontrado."
# Limite alto de contexto – ideal seria chunking, mas aqui
só truncamos
return texto_total[:100000]
reader_tool = UniversalReaderTool()
# --- PROMPTS E AGENTES DINÂMICOS ---
PROMPT_DIAGNOSTICO = """
Você é o Chefe de Triagem.
TAREFA:
- Leia os documentos brutos fornecidos pelo Leitor Universal de
Pasta.
- NÃO invente fatos que não estejam nos documentos.
- NÃO estime datas/valores se não constarem de forma minimamente
clara.
- Gere um DIAGNÓSTICO ESTRUTURADO.
RESPONDA EXCLUSIVAMENTE EM JSON VÁLIDO, SEM TEXTO FORA DO OBJETO.
Você deve analisar:
1. Há indícios de mais de uma empresa (Grupo Econômico/
Terceirização)?
2. Há menção a Convenção Coletiva (CCT)?
3. Há relato de doença, acidente ou insalubridade?
4. Qual o tipo de rescisão (Sem Justa Causa, Pedido, Indireta etc.)?
SAÍDA OBRIGATÓRIA (JSON PURO, EXEMPLO DE ESTRUTURA):
{
"tipo_acao": "Reclamação Trabalhista / Rescisão Indireta /
Acidente",
"tem_cct": true,
"multiplas_empresas": false,
"tem_doenca_insalubridade": false,
"resumo_fatos": "Resumo cronológico detalhado apenas com fatos
constantes dos documentos...",
"dados_contrato": "Admissão, Demissão, Salário, Função..."
}
"""
PROMPT_CALCULISTA = """
Você é um Calculista Trabalhista.
TAREFA:
- A partir do diagnóstico, dos fatos e dos documentos, gere um JSON
ESTRUTURADO para cálculo.
- NÃO invente verbas que não tenham amparo mínimo nos fatos/
documentos.
- Use os campos exatamente como no schema fornecido.
SCHEMA (EXEMPLO DE ESTRUTURA – SIGA OS MESMOS NOMES DE CAMPOS):
{
"dados_contrato": "Texto com datas de admissão/demissão, salário,
função etc.",
"verbas": [
{
"nome": "Aviso Prévio Indenizado",
"parametro": "1 salário mensal",
"obs": "Se aplicável."
},
{
"nome": "Férias Vencidas",
"parametro": "1 período aquisitivo completo",
"obs": "Com 1/3 constitucional."
}
]
}
RESPONDA EXCLUSIVAMENTE EM JSON VÁLIDO, SEM TEXTOS EXTERNOS.
"""
PROMPT_REDATOR = """
Você é um Redator Sênior especializado em Direito do Trabalho.
TAREFA:
- Com base em TODOS os relatórios anteriores (diagnóstico, pesquisa,
CCT, perícia, narrativa, cálculo),
redija a Petição Inicial TRABALHISTA completa em Markdown.
REGRAS:
- NÃO invente fatos que não estejam no resumo dos fatos ou nos
documentos.
- Se algum dado for incerto (datas exatas, valores), use expressões
moderadas: "cerca de", "aproximadamente", "ao redor de".
- Estruture a inicial com: Preâmbulo, Dos Fatos, Do Direito, Dos
Pedidos, Valor da Causa, Protesta por Provas, etc.
- Utilize linguagem respeitosa e técnica, sem adjetivações gratuitas
contra a parte contrária.
- Trate os pedidos de forma clara, numerados, relacionando-os com as
verbas do cálculo.
SAÍDA:
- Apenas o texto completo da Petição Inicial em Markdown.
"""
def criar_agente(
role,
goal,
tools=None,
backstory="Especialista jurídico focado em alta performance.",
):
if tools is None:
tools = []
return Agent(
role=role,
goal=goal,
backstory=backstory,
tools=tools,
llm=llm_gpt4,
verbose=True,
)
# --- FUNÇÕES DE EXPORTAÇÃO ---
def gerar_planilha_calculo(dados_json, caminho_saida):
"""Gera Excel formatado a partir do JSON do Calculista (já
validado pelo modelo Pydantic)."""
try:
if isinstance(dados_json, str):
dados_json = dados_json.replace("```json",
"").replace("```", "").strip()
dados = [Link](dados_json)
else:
dados = dados_json
except Exception as e:
print("Erro ao processar JSON do calculista. Salvando bruto
para debug.")
try:
with open(caminho_saida.replace(".xlsx", "_bruto.json"),
"w", encoding="utf-8") as f:
[Link](str(dados_json))
except Exception:
pass
print(f"Detalhe do erro: {e}")
return
wb = [Link]()
ws = [Link]
[Link] = "Parâmetros"
# Estilos
header_font = Font(bold=True, color="FFFFFF")
header_fill = PatternFill(start_color="1F4788",
end_color="1F4788", fill_type="solid")
# Dados Contratuais
ws["A1"] = "DADOS DO CONTRATO"
ws["A1"].fill = header_fill
ws["A1"].font = header_font
row = 2
# Aqui assumimos que "dados_contrato" é um texto único
ws[f"A{row}"] = "DADOS_CONTRATO"
ws[f"B{row}"] = str([Link]("dados_contrato", ""))
row += 2
# Tabela de Verbas
headers = ["VERBA", "PARÂMETRO", "OBSERVAÇÃO", "VALOR (R$)"]
for col_num, header in enumerate(headers, 1):
cell = [Link](row=row, column=col_num, value=header)
[Link] = header_fill
[Link] = header_font
row += 1
for verba in [Link]("verbas", []):
[Link](row=row, column=1, value=[Link]("nome", ""))
[Link](row=row, column=2, value=[Link]("parametro", ""))
[Link](row=row, column=3, value=[Link]("obs", ""))
row += 1
# Ajuste Colunas
ws.column_dimensions["A"].width = 30
ws.column_dimensions["B"].width = 30
ws.column_dimensions["C"].width = 40
ws.column_dimensions["D"].width = 18
[Link](caminho_saida)
# --- FLUXO PRINCIPAL ---
def rodar_banca():
# 1. Input via linha de comando
pasta_input = input("Digite o caminho COMPLETO da pasta do
cliente: ").strip()
if not pasta_input or not [Link](pasta_input):
print("Caminho inválido. Encerrando.")
return
# 2. Diagnóstico (O Cérebro Central)
print("\n>>> FASE 1: TRIAGEM E DIAGNÓSTICO...")
triador = criar_agente(
"Triador Chefe",
"Ler pasta e Diagnosticar Caso a partir dos documentos, sem
inventar fatos.",
[reader_tool],
)
task_diag = Task(
description=f"Leia integralmente a pasta do cliente em
'{pasta_input}' usando o Leitor Universal de Pasta.
{PROMPT_DIAGNOSTICO}",
agent=triador,
expected_output="JSON de Diagnóstico conforme schema.",
output_json=DiagnosticoModelo,
)
crew_diag = Crew(agents=[triador], tasks=[task_diag],
process=[Link])
res_diag = crew_diag.kickoff()
diagnostico_dict = {}
try:
if task_diag.output.json_dict:
diagnostico_dict = task_diag.output.json_dict
elif task_diag.[Link]:
diag_raw = (
task_diag.[Link]("```json",
"").replace("```", "").strip()
)
diagnostico_dict = [Link](diag_raw)
print(
"\n>>> DIAGNÓSTICO IA:",
[Link](diagnostico_dict, indent=2,
ensure_ascii=False),
)
except Exception as e:
print("ALERTA: Falha no JSON de diagnóstico. Usando fluxo
padrão reduzido.")
print(f"Detalhe do erro: {e}")
diagnostico_dict = {
"tipo_acao": "Reclamação Trabalhista",
"tem_cct": False,
"multiplas_empresas": False,
"tem_doenca_insalubridade": False,
"resumo_fatos": task_diag.[Link]
if task_diag.output and task_diag.[Link]
else "",
"dados_contrato": "",
}
# 3. Montagem da Equipe Dinâmica
agentes = []
tarefas = []
# Agentes Fixos
pesquisador = criar_agente(
"Pesquisador",
"Buscar jurisprudência e teses atualizadas em Direito do
Trabalho.",
[search_tool],
)
narrador = criar_agente(
"Narrador",
"Construir narrativa DOS FATOS com base apenas nos fatos
identificados.",
)
# Tarefa Base Pesquisa (contexto: diagnóstico)
task_pesquisa_base = Task(
description=(
f"Com base no seguinte diagnóstico:
{[Link](diagnostico_dict, ensure_ascii=False)} "
f"pesquise jurisprudência recente para:
{diagnostico_dict.get('tipo_acao', 'Reclamação Trabalhista')}. "
f"Fatos principais:
{diagnostico_dict.get('resumo_fatos', '')}. "
"Inclua teses sobre valores usualmente arbitrados, sem
fixar quantias exatas."
),
agent=pesquisador,
expected_output="Dossiê Jurídico com jurisprudência e teses,
em texto organizado.",
)
[Link](task_pesquisa_base)
[Link](pesquisador)
# --- CONDICIONAIS ---
# A. SINDICALISTA (Se tem CCT)
task_sindicalista = None
if diagnostico_dict.get("tem_cct"):
sindicalista = criar_agente(
"Sindicalista",
"Analisar cláusulas de Convenções Coletivas
aplicáveis.",
[reader_tool, search_tool],
)
task_sindicalista = Task(
description=(
"Analise a eventual CCT presente na pasta (via
Leitor Universal) "
"e, se necessário, complemente com busca na web. "
"Liste de forma objetiva: Piso Salarial, Auxílio
Alimentação, Quebra de Caixa, "
"Adicionais e Multas convencionais aplicáveis."
),
agent=sindicalista,
expected_output="Relatório de Cláusulas CCT
aplicáveis.",
)
[Link](task_sindicalista)
[Link](sindicalista)
# B. MÉDICO/PERITO (Se tem Doença/Insalubridade)
task_medico = None
if diagnostico_dict.get("tem_doenca_insalubridade"):
medico = criar_agente(
"Perito Médico",
"Fundamentar nexo causal e insalubridade à luz da
legislação e medicina do trabalho.",
[search_tool],
)
task_medico = Task(
description=(
f"Com base nos fatos:
{diagnostico_dict.get('resumo_fatos', '')}, "
"pesquise nexo causal, EPIs obrigatórios para a
função descrita, "
"e fundamentos para pedido de perícia técnica, sem
exageros não amparados em literatura."
),
agent=medico,
expected_output="Parecer Técnico Preliminar sobre
doença/acidente/insalubridade.",
)
[Link](task_medico)
[Link](medico)
# C. GRUPO ECONÔMICO (Adiciona tarefa ao Pesquisador)
task_responsabilidade = None
if diagnostico_dict.get("multiplas_empresas"):
task_responsabilidade = Task(
description=(
f"Considerando os fatos:
{diagnostico_dict.get('resumo_fatos', '')}, "
"pesquise tese de responsabilidade solidária/
subsidiária, grupo econômico e eventual fraude na terceirização, "
"com base na CLT, Súmulas do TST e jurisprudência
recente."
),
agent=pesquisador, # Reusa o pesquisador
expected_output="Tese de Responsabilidade com
fundamentos e precedentes.",
)
[Link](task_responsabilidade)
# Tarefa Narrativa (Fixa – contexto: diagnóstico)
task_narrativa = Task(
description=(
"Escreva a seção DOS FATOS com base EXCLUSIVA no
seguinte resumo de fatos, "
"sem inventar dados: "
f"{diagnostico_dict.get('resumo_fatos', '')}. "
"Use tom firme, mas respeitoso, destacando a conduta da
empresa apenas nos pontos descritos."
),
agent=narrador,
expected_output="Texto Narrativo dos Fatos, coerente com o
resumo fornecido.",
)
[Link](task_narrativa)
[Link](narrador)
# 4. Finalização (Cálculo e Redação)
calculista = criar_agente(
"Calculista",
"Gerar JSON de cálculo trabalhista estruturado, com base em
fatos e diagnóstico.",
)
redator = criar_agente(
"Redator Sênior",
"Escrever a Petição Inicial completa em Markdown, com base
em todos os relatórios.",
)
task_calculo = Task(
description=(
f"Utilize o seguinte diagnóstico e resumo de fatos para
montar o JSON de cálculo: "
f"{[Link](diagnostico_dict, ensure_ascii=False)}. "
f"Considere também, se existentes, os relatórios da
pesquisa jurídica, CCT e perícia. "
f"{PROMPT_CALCULISTA}"
),
agent=calculista,
expected_output="JSON de Cálculo Estruturado conforme
schema.",
output_json=CalculoModelo, # saída estruturada [web:1]
[web:20]
)
# Tarefa do Redator, recebendo contexto explícito das demais
tasks [web:23]
# (se alguma task condicional não existir, simplesmente não
entra no contexto)
contexto_redator = [task_diag, task_pesquisa_base,
task_narrativa, task_calculo]
if task_sindicalista:
contexto_redator.append(task_sindicalista)
if task_medico:
contexto_redator.append(task_medico)
if task_responsabilidade:
contexto_redator.append(task_responsabilidade)
task_redacao = Task(
description=PROMPT_REDATOR,
agent=redator,
expected_output="Petição Final em Markdown.",
context=contexto_redator, # garante que o output anterior
entra como contexto [web:23]
)
[Link]([task_calculo, task_redacao])
[Link]([calculista, redator])
# 5. Execução
print(f"\n>>> FASE 2: EXECUÇÃO ({len(agentes)} Agentes
Ativos)...")
crew_final = Crew(
agents=agentes, tasks=tarefas, process=[Link],
verbose=True
)
res_final = crew_final.kickoff()
# 6. Geração de Arquivos
print("\n>>> FASE 3: GERANDO ARQUIVOS...")
timestamp = [Link]().strftime("%Y%m%d_%H%M")
# A. Excel (usando saída estruturada do calculista)
try:
output_calc_json = None
if task_calculo.output.json_dict:
output_calc_json = task_calculo.output.json_dict
elif task_calculo.[Link]:
output_calc_json = (
task_calculo.[Link]("```json", "")
.replace("```", "")
.strip()
)
path_xlsx = [Link](OUTPUT_DIR,
f"Calculo_{timestamp}.xlsx")
gerar_planilha_calculo(output_calc_json, path_xlsx)
print(f"✅ Excel: {path_xlsx}")
except Exception as e:
print(f"Erro Excel: {e}")
# B. Word (Petição)
try:
doc = DocxTemplate(TEMPLATE_PATH)
# res_final.raw é o output da última tarefa (Redator) em
CrewOutput [web:1]
conteudo_peticao = (
res_final.raw if hasattr(res_final, "raw") else
task_redacao.[Link]
)
[Link]({"CONTEUDO_PETICAO": conteudo_peticao})
path_docx = [Link](OUTPUT_DIR,
f"Inicial_{timestamp}.docx")
[Link](path_docx)
print(f"✅ Word: {path_docx}")
# Abre arquivos (macOS/Linux; em Windows você usaria start
ou equivalente)
try:
[Link](f"open '{path_xlsx}'")
except Exception:
pass
try:
[Link](f"open '{path_docx}'")
except Exception:
pass
[Link]("Sucesso", "Processo Concluído! Confira os
arquivos gerados.")
except Exception as e:
[Link]("Erro Word", str(e))
if __name__ == "__main__":
rodar_banca()