""" Nó de carregamento de PDF para o AgentPDF. Este nó é responsável por carregar e extrair texto de arquivos PDF usando PyPDF2 e preparar o conteúdo para processamento posterior. """ import os from typing import Dict, Any from PyPDF2 import PdfReader from langchain_core.runnables import RunnableConfig from agents.state import PDFState, ProcessingStatus from utils.logger import log_node_execution, main_logger def load_pdf_node(state: PDFState, config: RunnableConfig) -> Dict[str, Any]: """ Nó responsável por carregar e extrair texto de arquivos PDF. Este nó: 1. Verifica se o caminho do PDF é válido 2. Carrega o PDF usando PyPDF2 3. Extrai todo o texto do documento 4. Atualiza o estado com o texto extraído Args: state: Estado atual do grafo contendo informações do PDF config: Configuração do LangGraph Returns: Dict[str, Any]: Atualizações para o estado """ log_node_execution("PDF_LOADER", "START", "Iniciando carregamento do PDF") try: # Verifica se o caminho do PDF foi fornecido pdf_path = state.get("pdf_path") if not pdf_path: error_msg = "Caminho do PDF não fornecido" log_node_execution("PDF_LOADER", "ERROR", error_msg) return { "processing_status": ProcessingStatus.ERROR, "error_message": error_msg } # Verifica se o arquivo existe if not os.path.exists(pdf_path): error_msg = f"Arquivo PDF não encontrado: {pdf_path}" log_node_execution("PDF_LOADER", "ERROR", error_msg) return { "processing_status": ProcessingStatus.ERROR, "error_message": error_msg } # Atualiza status para carregamento log_node_execution("PDF_LOADER", "PROCESSING", f"Carregando PDF: {pdf_path}") # Carrega e extrai texto do PDF extracted_text = extract_text_from_pdf(pdf_path) if not extracted_text.strip(): error_msg = "Nenhum texto foi extraído do PDF" log_node_execution("PDF_LOADER", "ERROR", error_msg) return { "processing_status": ProcessingStatus.ERROR, "error_message": error_msg } # Sucesso - retorna texto extraído log_node_execution( "PDF_LOADER", "SUCCESS", f"Texto extraído com sucesso. Tamanho: {len(extracted_text)} caracteres" ) return { "pdf_text": extracted_text, "processing_status": ProcessingStatus.PROCESSING_TEXT, "error_message": None } except Exception as e: error_msg = f"Erro ao carregar PDF: {str(e)}" log_node_execution("PDF_LOADER", "ERROR", error_msg) main_logger.exception("Erro detalhado no carregamento do PDF:") return { "processing_status": ProcessingStatus.ERROR, "error_message": error_msg } def extract_text_from_pdf(pdf_path: str) -> str: """ Extrai texto de um arquivo PDF usando PyPDF2. Args: pdf_path: Caminho para o arquivo PDF Returns: str: Texto extraído do PDF Raises: Exception: Se houver erro na leitura do PDF """ try: text_content = [] # Abre e lê o PDF with open(pdf_path, 'rb') as file: pdf_reader = PdfReader(file) # Extrai texto de cada página for page_num, page in enumerate(pdf_reader.pages): try: page_text = page.extract_text() if page_text.strip(): # Só adiciona se a página tem texto text_content.append(page_text) main_logger.debug(f"Texto extraído da página {page_num + 1}") except Exception as e: main_logger.warning(f"Erro ao extrair texto da página {page_num + 1}: {e}") continue # Junta todo o texto full_text = "\n\n".join(text_content) # Limpa o texto (remove espaços extras, quebras de linha desnecessárias) cleaned_text = clean_extracted_text(full_text) main_logger.info(f"PDF processado: {len(pdf_reader.pages)} páginas, {len(cleaned_text)} caracteres") return cleaned_text except Exception as e: main_logger.error(f"Erro ao extrair texto do PDF {pdf_path}: {e}") raise def clean_extracted_text(text: str) -> str: """ Limpa e normaliza o texto extraído do PDF. Args: text: Texto bruto extraído do PDF Returns: str: Texto limpo e normalizado """ if not text: return "" # Remove quebras de linha excessivas text = text.replace('\n\n\n', '\n\n') # Remove espaços extras lines = [] for line in text.split('\n'): cleaned_line = ' '.join(line.split()) # Remove espaços extras if cleaned_line: # Só adiciona linhas não vazias lines.append(cleaned_line) # Junta as linhas limpas cleaned_text = '\n'.join(lines) return cleaned_text def validate_pdf_file(pdf_path: str) -> tuple[bool, str]: """ Valida se um arquivo PDF é válido e pode ser processado. Args: pdf_path: Caminho para o arquivo PDF Returns: tuple[bool, str]: (é_válido, mensagem_de_erro) """ try: # Verifica se o arquivo existe if not os.path.exists(pdf_path): return False, f"Arquivo não encontrado: {pdf_path}" # Verifica se é um arquivo PDF if not pdf_path.lower().endswith('.pdf'): return False, "Arquivo deve ter extensão .pdf" # Tenta abrir o PDF para verificar se é válido with open(pdf_path, 'rb') as file: pdf_reader = PdfReader(file) # Verifica se tem pelo menos uma página if len(pdf_reader.pages) == 0: return False, "PDF não contém páginas" return True, "PDF válido" except Exception as e: return False, f"Erro ao validar PDF: {str(e)}"