Diagrama del sistema
Stack tecnológico
Django 5.1 + DRF
Framework robusto, ORM, admin automático
DeepSeek API
1M tokens contexto, JSON nativo
Ollama (local)
Fallback offline sin costo
PostgreSQL
Integridad transaccional, JSON nativo
Celery + Redis + RabbitMQ
Procesamiento asíncrono masivo
Google Cloud Storage
Escalabilidad para PDFs
PyMuPDF + pdfplumber
Layout + tablas + OCR (Tesseract)
pandas + openpyxl
Importación desde Excel
SendGrid
Notificaciones transaccionales
Sentry SDK
Monitoreo de errores en producción
Gunicorn + Nginx
Servidor WSGI + reverse proxy SSL
AdminLTE + Bootstrap 4
UI responsive reutilizable
Pipeline de extracción híbrida
Python
def extraer_contenido_hibrido(self, pdf_origen) -> Dict[str, Any]:
"""
Sistema híbrido que combina múltiples métodos de extracción.
PyMuPDF → layout visual · pdfplumber → tablas · Tesseract → OCR
"""
contenido = {
"texto_crudo": "", "texto_estructurado": "",
"tablas": "", "metadatos": "", "tiene_imagenes": False,
"metodo_extraccion": self.metodo_extraccion,
}
# Método 1: PyMuPDF para layout visual general
if 'layout' in self.configuracion['METODOS_EXTRACCION']:
contenido["texto_estructurado"] = self.extraer_texto_con_layout(pdf_origen)
# Método 2: pdfplumber para detección de tablas
if 'tablas' in self.configuracion['METODOS_EXTRACCION']:
contenido["tablas"] = self.extraer_tablas_con_pdfplumber(pdf_origen)
# Método 3: Verificar si necesita OCR (documento escaneado)
necesita_ocr = self._necesita_ocr(pdf_origen)
if necesita_ocr and self.usar_ocr:
contenido["texto_crudo"] = self._extraer_con_ocr(pdf_origen)
contenido["metodo_extraccion"] = "ocr"
else:
contenido["texto_crudo"] = contenido["texto_estructurado"] + "\n\n" + contenido["tablas"]
# Fallback a texto básico si los métodos anteriores fallan
if not contenido["texto_crudo"].strip():
contenido["texto_crudo"] = self.extraer_texto_de_pdf(pdf_origen)
return contenido
Estrategia de reintentos y manejo de errores
Python
try:
# Procesamiento directo con DeepSeek
return self._procesar_con_deepseek(prompt_completo)
except APIConnectionError as error:
# Error de conectividad → notificar y detener
mensaje_error = f"Error de conexión con DeepSeek API: {str(error)}"
return {"status": "error", "errors": [mensaje_error]}
except APIStatusError as error:
# Error HTTP (402, 429, 500, etc.)
if error.status_code == 402:
# Notificar al staff por email: saldo agotado
enviar_email.delay(
subject="Error 402 - Pago requerido en DeepSeek API",
message=f"Error al procesar PDF. Saldo agotado.",
email=list(Usuario.objects.filter(is_staff=True)
.values_list('email', flat=True))
)
return {"status": "error", "errors": [str(error)]}
except BadRequestError as error:
# Prompt mal formado → no reintentar
return {"status": "error", "errors": [str(error)]}
except Exception as error:
# Error desconocido → fallback a procesamiento iterativo
if hasattr(archivo_pdf, 'seek'):
archivo_pdf.seek(0)
return self.procesar_pdf_grande_iterativamente(
archivo_pdf, nombre_proceso, nombre_archivo)
Prompt engineering: respuesta JSON estructurada
Python
# Llamada a DeepSeek con response_format JSON
respuesta = openai.chat.completions.create(
model=self.modelo, # deepseek-v4-pro o deepseek-v4-flash
messages=[
{"role": "system", "content": prompt_sistema}, # 17 reglas de negocio
{"role": "user", "content": "Analizá el documento fiscal proporcionado."}
],
response_format={'type': 'json_object'}, # Respuesta JSON garantizada
temperature=self.temperatura,
max_tokens=self.max_tokens # Hasta 384K tokens de salida
)
# La respuesta del modelo es un JSON con esta estructura:
{
"facturas": [
{
"poliza": 229910,
"endoso": 44,
"vigencia_desde": "2025-01-01",
"vigencia_hasta": "2025-06-30",
"factura": "0002-00294656",
"premio": 150000.00,
"suma_asegurada": 5000000.00,
"vencimiento_cuota": "2025-02-15",
"status": "ok",
"messages": [],
"comments": "Factura procesada. Layout estándar ACG.",
"anulada": false,
"motivo_anulada": ""
}
]
}
