Coverage for haystack/components/preprocessors/document_cleaner.py: 99%
125 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-12 09:42 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import re
6from collections.abc import Generator
7from copy import deepcopy
8from functools import partial, reduce
9from itertools import chain
10from typing import Literal
11from unicodedata import normalize
13from haystack import Document, component, logging
15logger = logging.getLogger(__name__)
18@component
19class DocumentCleaner:
20 """
21 Cleans the text in the documents.
23 It removes extra whitespaces,
24 empty lines, specified substrings, regexes,
25 page headers and footers (in this order).
27 ### Usage example:
29 ```python
30 from haystack import Document
31 from haystack.components.preprocessors import DocumentCleaner
33 doc = Document(content="This is a document to clean\\n\\n\\nsubstring to remove")
35 cleaner = DocumentCleaner(remove_substrings = ["substring to remove"])
36 result = cleaner.run(documents=[doc])
38 assert result["documents"][0].content == "This is a document to clean "
39 ```
40 """
42 def __init__(
43 self,
44 remove_empty_lines: bool = True,
45 remove_extra_whitespaces: bool = True,
46 remove_repeated_substrings: bool = False,
47 keep_id: bool = False,
48 remove_substrings: list[str] | None = None,
49 remove_regex: str | None = None,
50 unicode_normalization: Literal["NFC", "NFKC", "NFD", "NFKD"] | None = None,
51 ascii_only: bool = False,
52 strip_whitespaces: bool = False,
53 replace_regexes: dict[str, str] | None = None,
54 min_content_length: int = 0,
55 ) -> None:
56 """
57 Initialize DocumentCleaner.
59 :param remove_empty_lines: If `True`, removes empty lines.
60 :param remove_extra_whitespaces: If `True`, removes extra whitespaces.
61 :param remove_repeated_substrings: If `True`, removes repeated substrings (headers and footers) from pages.
62 Pages must be separated by a form feed character "\\f",
63 which is supported by `TextFileToDocument` and `AzureOCRDocumentConverter`.
64 :param remove_substrings: List of substrings to remove from the text.
65 :param remove_regex: Regex to match and replace substrings by "".
66 :param keep_id: If `True`, keeps the IDs of the original documents.
67 :param unicode_normalization: Unicode normalization form to apply to the text.
68 Note: This will run before any other steps.
69 :param ascii_only: Whether to convert the text to ASCII only.
70 Will remove accents from characters and replace them with ASCII characters.
71 Other non-ASCII characters will be removed.
72 Note: This will run before any pattern matching or removal.
73 :param strip_whitespaces: If `True`, removes leading and trailing whitespace from the document content
74 using Python's `str.strip()`. Unlike `remove_extra_whitespaces`, this only affects the beginning
75 and end of the text, preserving internal whitespace (useful for markdown formatting).
76 :param replace_regexes: A dictionary mapping regex patterns to their replacement strings.
77 For example, `{r'\\n\\n+': '\\n'}` replaces multiple consecutive newlines with a single newline.
78 This is applied after `remove_regex` and allows custom replacements instead of just removal.
79 :param min_content_length: Minimum length of the cleaned document content after stripping leading and trailing
80 whitespace. Documents shorter than this value are dropped. A value of `0` keeps all documents.
81 """
83 self._validate_params(unicode_normalization=unicode_normalization)
84 if min_content_length < 0:
85 raise ValueError("min_content_length must be greater than or equal to 0.")
87 self.remove_empty_lines = remove_empty_lines
88 self.remove_extra_whitespaces = remove_extra_whitespaces
89 self.remove_repeated_substrings = remove_repeated_substrings
90 self.remove_substrings = remove_substrings
91 self.remove_regex = remove_regex
92 self.keep_id = keep_id
93 self.unicode_normalization = unicode_normalization
94 self.ascii_only = ascii_only
95 self.strip_whitespaces = strip_whitespaces
96 self.replace_regexes = replace_regexes
97 self.min_content_length = min_content_length
99 def _validate_params(self, unicode_normalization: str | None) -> None:
100 """
101 Validate the parameters of the DocumentCleaner.
103 :param unicode_normalization: Unicode normalization form to apply to the text.
104 :raises ValueError: if the parameters are not valid.
105 """
106 if unicode_normalization and unicode_normalization not in ["NFC", "NFKC", "NFD", "NFKD"]:
107 raise ValueError("unicode_normalization must be one of 'NFC', 'NFKC', 'NFD', 'NFKD'.")
109 @component.output_types(documents=list[Document])
110 def run(self, documents: list[Document]) -> dict[str, list[Document]]:
111 """
112 Cleans up the documents.
114 :param documents: List of Documents to clean.
116 :returns: A dictionary with the following key:
117 - `documents`: List of cleaned Documents.
119 :raises TypeError: if documents is not a list of Documents.
120 """
121 if not isinstance(documents, list) or documents and not isinstance(documents[0], Document):
122 raise TypeError("DocumentCleaner expects a List of Documents as input.")
124 cleaned_docs = []
125 for doc in documents:
126 if doc.content is None:
127 logger.warning(
128 "DocumentCleaner only cleans text documents but document.content for document ID"
129 " {document_id} is None.",
130 document_id=doc.id,
131 )
132 cleaned_docs.append(doc)
133 continue
134 text = doc.content
136 if self.unicode_normalization:
137 text = self._normalize_unicode(text, self.unicode_normalization)
138 if self.ascii_only:
139 text = self._ascii_only(text)
140 if self.remove_extra_whitespaces:
141 text = self._remove_extra_whitespaces(text)
142 if self.remove_empty_lines:
143 text = self._remove_empty_lines(text)
144 if self.remove_substrings:
145 text = self._remove_substrings(text, self.remove_substrings)
146 if self.remove_regex:
147 text = self._remove_regex(text, self.remove_regex)
148 if self.replace_regexes:
149 text = self._replace_regexes(text, self.replace_regexes)
150 if self.remove_repeated_substrings:
151 text = self._remove_repeated_substrings(text)
152 if self.strip_whitespaces:
153 text = text.strip()
155 if self.min_content_length:
156 cleaned_content = text.strip()
157 if len(cleaned_content) < self.min_content_length:
158 logger.debug(
159 "Document ID {doc_id} has cleaned content shorter than min_content_length "
160 "({content_length} < {min_content_length}). Skipping this document.",
161 doc_id=doc.id,
162 content_length=len(cleaned_content),
163 min_content_length=self.min_content_length,
164 )
165 continue
167 clean_doc = Document(
168 id=doc.id if self.keep_id else "",
169 content=text,
170 blob=doc.blob,
171 meta=deepcopy(doc.meta),
172 score=doc.score,
173 embedding=doc.embedding,
174 sparse_embedding=doc.sparse_embedding,
175 )
176 cleaned_docs.append(clean_doc)
178 return {"documents": cleaned_docs}
180 def _normalize_unicode(self, text: str, form: Literal["NFC", "NFKC", "NFD", "NFKD"]) -> str:
181 """
182 Normalize the unicode of the text.
184 :param text: Text to normalize.
185 :param form: Unicode normalization form to apply to the text.
186 Options: "NFC", "NFKC", "NFD", "NFKD".
187 :returns: The normalized text.
188 """
189 return normalize(form, text)
191 def _ascii_only(self, text: str) -> str:
192 """
193 Convert the text to ASCII only.
195 Will remove accents from characters and replace them with ASCII characters.
196 Other non-ASCII characters will be removed.
198 :param text: Text to convert to ASCII only.
199 :returns: The text in ASCII only.
200 """
202 # First normalize the text to NFKD to separate the characters and their diacritics
203 # Then encode it to ASCII and ignore any characters that can't be encoded
204 return self._normalize_unicode(text, "NFKD").encode("ascii", "ignore").decode("utf-8")
206 def _remove_empty_lines(self, text: str) -> str:
207 """
208 Remove empty lines and lines that contain nothing but whitespaces from text.
210 :param text: Text to clean.
211 :returns: The text without empty lines.
212 """
213 pages = text.split("\f")
214 cleaned_pages = ["\n".join(line for line in page.split("\n") if line.strip()) for page in pages]
215 return "\f".join(cleaned_pages)
217 def _remove_extra_whitespaces(self, text: str) -> str:
218 """
219 Remove extra whitespaces from text.
221 :param text: Text to clean.
222 :returns: The text without extra whitespaces.
223 """
224 texts = text.split("\f")
225 cleaned_text = [re.sub(r"\s\s+", " ", text).strip() for text in texts]
226 return "\f".join(cleaned_text)
228 def _remove_regex(self, text: str, regex: str) -> str:
229 """
230 Remove substrings that match the specified regex from the text.
232 :param text: Text to clean.
233 :param regex: Regex to match and replace substrings by "".
234 :returns: The text without the substrings that match the regex.
235 """
236 texts = text.split("\f")
237 cleaned_text = [re.sub(regex, "", text).strip() for text in texts]
238 return "\f".join(cleaned_text)
240 def _replace_regexes(self, text: str, replace_regexes: dict[str, str]) -> str:
241 """
242 Replace substrings that match the specified regex patterns with custom replacement strings.
244 :param text: Text to clean.
245 :param replace_regexes: A dictionary mapping regex patterns to their replacement strings.
246 :returns: The text with the regex matches replaced by the specified strings.
247 """
248 pages = text.split("\f")
249 cleaned_pages = []
250 for page in pages:
251 for pattern, replacement in replace_regexes.items():
252 page = re.sub(pattern, replacement, page)
253 cleaned_pages.append(page)
254 return "\f".join(cleaned_pages)
256 def _remove_substrings(self, text: str, substrings: list[str]) -> str:
257 """
258 Remove all specified substrings from the text.
260 :param text: Text to clean.
261 :param substrings: Substrings to remove.
262 :returns: The text without the specified substrings.
263 """
264 for substring in substrings:
265 text = text.replace(substring, "")
266 return text
268 def _remove_repeated_substrings(self, text: str) -> str:
269 """
270 Remove any substrings from the text that occur repeatedly on every page. For example headers or footers.
272 Pages in the text need to be separated by form feed character "\f".
273 :param text: Text to clean.
274 :returns: The text without the repeated substrings.
275 """
276 return self._find_and_remove_header_footer(
277 text, n_chars=300, n_first_pages_to_ignore=1, n_last_pages_to_ignore=1
278 )
280 def _find_and_remove_header_footer(
281 self, text: str, n_chars: int, n_first_pages_to_ignore: int, n_last_pages_to_ignore: int
282 ) -> str:
283 """
284 Heuristic to find footers and headers across different pages by searching for the longest common string.
286 Pages in the text need to be separated by form feed character "\f".
287 For headers, we only search in the first n_chars characters (for footer: last n_chars).
288 Note: This heuristic uses exact matches and therefore works well for footers like "Copyright 2019 by XXX",
289 but won't detect "Page 3 of 4" or similar.
291 :param text: The text to remove headers and footers from, with pages separated by the
292 form feed character described above.
293 :param n_chars: The number of first/last characters where the header/footer shall be searched in.
294 :param n_first_pages_to_ignore: The number of first pages to ignore
295 (e.g. TOCs often don't contain footer/header).
296 :param n_last_pages_to_ignore: The number of last pages to ignore.
297 :returns: The text without the found headers and footers.
298 """
300 pages = text.split("\f")
302 # header
303 start_of_pages = [p[:n_chars] for p in pages[n_first_pages_to_ignore:-n_last_pages_to_ignore]]
304 found_header = self._find_longest_common_ngram(start_of_pages)
305 if found_header:
306 pages = [page.replace(found_header, "") for page in pages]
308 # footer
309 end_of_pages = [p[-n_chars:] for p in pages[n_first_pages_to_ignore:-n_last_pages_to_ignore]]
310 found_footer = self._find_longest_common_ngram(end_of_pages)
311 if found_footer:
312 pages = [page.replace(found_footer, "") for page in pages]
314 logger.debug(
315 "Removed header '{header}' and footer '{footer}' in document", header=found_header, footer=found_footer
316 )
317 return "\f".join(pages)
319 def _ngram(self, seq: str, n: int) -> Generator[str, None, None]:
320 """
321 Return all ngrams of length n from a text sequence. Each ngram consists of n words split by whitespace.
323 :param seq: The sequence to generate ngrams from.
324 :param n: The length of the ngrams to generate.
325 :returns: A Generator generating all ngrams of length n from the given sequence.
326 """
328 # In order to maintain the original whitespace, but still consider \n and \t for n-gram tokenization,
329 # we add a space here and remove it after creation of the ngrams again (see below)
330 seq = seq.replace("\n", " \n")
331 seq = seq.replace("\t", " \t")
333 words = seq.split(" ")
334 return (" ".join(words[i : i + n]).replace(" \n", "\n").replace(" \t", "\t") for i in range(len(words) - n + 1))
336 def _allngram(self, seq: str, min_ngram: int, max_ngram: int) -> set[str]:
337 """
338 Generates all possible ngrams from a given sequence of text.
340 Considering all ngram lengths between the minimum and maximum length.
342 :param seq: The sequence to generate ngrams from.
343 :param min_ngram: The minimum length of ngram to consider.
344 :param max_ngram: The maximum length of ngram to consider.
345 :returns: A set of all ngrams from the given sequence.
346 """
347 lengths = range(min_ngram, max_ngram) if max_ngram else range(min_ngram, len(seq))
348 ngrams = map(partial(self._ngram, seq), lengths)
349 return set(chain.from_iterable(ngrams))
351 def _find_longest_common_ngram(self, sequences: list[str], min_ngram: int = 3, max_ngram: int = 30) -> str:
352 """
353 Find the longest common ngram across a list of text sequences (e.g. start of pages).
355 Considering all ngram lengths between the minimum and maximum length. Helpful for finding footers, headers etc.
356 Empty sequences are ignored.
358 :param sequences: The list of strings that shall be searched for common n_grams.
359 :param max_ngram: The maximum length of ngram to consider.
360 :param min_ngram: The minimum length of ngram to consider.
361 :returns: The longest ngram that all sequences have in common.
362 """
363 sequences = [s for s in sequences if s] # filter empty sequences
364 if len(sequences) < 2:
365 # a single sequence has no ngram "in common" with any other; treating
366 # its own longest ngram as a repeated header/footer would wipe it
367 return ""
368 seqs_ngrams = map(partial(self._allngram, min_ngram=min_ngram, max_ngram=max_ngram), sequences)
369 intersection = reduce(set.intersection, seqs_ngrams)
371 longest = max(intersection, key=len, default="")
372 return longest if longest.strip() else ""