Coverage for haystack/components/converters/docx.py: 99%
172 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 csv
6import io
7import os
8import re
9from dataclasses import asdict, dataclass
10from enum import Enum
11from io import StringIO
12from pathlib import Path
13from typing import Any
15from haystack import Document, component, default_from_dict, default_to_dict, logging
16from haystack.components.converters.utils import LinkFormat, get_bytestream_from_source, normalize_metadata
17from haystack.dataclasses import ByteStream
18from haystack.lazy_imports import LazyImport
20logger = logging.getLogger(__name__)
22_WORD_NS = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
23_MARKUP_COMPATIBILITY_NS = "http://schemas.openxmlformats.org/markup-compatibility/2006"
24# A text box keeps its paragraphs here, nested under the drawing inside a run.
25_TEXT_BOX_CONTENT_TAG = f"{{{_WORD_NS}}}txbxContent"
26# Word writes a text box twice inside `mc:AlternateContent`: a `wps:txbx` under
27# `mc:Choice` and the same text as a VML text box under `mc:Fallback`.
28_FALLBACK_TAG = f"{{{_MARKUP_COMPATIBILITY_NS}}}Fallback"
30# A Markdown table row ends at a line break and its columns are separated by pipes, so
31# neither can survive inside a cell.
32_MARKDOWN_CELL_BREAK_PATTERN = re.compile(r"\s*(?:\r\n|\r|\n)\s*")
33_MARKDOWN_CELL_PIPE_PATTERN = re.compile(r"(?<!\\)(\\*)\|")
35with LazyImport("Run 'pip install python-docx'") as docx_import:
36 import docx
37 from docx.document import Document as DocxDocument
38 from docx.table import Table, _Cell
39 from docx.text.hyperlink import Hyperlink
40 from docx.text.paragraph import Paragraph
41 from docx.text.run import Run
42 from lxml.etree import _Comment
45@dataclass
46class DOCXMetadata:
47 """
48 Describes the metadata of Docx file.
50 :param author: The author
51 :param category: The category
52 :param comments: The comments
53 :param content_status: The content status
54 :param created: The creation date (ISO formatted string)
55 :param identifier: The identifier
56 :param keywords: Available keywords
57 :param language: The language of the document
58 :param last_modified_by: User who last modified the document
59 :param last_printed: The last printed date (ISO formatted string)
60 :param modified: The last modification date (ISO formatted string)
61 :param revision: The revision number
62 :param subject: The subject
63 :param title: The title
64 :param version: The version
65 """
67 author: str
68 category: str
69 comments: str
70 content_status: str
71 created: str | None
72 identifier: str
73 keywords: str
74 language: str
75 last_modified_by: str
76 last_printed: str | None
77 modified: str | None
78 revision: int
79 subject: str
80 title: str
81 version: str
84class DOCXTableFormat(Enum):
85 """
86 Supported formats for storing DOCX tabular data in a Document.
87 """
89 MARKDOWN = "markdown"
90 CSV = "csv"
92 def __str__(self) -> str:
93 return self.value
95 @staticmethod
96 def from_str(string: str) -> "DOCXTableFormat":
97 """
98 Convert a string to a DOCXTableFormat enum.
99 """
100 enum_map = {e.value: e for e in DOCXTableFormat}
101 table_format = enum_map.get(string.lower())
102 if table_format is None:
103 msg = f"Unknown table format '{string}'. Supported formats are: {list(enum_map.keys())}"
104 raise ValueError(msg)
105 return table_format
108DOCXLinkFormat = LinkFormat
111@component
112class DOCXToDocument:
113 """
114 Converts DOCX files to Documents.
116 Uses `python-docx` library to convert the DOCX file to a document.
117 This component does not preserve page breaks in the original document.
119 Usage example:
121 ```python
122 from haystack.components.converters.docx import DOCXToDocument, DOCXTableFormat, DOCXLinkFormat
123 from datetime import datetime
125 converter = DOCXToDocument(table_format=DOCXTableFormat.CSV, link_format=DOCXLinkFormat.MARKDOWN)
126 results = converter.run(
127 sources=["test/test_files/docx/sample_docx.docx"], meta={"date_added": datetime.now().isoformat()}
128 )
129 documents = results["documents"]
131 print(documents[0].content)
132 # >> 'This is a text from the DOCX file.'
133 ```
134 """
136 def __init__(
137 self,
138 table_format: str | DOCXTableFormat = DOCXTableFormat.CSV,
139 link_format: str | DOCXLinkFormat = DOCXLinkFormat.NONE,
140 store_full_path: bool = False,
141 ) -> None:
142 """
143 Create a DOCXToDocument component.
145 :param table_format: The format for table output. Can be either DOCXTableFormat.MARKDOWN,
146 DOCXTableFormat.CSV, "markdown", or "csv".
147 :param link_format: The format for link output. Can be either:
148 DOCXLinkFormat.MARKDOWN or "markdown" to get `[text](address)`,
149 DOCXLinkFormat.PLAIN or "plain" to get text (address),
150 DOCXLinkFormat.NONE or "none" to get text without links.
151 :param store_full_path:
152 If True, the full path of the file is stored in the metadata of the document.
153 If False, only the file name is stored.
154 """
155 docx_import.check()
156 self.table_format = DOCXTableFormat.from_str(table_format) if isinstance(table_format, str) else table_format
157 self.link_format = DOCXLinkFormat.from_str(link_format) if isinstance(link_format, str) else link_format
158 self.store_full_path = store_full_path
160 def to_dict(self) -> dict[str, Any]:
161 """
162 Serializes the component to a dictionary.
164 :returns:
165 Dictionary with serialized data.
166 """
167 return default_to_dict(
168 self,
169 table_format=str(self.table_format),
170 link_format=str(self.link_format),
171 store_full_path=self.store_full_path,
172 )
174 @classmethod
175 def from_dict(cls, data: dict[str, Any]) -> "DOCXToDocument":
176 """
177 Deserializes the component from a dictionary.
179 :param data:
180 The dictionary to deserialize from.
181 :returns:
182 The deserialized component.
183 """
184 if "table_format" in data["init_parameters"]:
185 data["init_parameters"]["table_format"] = DOCXTableFormat.from_str(data["init_parameters"]["table_format"])
186 if "link_format" in data["init_parameters"]:
187 data["init_parameters"]["link_format"] = DOCXLinkFormat.from_str(data["init_parameters"]["link_format"])
188 return default_from_dict(cls, data)
190 @component.output_types(documents=list[Document])
191 def run(
192 self, sources: list[str | Path | ByteStream], meta: dict[str, Any] | list[dict[str, Any]] | None = None
193 ) -> dict[str, Any]:
194 """
195 Converts DOCX files to Documents.
197 :param sources:
198 List of file paths or ByteStream objects.
199 :param meta:
200 Optional metadata to attach to the Documents.
201 This value can be either a list of dictionaries or a single dictionary.
202 If it's a single dictionary, its content is added to the metadata of all produced Documents.
203 If it's a list, the length of the list must match the number of sources, because the two lists will
204 be zipped.
205 If `sources` contains ByteStream objects, their `meta` will be added to the output Documents.
206 :returns:
207 A dictionary with the following keys:
208 - `documents`: Created Documents
209 """
210 documents = []
211 meta_list = normalize_metadata(meta=meta, sources_count=len(sources))
213 for source, metadata in zip(sources, meta_list, strict=True):
214 try:
215 bytestream = get_bytestream_from_source(source)
216 except Exception as e:
217 logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=e)
218 continue
219 try:
220 docx_document = docx.Document(io.BytesIO(bytestream.data))
221 elements = self._extract_elements(docx_document)
222 text = "\n".join(elements)
223 except Exception as e:
224 logger.warning(
225 "Could not read {source} and convert it to a DOCX Document, skipping. Error: {error}",
226 source=source,
227 error=e,
228 )
229 continue
231 docx_metadata = asdict(self._get_docx_metadata(document=docx_document))
232 merged_metadata = {**bytestream.meta, **metadata, "docx": docx_metadata}
234 if not self.store_full_path and "file_path" in bytestream.meta:
235 file_path = bytestream.meta.get("file_path")
236 if file_path: # Ensure the value is not None for mypy
237 merged_metadata["file_path"] = os.path.basename(file_path)
239 document = Document(content=text, meta=merged_metadata)
240 documents.append(document)
242 return {"documents": documents}
244 def _extract_elements(self, document: "DocxDocument") -> list[str]:
245 """
246 Extracts elements from a DOCX file.
248 :param document: The DOCX Document object.
249 :returns: List of strings (paragraph texts and table representations) with page breaks added as '\f' characters.
250 """
251 elements = []
252 for element in document.element.body:
253 if isinstance(element, _Comment):
254 continue
255 if element.tag.endswith("p"):
256 paragraph = Paragraph(element, document)
257 if paragraph.contains_page_break:
258 para_text = self._process_paragraph_with_page_breaks(paragraph)
259 else:
260 para_text = self._process_links_in_paragraph(paragraph)
261 elements.append(para_text)
262 elements.extend(self._extract_text_boxes(element, document))
263 elif element.tag.endswith("tbl"):
264 table = docx.table.Table(element, document)
265 table_str = (
266 self._table_to_markdown(table)
267 if self.table_format == DOCXTableFormat.MARKDOWN
268 else self._table_to_csv(table)
269 )
270 elements.append(table_str)
272 return elements
274 def _extract_text_boxes(self, element: Any, document: "DocxDocument") -> list[str]:
275 """
276 Extracts the text of any text box anchored to a paragraph.
278 A text box holds its own paragraphs and tables under `w:txbxContent`, nested in
279 the drawing inside a run, so the anchoring paragraph's text never reaches them.
281 :param element: The `w:p` element to look under.
282 :param document: The DOCX Document object.
283 :returns: List of strings, one per paragraph or table found inside a text box.
284 """
285 elements = []
286 for text_box in element.iter(_TEXT_BOX_CONTENT_TAG):
287 # Skip the `mc:Fallback` copy, or the same text is emitted twice.
288 if any(ancestor.tag == _FALLBACK_TAG for ancestor in text_box.iterancestors()):
289 continue
290 for child in text_box:
291 if child.tag.endswith("p"):
292 elements.append(self._process_links_in_paragraph(Paragraph(child, document)))
293 elif child.tag.endswith("tbl"):
294 table = docx.table.Table(child, document)
295 elements.append(
296 self._table_to_markdown(table)
297 if self.table_format == DOCXTableFormat.MARKDOWN
298 else self._table_to_csv(table)
299 )
300 return elements
302 def _process_paragraph_with_page_breaks(self, paragraph: "Paragraph") -> str:
303 """
304 Processes a paragraph with page breaks.
306 :param paragraph: The DOCX paragraph to process.
307 :returns: A string with page breaks added as '\f' characters.
308 """
309 para_text = ""
310 # Usually, just 1 page break exists, but could be more if paragraph is really long, so we loop over them
311 for pb_index, page_break in enumerate(paragraph.rendered_page_breaks):
312 # Can only extract text from first paragraph page break, unfortunately
313 if pb_index == 0:
314 if page_break.preceding_paragraph_fragment:
315 para_text += self._process_links_in_paragraph(page_break.preceding_paragraph_fragment)
316 para_text += "\f"
317 if page_break.following_paragraph_fragment:
318 # following_paragraph_fragment contains all text for remainder of paragraph.
319 # However, if the remainder of the paragraph spans multiple page breaks, it won't include
320 # those later page breaks so we have to add them at end of text in the `else` block below.
321 # This is not ideal, but this case should be very rare and this is likely good enough.
322 para_text += self._process_links_in_paragraph(page_break.following_paragraph_fragment)
323 else:
324 para_text += "\f"
325 return para_text
327 def _process_links_in_paragraph(self, paragraph: "Paragraph") -> str:
328 """
329 Processes links in a paragraph and formats them according to the specified link format.
331 :param paragraph: The DOCX paragraph to process.
332 :returns: A string with links formatted according to the specified format.
333 """
334 if self.link_format == DOCXLinkFormat.NONE:
335 return paragraph.text
336 text = ""
337 # Iterate over all hyperlinks and other content in the paragraph
338 # https://python-docx.readthedocs.io/en/latest/api/text.html#docx.text.paragraph.Paragraph.iter_inner_content
339 for content in paragraph.iter_inner_content():
340 if isinstance(content, Run):
341 text += content.text
342 elif isinstance(content, Hyperlink):
343 if self.link_format == DOCXLinkFormat.MARKDOWN:
344 formatted_link = f"[{content.text}]({content.address})"
345 else: # PLAIN format
346 formatted_link = f"{content.text} ({content.address})"
347 text += formatted_link
349 return text
351 @staticmethod
352 def _escape_markdown_cell(text: str) -> str:
353 """
354 Makes a cell's text safe to put between the pipes of a Markdown table row.
356 A cell spanning several paragraphs arrives with newlines in it, which would end
357 the row in the middle, and a pipe in a cell would be read as a column separator.
359 :param text: The cell text.
360 :returns: The text with line breaks collapsed and pipes escaped.
361 """
362 text = _MARKDOWN_CELL_BREAK_PATTERN.sub(" ", text)
363 # The backslash run in front of the pipe is doubled first, so a backslash the
364 # cell already contains cannot consume the escape.
365 return _MARKDOWN_CELL_PIPE_PATTERN.sub(lambda match: match.group(1) * 2 + r"\|", text)
367 def _cell_text(self, cell: "_Cell") -> str:
368 """
369 Returns a table cell's text with links formatted like links in body paragraphs.
371 :param cell: The DOCX table cell.
372 :returns: The cell's paragraphs joined by newlines, as `cell.text` joins them.
373 """
374 return "\n".join(self._process_links_in_paragraph(paragraph) for paragraph in cell.paragraphs)
376 def _table_to_markdown(self, table: "Table") -> str:
377 """
378 Converts a DOCX table to a Markdown string.
380 :param table: The DOCX table to convert.
381 :returns: A Markdown string representation of the table.
382 """
383 markdown: list[str] = []
384 max_col_widths: list[int] = []
386 # Calculate max width for each column, on the escaped text that is written out
387 for row in table.rows:
388 for i, cell in enumerate(row.cells):
389 cell_text = self._escape_markdown_cell(self._cell_text(cell).strip())
390 if i >= len(max_col_widths):
391 max_col_widths.append(len(cell_text))
392 else:
393 max_col_widths[i] = max(max_col_widths[i], len(cell_text))
395 # Process rows
396 for i, row in enumerate(table.rows):
397 md_row = [
398 self._escape_markdown_cell(self._cell_text(cell).strip()).ljust(max_col_widths[j])
399 for j, cell in enumerate(row.cells)
400 ]
401 markdown.append("| " + " | ".join(md_row) + " |")
403 # Add separator after header row
404 if i == 0:
405 separator = ["-" * max_col_widths[j] for j in range(len(row.cells))]
406 markdown.append("| " + " | ".join(separator) + " |")
408 return "\n".join(markdown)
410 def _table_to_csv(self, table: "Table") -> str:
411 """
412 Converts a DOCX table to a CSV string.
414 :param table: The DOCX table to convert.
415 :returns: A CSV string representation of the table.
416 """
417 csv_output = StringIO()
418 csv_writer = csv.writer(csv_output, quoting=csv.QUOTE_MINIMAL)
420 # Process rows
421 for row in table.rows:
422 csv_row = [self._cell_text(cell).strip() for cell in row.cells]
423 csv_writer.writerow(csv_row)
425 # Get the CSV as a string and strip any trailing newlines
426 csv_string = csv_output.getvalue().strip()
427 csv_output.close()
429 return csv_string
431 def _get_docx_metadata(self, document: "DocxDocument") -> DOCXMetadata:
432 """
433 Get all relevant data from the 'core_properties' attribute from a DOCX Document.
435 :param document:
436 The DOCX Document you want to extract metadata from
438 :returns:
439 A `DOCXMetadata` dataclass all the relevant fields from the 'core_properties'
440 """
441 return DOCXMetadata(
442 author=document.core_properties.author,
443 category=document.core_properties.category,
444 comments=document.core_properties.comments,
445 content_status=document.core_properties.content_status,
446 created=(document.core_properties.created.isoformat() if document.core_properties.created else None),
447 identifier=document.core_properties.identifier,
448 keywords=document.core_properties.keywords,
449 language=document.core_properties.language,
450 last_modified_by=document.core_properties.last_modified_by,
451 last_printed=(
452 document.core_properties.last_printed.isoformat() if document.core_properties.last_printed else None
453 ),
454 modified=(document.core_properties.modified.isoformat() if document.core_properties.modified else None),
455 revision=document.core_properties.revision,
456 subject=document.core_properties.subject,
457 title=document.core_properties.title,
458 version=document.core_properties.version,
459 )