Coverage for haystack/components/preprocessors/document_cleaner.py: 99%
117 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +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 ) -> None:
55 """
56 Initialize DocumentCleaner.
58 :param remove_empty_lines: If `True`, removes empty lines.
59 :param remove_extra_whitespaces: If `True`, removes extra whitespaces.
60 :param remove_repeated_substrings: If `True`, removes repeated substrings (headers and footers) from pages.
61 Pages must be separated by a form feed character "\\f",
62 which is supported by `TextFileToDocument` and `AzureOCRDocumentConverter`.
63 :param remove_substrings: List of substrings to remove from the text.
64 :param remove_regex: Regex to match and replace substrings by "".
65 :param keep_id: If `True`, keeps the IDs of the original documents.
66 :param unicode_normalization: Unicode normalization form to apply to the text.
67 Note: This will run before any other steps.
68 :param ascii_only: Whether to convert the text to ASCII only.
69 Will remove accents from characters and replace them with ASCII characters.
70 Other non-ASCII characters will be removed.
71 Note: This will run before any pattern matching or removal.
72 :param strip_whitespaces: If `True`, removes leading and trailing whitespace from the document content
73 using Python's `str.strip()`. Unlike `remove_extra_whitespaces`, this only affects the beginning
74 and end of the text, preserving internal whitespace (useful for markdown formatting).
75 :param replace_regexes: A dictionary mapping regex patterns to their replacement strings.
76 For example, `{r'\\n\\n+': '\\n'}` replaces multiple consecutive newlines with a single newline.
77 This is applied after `remove_regex` and allows custom replacements instead of just removal.
78 """
80 self._validate_params(unicode_normalization=unicode_normalization)
82 self.remove_empty_lines = remove_empty_lines
83 self.remove_extra_whitespaces = remove_extra_whitespaces
84 self.remove_repeated_substrings = remove_repeated_substrings
85 self.remove_substrings = remove_substrings
86 self.remove_regex = remove_regex
87 self.keep_id = keep_id
88 self.unicode_normalization = unicode_normalization
89 self.ascii_only = ascii_only
90 self.strip_whitespaces = strip_whitespaces
91 self.replace_regexes = replace_regexes
93 def _validate_params(self, unicode_normalization: str | None) -> None:
94 """
95 Validate the parameters of the DocumentCleaner.
97 :param unicode_normalization: Unicode normalization form to apply to the text.
98 :raises ValueError: if the parameters are not valid.
99 """
100 if unicode_normalization and unicode_normalization not in ["NFC", "NFKC", "NFD", "NFKD"]:
101 raise ValueError("unicode_normalization must be one of 'NFC', 'NFKC', 'NFD', 'NFKD'.")
103 @component.output_types(documents=list[Document])
104 def run(self, documents: list[Document]) -> dict[str, list[Document]]:
105 """
106 Cleans up the documents.
108 :param documents: List of Documents to clean.
110 :returns: A dictionary with the following key:
111 - `documents`: List of cleaned Documents.
113 :raises TypeError: if documents is not a list of Documents.
114 """
115 if not isinstance(documents, list) or documents and not isinstance(documents[0], Document):
116 raise TypeError("DocumentCleaner expects a List of Documents as input.")
118 cleaned_docs = []
119 for doc in documents:
120 if doc.content is None:
121 logger.warning(
122 "DocumentCleaner only cleans text documents but document.content for document ID"
123 " {document_id} is None.",
124 document_id=doc.id,
125 )
126 cleaned_docs.append(doc)
127 continue
128 text = doc.content
130 if self.unicode_normalization:
131 text = self._normalize_unicode(text, self.unicode_normalization)
132 if self.ascii_only:
133 text = self._ascii_only(text)
134 if self.remove_extra_whitespaces:
135 text = self._remove_extra_whitespaces(text)
136 if self.remove_empty_lines:
137 text = self._remove_empty_lines(text)
138 if self.remove_substrings:
139 text = self._remove_substrings(text, self.remove_substrings)
140 if self.remove_regex:
141 text = self._remove_regex(text, self.remove_regex)
142 if self.replace_regexes:
143 text = self._replace_regexes(text, self.replace_regexes)
144 if self.remove_repeated_substrings:
145 text = self._remove_repeated_substrings(text)
146 if self.strip_whitespaces:
147 text = text.strip()
149 clean_doc = Document(
150 id=doc.id if self.keep_id else "",
151 content=text,
152 blob=doc.blob,
153 meta=deepcopy(doc.meta),
154 score=doc.score,
155 embedding=doc.embedding,
156 sparse_embedding=doc.sparse_embedding,
157 )
158 cleaned_docs.append(clean_doc)
160 return {"documents": cleaned_docs}
162 def _normalize_unicode(self, text: str, form: Literal["NFC", "NFKC", "NFD", "NFKD"]) -> str:
163 """
164 Normalize the unicode of the text.
166 :param text: Text to normalize.
167 :param form: Unicode normalization form to apply to the text.
168 Options: "NFC", "NFKC", "NFD", "NFKD".
169 :returns: The normalized text.
170 """
171 return normalize(form, text)
173 def _ascii_only(self, text: str) -> str:
174 """
175 Convert the text to ASCII only.
177 Will remove accents from characters and replace them with ASCII characters.
178 Other non-ASCII characters will be removed.
180 :param text: Text to convert to ASCII only.
181 :returns: The text in ASCII only.
182 """
184 # First normalize the text to NFKD to separate the characters and their diacritics
185 # Then encode it to ASCII and ignore any characters that can't be encoded
186 return self._normalize_unicode(text, "NFKD").encode("ascii", "ignore").decode("utf-8")
188 def _remove_empty_lines(self, text: str) -> str:
189 """
190 Remove empty lines and lines that contain nothing but whitespaces from text.
192 :param text: Text to clean.
193 :returns: The text without empty lines.
194 """
195 pages = text.split("\f")
196 cleaned_pages = ["\n".join(line for line in page.split("\n") if line.strip()) for page in pages]
197 return "\f".join(cleaned_pages)
199 def _remove_extra_whitespaces(self, text: str) -> str:
200 """
201 Remove extra whitespaces from text.
203 :param text: Text to clean.
204 :returns: The text without extra whitespaces.
205 """
206 texts = text.split("\f")
207 cleaned_text = [re.sub(r"\s\s+", " ", text).strip() for text in texts]
208 return "\f".join(cleaned_text)
210 def _remove_regex(self, text: str, regex: str) -> str:
211 """
212 Remove substrings that match the specified regex from the text.
214 :param text: Text to clean.
215 :param regex: Regex to match and replace substrings by "".
216 :returns: The text without the substrings that match the regex.
217 """
218 texts = text.split("\f")
219 cleaned_text = [re.sub(regex, "", text).strip() for text in texts]
220 return "\f".join(cleaned_text)
222 def _replace_regexes(self, text: str, replace_regexes: dict[str, str]) -> str:
223 """
224 Replace substrings that match the specified regex patterns with custom replacement strings.
226 :param text: Text to clean.
227 :param replace_regexes: A dictionary mapping regex patterns to their replacement strings.
228 :returns: The text with the regex matches replaced by the specified strings.
229 """
230 pages = text.split("\f")
231 cleaned_pages = []
232 for page in pages:
233 for pattern, replacement in replace_regexes.items():
234 page = re.sub(pattern, replacement, page)
235 cleaned_pages.append(page)
236 return "\f".join(cleaned_pages)
238 def _remove_substrings(self, text: str, substrings: list[str]) -> str:
239 """
240 Remove all specified substrings from the text.
242 :param text: Text to clean.
243 :param substrings: Substrings to remove.
244 :returns: The text without the specified substrings.
245 """
246 for substring in substrings:
247 text = text.replace(substring, "")
248 return text
250 def _remove_repeated_substrings(self, text: str) -> str:
251 """
252 Remove any substrings from the text that occur repeatedly on every page. For example headers or footers.
254 Pages in the text need to be separated by form feed character "\f".
255 :param text: Text to clean.
256 :returns: The text without the repeated substrings.
257 """
258 return self._find_and_remove_header_footer(
259 text, n_chars=300, n_first_pages_to_ignore=1, n_last_pages_to_ignore=1
260 )
262 def _find_and_remove_header_footer(
263 self, text: str, n_chars: int, n_first_pages_to_ignore: int, n_last_pages_to_ignore: int
264 ) -> str:
265 """
266 Heuristic to find footers and headers across different pages by searching for the longest common string.
268 Pages in the text need to be separated by form feed character "\f".
269 For headers, we only search in the first n_chars characters (for footer: last n_chars).
270 Note: This heuristic uses exact matches and therefore works well for footers like "Copyright 2019 by XXX",
271 but won't detect "Page 3 of 4" or similar.
273 :param text: The text to remove headers and footers from, with pages separated by the
274 form feed character described above.
275 :param n_chars: The number of first/last characters where the header/footer shall be searched in.
276 :param n_first_pages_to_ignore: The number of first pages to ignore
277 (e.g. TOCs often don't contain footer/header).
278 :param n_last_pages_to_ignore: The number of last pages to ignore.
279 :returns: The text without the found headers and footers.
280 """
282 pages = text.split("\f")
284 # header
285 start_of_pages = [p[:n_chars] for p in pages[n_first_pages_to_ignore:-n_last_pages_to_ignore]]
286 found_header = self._find_longest_common_ngram(start_of_pages)
287 if found_header:
288 pages = [page.replace(found_header, "") for page in pages]
290 # footer
291 end_of_pages = [p[-n_chars:] for p in pages[n_first_pages_to_ignore:-n_last_pages_to_ignore]]
292 found_footer = self._find_longest_common_ngram(end_of_pages)
293 if found_footer:
294 pages = [page.replace(found_footer, "") for page in pages]
296 logger.debug(
297 "Removed header '{header}' and footer '{footer}' in document", header=found_header, footer=found_footer
298 )
299 return "\f".join(pages)
301 def _ngram(self, seq: str, n: int) -> Generator[str, None, None]:
302 """
303 Return all ngrams of length n from a text sequence. Each ngram consists of n words split by whitespace.
305 :param seq: The sequence to generate ngrams from.
306 :param n: The length of the ngrams to generate.
307 :returns: A Generator generating all ngrams of length n from the given sequence.
308 """
310 # In order to maintain the original whitespace, but still consider \n and \t for n-gram tokenization,
311 # we add a space here and remove it after creation of the ngrams again (see below)
312 seq = seq.replace("\n", " \n")
313 seq = seq.replace("\t", " \t")
315 words = seq.split(" ")
316 return (" ".join(words[i : i + n]).replace(" \n", "\n").replace(" \t", "\t") for i in range(len(words) - n + 1))
318 def _allngram(self, seq: str, min_ngram: int, max_ngram: int) -> set[str]:
319 """
320 Generates all possible ngrams from a given sequence of text.
322 Considering all ngram lengths between the minimum and maximum length.
324 :param seq: The sequence to generate ngrams from.
325 :param min_ngram: The minimum length of ngram to consider.
326 :param max_ngram: The maximum length of ngram to consider.
327 :returns: A set of all ngrams from the given sequence.
328 """
329 lengths = range(min_ngram, max_ngram) if max_ngram else range(min_ngram, len(seq))
330 ngrams = map(partial(self._ngram, seq), lengths)
331 return set(chain.from_iterable(ngrams))
333 def _find_longest_common_ngram(self, sequences: list[str], min_ngram: int = 3, max_ngram: int = 30) -> str:
334 """
335 Find the longest common ngram across a list of text sequences (e.g. start of pages).
337 Considering all ngram lengths between the minimum and maximum length. Helpful for finding footers, headers etc.
338 Empty sequences are ignored.
340 :param sequences: The list of strings that shall be searched for common n_grams.
341 :param max_ngram: The maximum length of ngram to consider.
342 :param min_ngram: The minimum length of ngram to consider.
343 :returns: The longest ngram that all sequences have in common.
344 """
345 sequences = [s for s in sequences if s] # filter empty sequences
346 if len(sequences) < 2:
347 # a single sequence has no ngram "in common" with any other; treating
348 # its own longest ngram as a repeated header/footer would wipe it
349 return ""
350 seqs_ngrams = map(partial(self._allngram, min_ngram=min_ngram, max_ngram=max_ngram), sequences)
351 intersection = reduce(set.intersection, seqs_ngrams)
353 longest = max(intersection, key=len, default="")
354 return longest if longest.strip() else ""