preprocessing.py

"""Clean raw document text before chunking."""

import re


def clean_text(text: str) -> str:
    """Remove excessive whitespace, headers noise, and non-printable characters."""
    # Remove non-printable characters except newlines and tabs
    text = re.sub(r"[^\x09\x0A\x0D\x20-\x7E\u00A0-\uFFFF]", " ", text)
    # Collapse multiple blank lines into one
    text = re.sub(r"\n{3,}", "\n\n", text)
    # Collapse multiple spaces into one
    text = re.sub(r" {2,}", " ", text)
    # Strip leading/trailing whitespace
    text = text.strip()
    return text


def preprocess_documents(documents: list[dict]) -> list[dict]:
    """Apply clean_text to every document."""
    cleaned = []
    for doc in documents:
        cleaned.append({
            "source": doc["source"],
            "text": clean_text(doc["text"]),
        })
    print(f"[preprocessing] Cleaned {len(cleaned)} documents.")
    return cleaned