Mejoras Implementadas en Extracción de PDFs con IA

He mejorado significativamente el sistema de extracción de texto de PDFs para el análisis con IA. Aquí están los cambios principales:

🎯 Problemas Resueltos
Límites muy pequeños → Aumentados significativamente según capacidad del modelo
No extraía todo el documento → Ahora procesa hasta 100k caracteres
Duplicación de contenido → Detecta y elimina páginas repetidas
Sin información de estado → Reporta páginas procesadas, truncado, etc.
This commit is contained in:
2025-12-03 00:55:11 -03:00
parent 582114a55a
commit 50909e4499
4 changed files with 279 additions and 52 deletions

View File

@@ -209,7 +209,74 @@ def send_completed_inspection_to_n8n(inspection, db):
# No lanzamos excepción para no interrumpir el flujo normal
BACKEND_VERSION = "1.0.94"
# ============================================================================
# UTILIDADES PARA PROCESAMIENTO DE PDFs
# ============================================================================
def extract_pdf_text_smart(pdf_content: bytes, max_chars: int = None) -> dict:
"""
Extrae texto de un PDF de forma inteligente, evitando duplicaciones
y manejando PDFs grandes.
Args:
pdf_content: Contenido del PDF en bytes
max_chars: Límite máximo de caracteres (None = sin límite)
Returns:
dict con 'text', 'pages', 'total_chars', 'truncated'
"""
from pypdf import PdfReader
from io import BytesIO
try:
pdf_file = BytesIO(pdf_content)
pdf_reader = PdfReader(pdf_file)
full_text = ""
pages_processed = 0
total_pages = len(pdf_reader.pages)
for page_num, page in enumerate(pdf_reader.pages, 1):
page_text = page.extract_text()
# Limpiar y validar texto de la página
if page_text and page_text.strip():
# Evitar duplicación: verificar si el texto ya existe
# (algunos PDFs pueden tener páginas repetidas)
if page_text.strip() not in full_text:
full_text += f"\n--- Página {page_num}/{total_pages} ---\n{page_text.strip()}\n"
pages_processed += 1
# Si hay límite y lo alcanzamos, detener
if max_chars and len(full_text) >= max_chars:
break
total_chars = len(full_text)
truncated = False
# Aplicar límite si se especificó
if max_chars and total_chars > max_chars:
full_text = full_text[:max_chars]
truncated = True
return {
'text': full_text,
'pages': total_pages,
'pages_processed': pages_processed,
'total_chars': total_chars,
'truncated': truncated,
'success': True
}
except Exception as e:
return {
'text': '',
'error': str(e),
'success': False
}
BACKEND_VERSION = "1.0.95"
app = FastAPI(title="Checklist Inteligente API", version=BACKEND_VERSION)
# S3/MinIO configuration
@@ -2598,7 +2665,6 @@ async def analyze_image(
# Guardar archivo temporalmente y procesar según tipo
import base64
from pypdf import PdfReader
contents = await file.read()
file_type = file.content_type
@@ -2610,34 +2676,30 @@ async def analyze_image(
if is_pdf:
print("📕 Detectado PDF - extrayendo texto...")
try:
from io import BytesIO
pdf_file = BytesIO(contents)
pdf_reader = PdfReader(pdf_file)
# Extraer texto de todas las páginas
pdf_text = ""
for page_num, page in enumerate(pdf_reader.pages):
pdf_text += f"\n--- Página {page_num + 1} ---\n"
pdf_text += page.extract_text()
print(f"✅ Texto extraído: {len(pdf_text)} caracteres")
if not pdf_text.strip():
return {
"status": "error",
"message": "No se pudo extraer texto del PDF. Puede ser un PDF escaneado sin OCR."
}
# Para PDFs usamos análisis de texto, no de imagen
image_b64 = None
except Exception as e:
print(f"❌ Error al procesar PDF: {str(e)}")
# Usar función inteligente de extracción
# Para análisis de imagen usamos hasta 100k caracteres (Gemini soporta mucho más)
pdf_result = extract_pdf_text_smart(contents, max_chars=100000)
if not pdf_result['success']:
return {
"status": "error",
"message": f"Error al procesar PDF: {str(e)}"
"message": f"Error al procesar PDF: {pdf_result.get('error', 'Unknown')}"
}
pdf_text = pdf_result['text']
print(f"✅ Texto extraído: {pdf_result['total_chars']} caracteres de {pdf_result['pages_processed']}/{pdf_result['pages']} páginas")
if pdf_result['truncated']:
print(f"⚠️ PDF truncado a 100k caracteres")
if not pdf_text.strip():
return {
"status": "error",
"message": "No se pudo extraer texto del PDF. Puede ser un PDF escaneado sin OCR."
}
# Para PDFs usamos análisis de texto, no de imagen
image_b64 = None
else:
# Es una imagen
image_b64 = base64.b64encode(contents).decode('utf-8')
@@ -2819,7 +2881,7 @@ NOTA:
{"role": "system", "content": system_prompt},
{
"role": "user",
"content": f"{user_message}\n\n--- CONTENIDO DEL DOCUMENTO PDF ---\n{pdf_text[:4000]}" # Limitar a 4000 chars
"content": f"{user_message}\n\n--- CONTENIDO DEL DOCUMENTO PDF ({len(pdf_text)} caracteres) ---\n{pdf_text[:30000]}" # 30k chars para GPT-4
}
]
else:
@@ -2860,8 +2922,8 @@ NOTA:
prompt = f"{system_prompt}\n\n{user_message}"
if is_pdf:
# Para PDF, solo texto
prompt_with_content = f"{prompt}\n\n--- CONTENIDO DEL DOCUMENTO PDF ---\n{pdf_text[:4000]}"
# Para PDF, solo texto - Gemini puede manejar contextos muy largos (2M tokens)
prompt_with_content = f"{prompt}\n\n--- CONTENIDO DEL DOCUMENTO PDF ({len(pdf_text)} caracteres) ---\n{pdf_text[:100000]}"
response = model.generate_content(prompt_with_content)
else:
# Para imagen, incluir imagen
@@ -3036,8 +3098,6 @@ async def chat_with_ai_assistant(
attached_files_data = []
if files:
import base64
from pypdf import PdfReader
from io import BytesIO
for file in files:
file_content = await file.read()
@@ -3051,18 +3111,22 @@ async def chat_with_ai_assistant(
# Si es PDF, extraer texto
if file_type == 'application/pdf' or file.filename.lower().endswith('.pdf'):
try:
pdf_file = BytesIO(file_content)
pdf_reader = PdfReader(pdf_file)
pdf_text = ""
for page in pdf_reader.pages:
pdf_text += page.extract_text()
# Usar función inteligente - límite de 50k para chat (balance entre contexto y tokens)
pdf_result = extract_pdf_text_smart(file_content, max_chars=50000)
if pdf_result['success']:
file_info['content_type'] = 'pdf'
file_info['text'] = pdf_text[:2000] # Limitar texto
print(f"📄 PDF procesado: {file.filename} - {len(pdf_text)} caracteres")
except Exception as e:
print(f"❌ Error procesando PDF {file.filename}: {str(e)}")
file_info['error'] = str(e)
file_info['text'] = pdf_result['text']
file_info['total_chars'] = pdf_result['total_chars']
file_info['pages'] = pdf_result['pages']
file_info['pages_processed'] = pdf_result['pages_processed']
file_info['truncated'] = pdf_result['truncated']
truncated_msg = " (TRUNCADO)" if pdf_result['truncated'] else ""
print(f"📄 PDF procesado: {file.filename} - {pdf_result['total_chars']} caracteres, {pdf_result['pages_processed']}/{pdf_result['pages']} páginas{truncated_msg}")
else:
print(f"❌ Error procesando PDF {file.filename}: {pdf_result.get('error', 'Unknown')}")
file_info['error'] = pdf_result.get('error', 'Error desconocido')
# Si es imagen, convertir a base64
elif file_type.startswith('image/'):
@@ -3120,9 +3184,12 @@ INFORMACIÓN DEL VEHÍCULO:
attached_context = f"\n\nARCHIVOS ADJUNTOS EN ESTE MENSAJE ({len(attached_files_data)} archivos):\n"
for idx, file_info in enumerate(attached_files_data, 1):
if file_info.get('content_type') == 'pdf':
attached_context += f"\n{idx}. PDF: {file_info['filename']}\n"
truncated_indicator = "TRUNCADO" if file_info.get('truncated') else ""
pages_info = f" ({file_info.get('pages_processed', '?')}/{file_info.get('pages', '?')} páginas, {file_info.get('total_chars', '?')} caracteres{truncated_indicator})" if 'pages' in file_info else ""
attached_context += f"\n{idx}. PDF: {file_info['filename']}{pages_info}\n"
if 'text' in file_info:
attached_context += f" Contenido: {file_info['text'][:500]}...\n"
# Mostrar más contexto del PDF (primeros 2000 caracteres como preview)
attached_context += f" Contenido: {file_info['text'][:2000]}...\n"
elif file_info.get('content_type') == 'image':
attached_context += f"\n{idx}. Imagen: {file_info['filename']}\n"