Spaces:
Sleeping
Sleeping
File size: 1,812 Bytes
c95a8f6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | import re
from typing import List, Any
import logging
logger = logging.getLogger(__name__)
class DocumentCleaner:
def __init__(self, min_length: int = 50):
self.min_length = min_length
# Regex for common PDF artifacts or excessive whitespace
self.whitespace_pattern = re.compile(r'\s+')
self.control_chars_pattern = re.compile(r'[\x00-\x1f\x7f-\x9f]')
def clean_text(self, text: str) -> str:
if not text:
return ""
# Safely remove surrogate code points that crash Windows terminals
text = text.encode('utf-8', 'ignore').decode('utf-8')
# Remove control characters
text = self.control_chars_pattern.sub('', text)
# Normalize whitespace (replace newlines/tabs with space and collapse)
text = self.whitespace_pattern.sub(' ', text).strip()
return text
def clean_documents(self, documents: List[Any]) -> List[Any]:
"""
Cleans a list of LangChain Document objects.
Filters out documents that are too short after cleaning.
"""
cleaned_docs = []
for doc in documents:
if not hasattr(doc, 'page_content') or doc.page_content is None:
continue
cleaned_text = self.clean_text(str(doc.page_content))
if len(cleaned_text) >= self.min_length:
# Update the document content with cleaned version
doc.page_content = cleaned_text
cleaned_docs.append(doc)
print(f"[INFO] Data Cleaning: {len(documents)} -> {len(cleaned_docs)} documents (filtered {len(documents) - len(cleaned_docs)})")
return cleaned_docs
def get_default_cleaner():
return DocumentCleaner()
|