|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
CHUNK_TARGET_CHARS = 1200
|
|
CHUNK_OVERLAP_CHARS = 150
|
|
MIN_CHUNK_CHARS = 80
|
|
PARAGRAPH_SPLIT = re.compile(r"\n\s*\n")
|
|
WHITESPACE = re.compile(r"[ \t]+")
|
|
|
|
|
|
def normalize_text(text: str) -> str:
|
|
cleaned = (text or "").replace("\r", "\n")
|
|
cleaned = WHITESPACE.sub(" ", cleaned)
|
|
lines = [line.strip() for line in cleaned.split("\n")]
|
|
return "\n".join(line for line in lines if line)
|
|
|
|
|
|
def chunk_text(text: str) -> list[str]:
|
|
normalized = normalize_text(text)
|
|
if not normalized:
|
|
return []
|
|
paragraphs = [p.strip() for p in PARAGRAPH_SPLIT.split(normalized) if p.strip()]
|
|
if not paragraphs:
|
|
paragraphs = [normalized]
|
|
chunks: list[str] = []
|
|
buffer = ""
|
|
for paragraph in paragraphs:
|
|
if len(paragraph) > CHUNK_TARGET_CHARS:
|
|
if buffer:
|
|
chunks.append(buffer)
|
|
buffer = ""
|
|
for start in range(0, len(paragraph), CHUNK_TARGET_CHARS - CHUNK_OVERLAP_CHARS):
|
|
piece = paragraph[start : start + CHUNK_TARGET_CHARS]
|
|
if len(piece) >= MIN_CHUNK_CHARS:
|
|
chunks.append(piece)
|
|
continue
|
|
if len(buffer) + len(paragraph) + 1 > CHUNK_TARGET_CHARS and buffer:
|
|
chunks.append(buffer)
|
|
buffer = paragraph
|
|
else:
|
|
buffer = f"{buffer}\n{paragraph}" if buffer else paragraph
|
|
if buffer and len(buffer) >= MIN_CHUNK_CHARS:
|
|
chunks.append(buffer)
|
|
elif buffer and not chunks:
|
|
chunks.append(buffer)
|
|
return chunks
|