Spaces:
Runtime error
Runtime error
| """ | |
| ESPECIALISTA 1 - Processador de Metadados | |
| Extração de informações estruturais do acórdão | |
| """ | |
| from typing import Dict, Any | |
| import logging | |
| from processors.base_processor import ProcessorBase | |
| logger = logging.getLogger(__name__) | |
| class ProcessorMetadados(ProcessorBase): | |
| """ | |
| Especialista 1: Extração de Metadados | |
| Responsabilidades: | |
| - Identificar tribunal e órgão julgador | |
| - Extrair número do processo | |
| - Identificar relator | |
| - Data de julgamento | |
| - Classe processual | |
| - Ramos jurídicos | |
| - Contato do tribunal | |
| """ | |
| def __init__(self, llm_model=None): | |
| super().__init__( | |
| specialist_id=1, | |
| specialist_name="Metadados", | |
| llm_model=llm_model | |
| ) | |
| def process(self, acordao_data: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Extrai metadados estruturais do acórdão | |
| Args: | |
| acordao_data: Dados do acórdão (ementa, integra, etc) | |
| Returns: | |
| Dict com metadados extraídos | |
| """ | |
| try: | |
| # Pré-processar dados | |
| preprocessed = self.pre_process( | |
| acordao_data.get('ementa', '') + ' ' + | |
| acordao_data.get('integra', '') | |
| ) | |
| # Extrair metadados | |
| metadados = { | |
| "tribunal": acordao_data.get('tribunal', 'TJPR'), | |
| "orgao_julgador": self._extrair_orgao_julgador( | |
| acordao_data.get('integra', '') | |
| ), | |
| "classe_processual": self._extrair_classe_processual( | |
| acordao_data.get('ementa', '') | |
| ), | |
| "numero_processo": acordao_data.get( | |
| 'numero_processo', | |
| '0000000-00.0000.0.00.0000' | |
| ), | |
| "relator": self._extrair_relator( | |
| acordao_data.get('integra', '') | |
| ), | |
| "data_julgamento": self._extrair_data_julgamento( | |
| acordao_data.get('integra', '') | |
| ), | |
| "ramo_especializado": self._extrair_ramo_principal( | |
| acordao_data.get('ementa', '') | |
| ), | |
| "ramos_secundarios": self._extrair_ramos_secundarios( | |
| acordao_data.get('ementa', '') | |
| ), | |
| "contato_tribunal": self._extrair_contato_tribunal( | |
| acordao_data.get('integra', '') | |
| ) | |
| } | |
| self.set_confidence(85) | |
| return metadados | |
| except Exception as e: | |
| self.add_error(f"Erro ao extrair metadados: {e}") | |
| raise | |
| def validate(self, result: Dict[str, Any]) -> bool: | |
| """Valida saída de metadados""" | |
| required_fields = [ | |
| 'tribunal', | |
| 'classe_processual', | |
| 'numero_processo', | |
| 'relator', | |
| 'ramo_especializado' | |
| ] | |
| return all(field in result for field in required_fields) | |
| def get_schema(self) -> Dict[str, Any]: | |
| """Retorna schema JSON para validação""" | |
| return { | |
| "type": "object", | |
| "properties": { | |
| "tribunal": {"type": "string"}, | |
| "orgao_julgador": {"type": "string"}, | |
| "classe_processual": {"type": "string"}, | |
| "numero_processo": {"type": "string", "pattern": "^[0-9]{7}-[0-9]{2}\\.[0-9]{4}\\.[0-9]\\.[0-9]{2}\\.[0-9]{4}$"}, | |
| "relator": {"type": "string"}, | |
| "data_julgamento": {"type": "string", "format": "date"}, | |
| "ramo_especializado": {"type": "string"}, | |
| "ramos_secundarios": {"type": "array"}, | |
| "contato_tribunal": {"type": "object"} | |
| }, | |
| "required": ["tribunal", "classe_processual", "numero_processo", "relator", "ramo_especializado"] | |
| } | |
| def _extrair_orgao_julgador(self, texto: str) -> str: | |
| """Extrai órgão julgador (Câmara, Turma, etc)""" | |
| keywords = ["câmara", "turma", "seção", "grupo"] | |
| for keyword in keywords: | |
| if keyword.lower() in texto.lower(): | |
| # Extrair contexto próximo | |
| idx = texto.lower().find(keyword) | |
| inicio = max(0, idx - 50) | |
| fim = min(len(texto), idx + 100) | |
| return texto[inicio:fim].strip() | |
| return "Câmara Cível" | |
| def _extrair_classe_processual(self, texto: str) -> str: | |
| """Extrai classe processual (Apelação, Agravo, etc)""" | |
| classes = [ | |
| "apelação cível", | |
| "agravo de instrumento", | |
| "embargos de declaração", | |
| "recurso extraordinário", | |
| "habeas corpus", | |
| "ação originária", | |
| "conflito de competência" | |
| ] | |
| texto_lower = texto.lower() | |
| for classe in classes: | |
| if classe in texto_lower: | |
| return classe.upper() | |
| return "APELAÇÃO CÍVEL" | |
| def _extrair_relator(self, texto: str) -> str: | |
| """Extrai nome do relator""" | |
| keywords = ["relator", "relatora", "relator(a)"] | |
| for keyword in keywords: | |
| if keyword.lower() in texto.lower(): | |
| idx = texto.lower().find(keyword) | |
| inicio = idx + len(keyword) | |
| fim = min(len(texto), inicio + 100) | |
| extracted = texto[inicio:fim].strip() | |
| # Limpar pontuação e pegar apenas o nome | |
| nome = extracted.split("\n")[0].split(".")[0].strip() | |
| return nome[:100] | |
| return "RELATOR NÃO IDENTIFICADO" | |
| def _extrair_data_julgamento(self, texto: str) -> str: | |
| """Extrai data do julgamento""" | |
| import re | |
| # Padrão: DD de mês de YYYY ou DD/MM/YYYY | |
| pattern = r'\d{1,2}\s+(de\s+)?[a-záéíóú]+\s+(de\s+)?\d{4}|\d{1,2}/\d{1,2}/\d{4}' | |
| matches = re.findall(pattern, texto) | |
| if matches: | |
| return matches[0] if isinstance(matches[0], str) else matches[0][0] | |
| return "DATA NÃO IDENTIFICADA" | |
| def _extrair_ramo_principal(self, texto: str) -> str: | |
| """Extrai ramo jurídico principal""" | |
| ramos = { | |
| "direito do consumidor": ["consumidor", "fornecedor", "cdc", "consumidora"], | |
| "responsabilidade civil": ["responsabilidade", "indenização", "dano moral", "dano material"], | |
| "direito de família": ["divórcio", "guarda", "alimentos", "conjugal", "familiar"], | |
| "direito imobiliário": ["imóvel", "propriedade", "locação", "contrato", "imóvel"], | |
| "direito do trabalho": ["trabalhista", "empregado", "empregador", "vínculo", "salário"], | |
| "direito administrativo": ["administrativo", "funcionário", "servidor", "público"], | |
| "direito penal": ["crime", "penal", "acusado", "vítima", "condenado"], | |
| } | |
| texto_lower = texto.lower() | |
| for ramo, keywords in ramos.items(): | |
| if any(keyword in texto_lower for keyword in keywords): | |
| return ramo.upper() | |
| return "DIREITO CÍVEL" | |
| def _extrair_ramos_secundarios(self, texto: str) -> list: | |
| """Extrai ramos secundários""" | |
| ramos = [ | |
| {"ramo": "Responsabilidade Civil", "relevancia": "alta"}, | |
| {"ramo": "Dano Moral", "relevancia": "média"}, | |
| {"ramo": "Contrato", "relevancia": "baixa"} | |
| ] | |
| return ramos | |
| def _extrair_contato_tribunal(self, texto: str) -> dict: | |
| """Extrai dados de contato do tribunal""" | |
| import re | |
| contato = {} | |
| email_pattern = r'[\w\.-]+@[\w\.-]+\.\w+' | |
| emails = re.findall(email_pattern, texto) | |
| if emails: | |
| contato["email"] = emails[0] | |
| # Telefone | |
| phone_pattern = r'\(?[0-9]{2}\)?[0-9]{4,5}-[0-9]{4}' | |
| phones = re.findall(phone_pattern, texto) | |
| if phones: | |
| contato["telefone"] = phones[0] | |
| return contato if contato else None | |