""" 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 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