core/pipeline/utils.py

# core/pipeline/utils
import re
from typing import Union


def limpiar_contexto_bruto(contexto: str) -> str:
    """Optimiza el contexto eliminando saltos de línea redundantes, espacios excesivos y corrigiendo puntuación."""
    contexto = contexto.strip()

    # 1️⃣ Reemplaza saltos de línea múltiples por un espacio si hay punto antes, o por punto y espacio si no hay.
    contexto = re.sub(
        r"\.\s*\n+", ". ", contexto
    )  # Si ya hay punto antes del salto, limpia
    contexto = re.sub(
        r"(?<!\.)\n+", ". ", contexto
    )  # Si NO hay punto antes, agrega punto

    # 2️⃣ Elimina espacios múltiples
    contexto = re.sub(r"\s{2,}", " ", contexto)

    # 3️⃣ Corrige puntos dobles o triples
    contexto = re.sub(r"\.{2,}", ".", contexto)

    # 4️⃣ Elimina punto al inicio si quedó por error
    contexto = re.sub(r"^\.\s*", "", contexto)

    # 5️⃣ Asegura que termine en punto
    if not contexto.endswith("."):
        contexto += "."

    return contexto.strip()


def limitar_contexto(contexto: str, max_tokens: int = 1500) -> str:
    """Recorta el contexto si excede el límite de tokens."""
    if len(contexto) > max_tokens:
        return contexto[:max_tokens] + "\n[Contexto truncado...]"
    return contexto


def validar_input(input_data: Union[str, bytes]) -> bool:
    """Valida que la entrada sea texto no vacío o bytes (imagen)."""
    if isinstance(input_data, str):
        return bool(input_data.strip())
    if isinstance(input_data, bytes):
        return True
    return False