Coverage for haystack/components/converters/markdown.py: 86%
71 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 json
6import os
7import re
8from pathlib import Path
9from typing import Any
11import yaml
12from tqdm import tqdm
14from haystack import Document, component, logging
15from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata
16from haystack.dataclasses import ByteStream
17from haystack.lazy_imports import LazyImport
19with LazyImport("Run 'pip install markdown-it-py mdit_plain'") as markdown_conversion_imports:
20 from markdown_it import MarkdownIt
21 from mdit_plain.renderer import RendererPlain
24logger = logging.getLogger(__name__)
26_FRONTMATTER_PATTERN = re.compile(r"\A---[ \t]*\r?\n(?P<frontmatter>.*?)(?:\r?\n)---[ \t]*(?:\r?\n|$)", re.DOTALL)
29@component
30class MarkdownToDocument:
31 """
32 Converts a Markdown file into a text Document.
34 Usage example:
36 ```python
37 from haystack.components.converters import MarkdownToDocument
38 from datetime import datetime
40 converter = MarkdownToDocument()
41 results = converter.run(
42 sources=["test/test_files/markdown/sample.md"], meta={"date_added": datetime.now().isoformat()}
43 )
44 documents = results["documents"]
45 print(documents[0].content)
46 # 'This is a text from the markdown file.'
47 ```
48 """
50 def __init__(
51 self,
52 table_to_single_line: bool = False,
53 progress_bar: bool = True,
54 store_full_path: bool = False,
55 encoding: str = "utf-8",
56 *,
57 extract_frontmatter: bool = False,
58 ) -> None:
59 """
60 Create a MarkdownToDocument component.
62 :param table_to_single_line:
63 If True converts table contents into a single line.
64 :param progress_bar:
65 If True shows a progress bar when running.
66 :param store_full_path:
67 If True, the full path of the file is stored in the metadata of the document.
68 If False, only the file name is stored.
69 :param encoding:
70 The default encoding to use when converting Markdown files. If the encoding is specified in the metadata
71 of a source ByteStream, it overrides this value.
72 :param extract_frontmatter:
73 If True, YAML frontmatter at the beginning of the Markdown file is
74 removed from the document content and added to the document metadata.
75 """
76 markdown_conversion_imports.check()
78 self.table_to_single_line = table_to_single_line
79 self.progress_bar = progress_bar
80 self.store_full_path = store_full_path
81 self.encoding = encoding
82 self.extract_frontmatter = extract_frontmatter
84 @component.output_types(documents=list[Document])
85 def run(
86 self, sources: list[str | Path | ByteStream], meta: dict[str, Any] | list[dict[str, Any]] | None = None
87 ) -> dict[str, Any]:
88 """
89 Converts a list of Markdown files to Documents.
91 :param sources:
92 List of file paths or ByteStream objects.
93 :param meta:
94 Optional metadata to attach to the Documents.
95 This value can be either a list of dictionaries or a single dictionary.
96 If it's a single dictionary, its content is added to the metadata of all produced Documents.
97 If it's a list, the length of the list must match the number of sources, because the two lists will
98 be zipped.
99 If `sources` contains ByteStream objects, their `meta` will be added to the output Documents.
101 :returns:
102 A dictionary with the following keys:
103 - `documents`: List of created Documents
104 """
105 parser = MarkdownIt(renderer_cls=RendererPlain)
106 if self.table_to_single_line:
107 parser.enable("table")
109 documents = []
110 meta_list = normalize_metadata(meta=meta, sources_count=len(sources))
112 for source, metadata in tqdm(
113 zip(sources, meta_list, strict=True),
114 total=len(sources),
115 desc="Converting markdown files to Documents",
116 disable=not self.progress_bar,
117 ):
118 try:
119 bytestream = get_bytestream_from_source(source)
120 except Exception as e:
121 logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=e)
122 continue
123 try:
124 encoding = bytestream.meta.get("encoding", self.encoding)
125 file_content = bytestream.data.decode(encoding)
126 file_content, frontmatter = self._extract_frontmatter(file_content, source)
127 text = parser.render(file_content)
128 except Exception as conversion_e:
129 logger.warning(
130 "Failed to extract text from {source}. Skipping it. Error: {error}",
131 source=source,
132 error=conversion_e,
133 )
134 continue
136 merged_metadata = {**bytestream.meta, **frontmatter, **metadata}
138 if not self.store_full_path and (file_path := bytestream.meta.get("file_path")):
139 merged_metadata["file_path"] = os.path.basename(file_path)
141 document = Document(content=text, meta=merged_metadata)
142 documents.append(document)
144 return {"documents": documents}
146 def _extract_frontmatter(self, file_content: str, source: str | Path | ByteStream) -> tuple[str, dict[str, Any]]:
147 if not self.extract_frontmatter:
148 return file_content, {}
150 match = _FRONTMATTER_PATTERN.match(file_content)
151 if not match:
152 return file_content, {}
154 frontmatter_text = match.group("frontmatter")
155 try:
156 frontmatter = json.loads(json.dumps(yaml.safe_load(frontmatter_text), default=str)) or {}
157 except yaml.YAMLError as error:
158 logger.warning(
159 "Could not parse YAML frontmatter in {source}. Keeping it as content. Error: {error}",
160 source=source,
161 error=error,
162 )
163 return file_content, {}
164 except (TypeError, ValueError) as error:
165 logger.warning(
166 "Could not convert YAML frontmatter in {source}. Keeping it as content. Error: {error}",
167 source=source,
168 error=error,
169 )
170 return file_content, {}
172 if not isinstance(frontmatter, dict):
173 logger.warning(
174 "Ignoring YAML frontmatter in {source}: expected a mapping, got {kind}.",
175 source=source,
176 kind=type(frontmatter).__name__,
177 )
178 return file_content, {}
180 return file_content[match.end() :], frontmatter