Coverage for haystack/components/converters/pptx.py: 97%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import io 

6import os 

7from pathlib import Path 

8from typing import Any, Literal 

9 

10from haystack import Document, component, default_to_dict, logging 

11from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata 

12from haystack.dataclasses import ByteStream 

13from haystack.lazy_imports import LazyImport 

14 

15with LazyImport("Run 'pip install python-pptx'") as pptx_import: 

16 from pptx import Presentation 

17 from pptx.text.text import _Paragraph 

18 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23@component 

24class PPTXToDocument: 

25 """ 

26 Converts PPTX files to Documents. 

27 

28 Usage example: 

29 

30 ```python 

31 from haystack.components.converters.pptx import PPTXToDocument 

32 from datetime import datetime 

33 

34 converter = PPTXToDocument() 

35 results = converter.run( 

36 sources=["test/test_files/pptx/sample_pptx.pptx"], meta={"date_added": datetime.now().isoformat()} 

37 ) 

38 documents = results["documents"] 

39 

40 print(documents[0].content) 

41 # >> 'This is the text from the PPTX file.' 

42 ``` 

43 """ 

44 

45 def __init__( 

46 self, store_full_path: bool = False, link_format: Literal["markdown", "plain", "none"] = "none" 

47 ) -> None: 

48 """ 

49 Create a PPTXToDocument component. 

50 

51 :param store_full_path: 

52 If True, the full path of the file is stored in the metadata of the document. 

53 If False, only the file name is stored. 

54 :param link_format: The format for link output. Possible options: 

55 - `"markdown"`: `[text](url)` 

56 - `"plain"`: `text (url)` 

57 - `"none"`: Only the text is extracted, link addresses are ignored. 

58 """ 

59 pptx_import.check() 

60 if link_format not in ("markdown", "plain", "none"): 

61 msg = f"Unknown link format '{link_format}'. Supported formats are: 'markdown', 'plain', 'none'" 

62 raise ValueError(msg) 

63 self.link_format = link_format 

64 self.store_full_path = store_full_path 

65 

66 def to_dict(self) -> dict[str, Any]: 

67 """ 

68 Serializes the component to a dictionary. 

69 

70 :returns: 

71 Dictionary with serialized data. 

72 """ 

73 return default_to_dict(self, link_format=self.link_format, store_full_path=self.store_full_path) 

74 

75 def _convert(self, file_content: io.BytesIO) -> str: 

76 """ 

77 Converts the PPTX file to text. 

78 """ 

79 pptx_presentation = Presentation(file_content) 

80 text_all_slides = [] 

81 for slide in pptx_presentation.slides: 

82 text_on_slide = [] 

83 for shape in slide.shapes: 

84 if shape.has_text_frame: 

85 paragraphs = [] 

86 for paragraph in shape.text_frame.paragraphs: 

87 paragraphs.append(self._process_paragraph(paragraph)) 

88 text_on_slide.append("\n".join(paragraphs)) 

89 elif hasattr(shape, "text"): 

90 text_on_slide.append(shape.text) 

91 text_all_slides.append("\n".join(text_on_slide)) 

92 return "\f".join(text_all_slides) 

93 

94 def _process_paragraph(self, paragraph: "_Paragraph") -> str: 

95 """ 

96 Processes a paragraph and formats hyperlinks according to the specified link format. 

97 

98 :param paragraph: The PPTX paragraph to process. 

99 :returns: A string with links formatted according to the specified format. 

100 """ 

101 if self.link_format == "none": 

102 return paragraph.text 

103 parts = [] 

104 for run in paragraph.runs: 

105 if run.hyperlink and run.hyperlink.address: 

106 if self.link_format == "markdown": 

107 parts.append(f"[{run.text}]({run.hyperlink.address})") 

108 else: 

109 parts.append(f"{run.text} ({run.hyperlink.address})") 

110 else: 

111 parts.append(run.text) 

112 return "".join(parts) 

113 

114 @component.output_types(documents=list[Document]) 

115 def run( 

116 self, sources: list[str | Path | ByteStream], meta: dict[str, Any] | list[dict[str, Any]] | None = None 

117 ) -> dict[str, Any]: 

118 """ 

119 Converts PPTX files to Documents. 

120 

121 :param sources: 

122 List of file paths or ByteStream objects. 

123 :param meta: 

124 Optional metadata to attach to the Documents. 

125 This value can be either a list of dictionaries or a single dictionary. 

126 If it's a single dictionary, its content is added to the metadata of all produced Documents. 

127 If it's a list, the length of the list must match the number of sources, because the two lists will 

128 be zipped. 

129 If `sources` contains ByteStream objects, their `meta` will be added to the output Documents. 

130 

131 :returns: 

132 A dictionary with the following keys: 

133 - `documents`: Created Documents 

134 """ 

135 documents = [] 

136 meta_list = normalize_metadata(meta, sources_count=len(sources)) 

137 

138 for source, metadata in zip(sources, meta_list, strict=True): 

139 try: 

140 bytestream = get_bytestream_from_source(source) 

141 except Exception as e: 

142 logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=e) 

143 continue 

144 try: 

145 text = self._convert(io.BytesIO(bytestream.data)) 

146 except Exception as e: 

147 logger.warning( 

148 "Could not read {source} and convert it to Document, skipping. {error}", source=source, error=e 

149 ) 

150 continue 

151 

152 merged_metadata = {**bytestream.meta, **metadata} 

153 

154 if not self.store_full_path and (file_path := bytestream.meta.get("file_path")): 

155 merged_metadata["file_path"] = os.path.basename(file_path) 

156 documents.append(Document(content=text, meta=merged_metadata)) 

157 

158 return {"documents": documents}