File size: 6,406 Bytes
6b29104 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 |
"""
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)}"
|