Coverage for haystack/components/converters/docx.py: 99%
146 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 csv
6import io
7import os
8from dataclasses import asdict, dataclass
9from enum import Enum
10from io import StringIO
11from pathlib import Path
12from typing import Any
14from haystack import Document, component, default_from_dict, default_to_dict, logging
15from haystack.components.converters.utils import LinkFormat, get_bytestream_from_source, normalize_metadata
16from haystack.dataclasses import ByteStream
17from haystack.lazy_imports import LazyImport
19logger = logging.getLogger(__name__)
21with LazyImport("Run 'pip install python-docx'") as docx_import:
22 import docx
23 from docx.document import Document as DocxDocument
24 from docx.table import Table
25 from docx.text.hyperlink import Hyperlink
26 from docx.text.paragraph import Paragraph
27 from docx.text.run import Run
28 from lxml.etree import _Comment
31@dataclass
32class DOCXMetadata:
33 """
34 Describes the metadata of Docx file.
36 :param author: The author
37 :param category: The category
38 :param comments: The comments
39 :param content_status: The content status
40 :param created: The creation date (ISO formatted string)
41 :param identifier: The identifier
42 :param keywords: Available keywords
43 :param language: The language of the document
44 :param last_modified_by: User who last modified the document
45 :param last_printed: The last printed date (ISO formatted string)
46 :param modified: The last modification date (ISO formatted string)
47 :param revision: The revision number
48 :param subject: The subject
49 :param title: The title
50 :param version: The version
51 """
53 author: str
54 category: str
55 comments: str
56 content_status: str
57 created: str | None
58 identifier: str
59 keywords: str
60 language: str
61 last_modified_by: str
62 last_printed: str | None
63 modified: str | None
64 revision: int
65 subject: str
66 title: str
67 version: str
70class DOCXTableFormat(Enum):
71 """
72 Supported formats for storing DOCX tabular data in a Document.
73 """
75 MARKDOWN = "markdown"
76 CSV = "csv"
78 def __str__(self) -> str:
79 return self.value
81 @staticmethod
82 def from_str(string: str) -> "DOCXTableFormat":
83 """
84 Convert a string to a DOCXTableFormat enum.
85 """
86 enum_map = {e.value: e for e in DOCXTableFormat}
87 table_format = enum_map.get(string.lower())
88 if table_format is None:
89 msg = f"Unknown table format '{string}'. Supported formats are: {list(enum_map.keys())}"
90 raise ValueError(msg)
91 return table_format
94DOCXLinkFormat = LinkFormat
97@component
98class DOCXToDocument:
99 """
100 Converts DOCX files to Documents.
102 Uses `python-docx` library to convert the DOCX file to a document.
103 This component does not preserve page breaks in the original document.
105 Usage example:
107 ```python
108 from haystack.components.converters.docx import DOCXToDocument, DOCXTableFormat, DOCXLinkFormat
109 from datetime import datetime
111 converter = DOCXToDocument(table_format=DOCXTableFormat.CSV, link_format=DOCXLinkFormat.MARKDOWN)
112 results = converter.run(
113 sources=["test/test_files/docx/sample_docx.docx"], meta={"date_added": datetime.now().isoformat()}
114 )
115 documents = results["documents"]
117 print(documents[0].content)
118 # >> 'This is a text from the DOCX file.'
119 ```
120 """
122 def __init__(
123 self,
124 table_format: str | DOCXTableFormat = DOCXTableFormat.CSV,
125 link_format: str | DOCXLinkFormat = DOCXLinkFormat.NONE,
126 store_full_path: bool = False,
127 ) -> None:
128 """
129 Create a DOCXToDocument component.
131 :param table_format: The format for table output. Can be either DOCXTableFormat.MARKDOWN,
132 DOCXTableFormat.CSV, "markdown", or "csv".
133 :param link_format: The format for link output. Can be either:
134 DOCXLinkFormat.MARKDOWN or "markdown" to get `[text](address)`,
135 DOCXLinkFormat.PLAIN or "plain" to get text (address),
136 DOCXLinkFormat.NONE or "none" to get text without links.
137 :param store_full_path:
138 If True, the full path of the file is stored in the metadata of the document.
139 If False, only the file name is stored.
140 """
141 docx_import.check()
142 self.table_format = DOCXTableFormat.from_str(table_format) if isinstance(table_format, str) else table_format
143 self.link_format = DOCXLinkFormat.from_str(link_format) if isinstance(link_format, str) else link_format
144 self.store_full_path = store_full_path
146 def to_dict(self) -> dict[str, Any]:
147 """
148 Serializes the component to a dictionary.
150 :returns:
151 Dictionary with serialized data.
152 """
153 return default_to_dict(
154 self,
155 table_format=str(self.table_format),
156 link_format=str(self.link_format),
157 store_full_path=self.store_full_path,
158 )
160 @classmethod
161 def from_dict(cls, data: dict[str, Any]) -> "DOCXToDocument":
162 """
163 Deserializes the component from a dictionary.
165 :param data:
166 The dictionary to deserialize from.
167 :returns:
168 The deserialized component.
169 """
170 if "table_format" in data["init_parameters"]:
171 data["init_parameters"]["table_format"] = DOCXTableFormat.from_str(data["init_parameters"]["table_format"])
172 if "link_format" in data["init_parameters"]:
173 data["init_parameters"]["link_format"] = DOCXLinkFormat.from_str(data["init_parameters"]["link_format"])
174 return default_from_dict(cls, data)
176 @component.output_types(documents=list[Document])
177 def run(
178 self, sources: list[str | Path | ByteStream], meta: dict[str, Any] | list[dict[str, Any]] | None = None
179 ) -> dict[str, Any]:
180 """
181 Converts DOCX files to Documents.
183 :param sources:
184 List of file paths or ByteStream objects.
185 :param meta:
186 Optional metadata to attach to the Documents.
187 This value can be either a list of dictionaries or a single dictionary.
188 If it's a single dictionary, its content is added to the metadata of all produced Documents.
189 If it's a list, the length of the list must match the number of sources, because the two lists will
190 be zipped.
191 If `sources` contains ByteStream objects, their `meta` will be added to the output Documents.
192 :returns:
193 A dictionary with the following keys:
194 - `documents`: Created Documents
195 """
196 documents = []
197 meta_list = normalize_metadata(meta=meta, sources_count=len(sources))
199 for source, metadata in zip(sources, meta_list, strict=True):
200 try:
201 bytestream = get_bytestream_from_source(source)
202 except Exception as e:
203 logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=e)
204 continue
205 try:
206 docx_document = docx.Document(io.BytesIO(bytestream.data))
207 elements = self._extract_elements(docx_document)
208 text = "\n".join(elements)
209 except Exception as e:
210 logger.warning(
211 "Could not read {source} and convert it to a DOCX Document, skipping. Error: {error}",
212 source=source,
213 error=e,
214 )
215 continue
217 docx_metadata = asdict(self._get_docx_metadata(document=docx_document))
218 merged_metadata = {**bytestream.meta, **metadata, "docx": docx_metadata}
220 if not self.store_full_path and "file_path" in bytestream.meta:
221 file_path = bytestream.meta.get("file_path")
222 if file_path: # Ensure the value is not None for mypy
223 merged_metadata["file_path"] = os.path.basename(file_path)
225 document = Document(content=text, meta=merged_metadata)
226 documents.append(document)
228 return {"documents": documents}
230 def _extract_elements(self, document: "DocxDocument") -> list[str]:
231 """
232 Extracts elements from a DOCX file.
234 :param document: The DOCX Document object.
235 :returns: List of strings (paragraph texts and table representations) with page breaks added as '\f' characters.
236 """
237 elements = []
238 for element in document.element.body:
239 if isinstance(element, _Comment):
240 continue
241 if element.tag.endswith("p"):
242 paragraph = Paragraph(element, document)
243 if paragraph.contains_page_break:
244 para_text = self._process_paragraph_with_page_breaks(paragraph)
245 else:
246 para_text = self._process_links_in_paragraph(paragraph)
247 elements.append(para_text)
248 elif element.tag.endswith("tbl"):
249 table = docx.table.Table(element, document)
250 table_str = (
251 self._table_to_markdown(table)
252 if self.table_format == DOCXTableFormat.MARKDOWN
253 else self._table_to_csv(table)
254 )
255 elements.append(table_str)
257 return elements
259 def _process_paragraph_with_page_breaks(self, paragraph: "Paragraph") -> str:
260 """
261 Processes a paragraph with page breaks.
263 :param paragraph: The DOCX paragraph to process.
264 :returns: A string with page breaks added as '\f' characters.
265 """
266 para_text = ""
267 # Usually, just 1 page break exists, but could be more if paragraph is really long, so we loop over them
268 for pb_index, page_break in enumerate(paragraph.rendered_page_breaks):
269 # Can only extract text from first paragraph page break, unfortunately
270 if pb_index == 0:
271 if page_break.preceding_paragraph_fragment:
272 para_text += self._process_links_in_paragraph(page_break.preceding_paragraph_fragment)
273 para_text += "\f"
274 if page_break.following_paragraph_fragment:
275 # following_paragraph_fragment contains all text for remainder of paragraph.
276 # However, if the remainder of the paragraph spans multiple page breaks, it won't include
277 # those later page breaks so we have to add them at end of text in the `else` block below.
278 # This is not ideal, but this case should be very rare and this is likely good enough.
279 para_text += self._process_links_in_paragraph(page_break.following_paragraph_fragment)
280 else:
281 para_text += "\f"
282 return para_text
284 def _process_links_in_paragraph(self, paragraph: "Paragraph") -> str:
285 """
286 Processes links in a paragraph and formats them according to the specified link format.
288 :param paragraph: The DOCX paragraph to process.
289 :returns: A string with links formatted according to the specified format.
290 """
291 if self.link_format == DOCXLinkFormat.NONE:
292 return paragraph.text
293 text = ""
294 # Iterate over all hyperlinks and other content in the paragraph
295 # https://python-docx.readthedocs.io/en/latest/api/text.html#docx.text.paragraph.Paragraph.iter_inner_content
296 for content in paragraph.iter_inner_content():
297 if isinstance(content, Run):
298 text += content.text
299 elif isinstance(content, Hyperlink):
300 if self.link_format == DOCXLinkFormat.MARKDOWN:
301 formatted_link = f"[{content.text}]({content.address})"
302 else: # PLAIN format
303 formatted_link = f"{content.text} ({content.address})"
304 text += formatted_link
306 return text
308 def _table_to_markdown(self, table: "Table") -> str:
309 """
310 Converts a DOCX table to a Markdown string.
312 :param table: The DOCX table to convert.
313 :returns: A Markdown string representation of the table.
314 """
315 markdown: list[str] = []
316 max_col_widths: list[int] = []
318 # Calculate max width for each column
319 for row in table.rows:
320 for i, cell in enumerate(row.cells):
321 cell_text = cell.text.strip()
322 if i >= len(max_col_widths):
323 max_col_widths.append(len(cell_text))
324 else:
325 max_col_widths[i] = max(max_col_widths[i], len(cell_text))
327 # Process rows
328 for i, row in enumerate(table.rows):
329 md_row = [cell.text.strip().ljust(max_col_widths[j]) for j, cell in enumerate(row.cells)]
330 markdown.append("| " + " | ".join(md_row) + " |")
332 # Add separator after header row
333 if i == 0:
334 separator = ["-" * max_col_widths[j] for j in range(len(row.cells))]
335 markdown.append("| " + " | ".join(separator) + " |")
337 return "\n".join(markdown)
339 def _table_to_csv(self, table: "Table") -> str:
340 """
341 Converts a DOCX table to a CSV string.
343 :param table: The DOCX table to convert.
344 :returns: A CSV string representation of the table.
345 """
346 csv_output = StringIO()
347 csv_writer = csv.writer(csv_output, quoting=csv.QUOTE_MINIMAL)
349 # Process rows
350 for row in table.rows:
351 csv_row = [cell.text.strip() for cell in row.cells]
352 csv_writer.writerow(csv_row)
354 # Get the CSV as a string and strip any trailing newlines
355 csv_string = csv_output.getvalue().strip()
356 csv_output.close()
358 return csv_string
360 def _get_docx_metadata(self, document: "DocxDocument") -> DOCXMetadata:
361 """
362 Get all relevant data from the 'core_properties' attribute from a DOCX Document.
364 :param document:
365 The DOCX Document you want to extract metadata from
367 :returns:
368 A `DOCXMetadata` dataclass all the relevant fields from the 'core_properties'
369 """
370 return DOCXMetadata(
371 author=document.core_properties.author,
372 category=document.core_properties.category,
373 comments=document.core_properties.comments,
374 content_status=document.core_properties.content_status,
375 created=(document.core_properties.created.isoformat() if document.core_properties.created else None),
376 identifier=document.core_properties.identifier,
377 keywords=document.core_properties.keywords,
378 language=document.core_properties.language,
379 last_modified_by=document.core_properties.last_modified_by,
380 last_printed=(
381 document.core_properties.last_printed.isoformat() if document.core_properties.last_printed else None
382 ),
383 modified=(document.core_properties.modified.isoformat() if document.core_properties.modified else None),
384 revision=document.core_properties.revision,
385 subject=document.core_properties.subject,
386 title=document.core_properties.title,
387 version=document.core_properties.version,
388 )