Coverage for haystack/components/converters/msg.py: 94%

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 

9 

10from haystack import Document, component, 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-oxmsg'") as oxmsg_import: 

16 from oxmsg import Message, recipient 

17 

18 

19logger = logging.getLogger(__name__) 

20 

21 

22@component 

23class MSGToDocument: 

24 """ 

25 Converts Microsoft Outlook .msg files into Haystack Documents. 

26 

27 This component extracts email metadata (such as sender, recipients, CC, BCC, subject) and body content from .msg 

28 files and converts them into structured Haystack Documents. Additionally, any file attachments within the .msg 

29 file are extracted as ByteStream objects. 

30 

31 ### Example Usage 

32 

33 ```python 

34 from haystack.components.converters.msg import MSGToDocument 

35 from datetime import datetime 

36 

37 converter = MSGToDocument() 

38 results = converter.run(sources=["test/test_files/msg/sample.msg"], meta={"date_added": datetime.now().isoformat()}) 

39 documents = results["documents"] 

40 attachments = results["attachments"] 

41 print(documents[0].content) 

42 ``` 

43 """ 

44 

45 def __init__(self, store_full_path: bool = False) -> None: 

46 """ 

47 Creates a MSGToDocument component. 

48 

49 :param store_full_path: 

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

51 If False, only the file name is stored. 

52 """ 

53 oxmsg_import.check() 

54 self.store_full_path = store_full_path 

55 

56 @staticmethod 

57 def _is_encrypted(msg: "Message") -> bool: 

58 """ 

59 Determines whether the provided MSG file is encrypted. 

60 

61 :param msg: The MSG file as a parsed Message object. 

62 :returns: True if the MSG file is encrypted, otherwise False. 

63 """ 

64 return "encrypted" in msg.message_headers.get("Content-Type", "") 

65 

66 @staticmethod 

67 def _create_recipient_str(recip: "recipient.Recipient") -> str: 

68 """ 

69 Formats a recipient's name and email into a single string. 

70 

71 :param recip: A recipient object extracted from the MSG file. 

72 :returns: A formatted string combining the recipient's name and email address. 

73 """ 

74 recip_str = "" 

75 if recip.name != "": 

76 recip_str += f"{recip.name} " 

77 if recip.email_address != "": 

78 recip_str += f"{recip.email_address}" 

79 return recip_str 

80 

81 def _convert(self, file_content: io.BytesIO) -> tuple[str, list[ByteStream]]: 

82 """ 

83 Converts the MSG file content into text and extracts any attachments. 

84 

85 :param file_content: The MSG file content as a binary stream. 

86 :returns: A tuple containing the extracted email text and a list of ByteStream objects for attachments. 

87 :raises ValueError: If the MSG file is encrypted and cannot be read. 

88 """ 

89 msg = Message.load(file_content) 

90 if self._is_encrypted(msg): 

91 raise ValueError("The MSG file is encrypted and cannot be read.") 

92 

93 txt = "" 

94 

95 # Sender 

96 if msg.sender is not None: 

97 txt += f"From: {msg.sender}\n" 

98 

99 # To 

100 recipients_str = ",".join(self._create_recipient_str(r) for r in msg.recipients) 

101 if recipients_str != "": 

102 txt += f"To: {recipients_str}\n" 

103 

104 # CC 

105 cc_header = msg.message_headers.get("Cc") or msg.message_headers.get("CC") 

106 if cc_header is not None: 

107 txt += f"Cc: {cc_header}\n" 

108 

109 # BCC 

110 bcc_header = msg.message_headers.get("Bcc") or msg.message_headers.get("BCC") 

111 if bcc_header is not None: 

112 txt += f"Bcc: {bcc_header}\n" 

113 

114 # Subject 

115 if msg.subject != "": 

116 txt += f"Subject: {msg.subject}\n" 

117 

118 # Body 

119 if msg.body is not None: 

120 txt += "\n" + msg.body 

121 

122 # attachments 

123 attachments = [ 

124 ByteStream( 

125 data=attachment.file_bytes, meta={"file_path": attachment.file_name}, mime_type=attachment.mime_type 

126 ) 

127 for attachment in msg.attachments 

128 if attachment.file_bytes is not None 

129 ] 

130 

131 return txt, attachments 

132 

133 @component.output_types(documents=list[Document], attachments=list[ByteStream]) 

134 def run( 

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

136 ) -> dict[str, list[Document] | list[ByteStream]]: 

137 """ 

138 Converts MSG files to Documents. 

139 

140 :param sources: 

141 List of file paths or ByteStream objects. 

142 :param meta: 

143 Optional metadata to attach to the Documents. 

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

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

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

147 be zipped. 

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

149 

150 :returns: 

151 A dictionary with the following keys: 

152 - `documents`: Created Documents. 

153 - `attachments`: Created ByteStream objects from file attachments. 

154 """ 

155 if len(sources) == 0: 

156 return {"documents": [], "attachments": []} 

157 

158 documents = [] 

159 all_attachments = [] 

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

161 

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

163 try: 

164 bytestream = get_bytestream_from_source(source) 

165 except Exception as e: 

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

167 continue 

168 try: 

169 text, attachments = self._convert(io.BytesIO(bytestream.data)) 

170 except Exception as e: 

171 logger.warning( 

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

173 ) 

174 continue 

175 

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

177 

178 if not self.store_full_path and "file_path" in bytestream.meta: 

179 merged_metadata["file_path"] = os.path.basename(bytestream.meta["file_path"]) 

180 

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

182 for attachment in attachments: 

183 attachment_meta = {**merged_metadata, "file_path": attachment.meta["file_path"]} 

184 if "file_path" in merged_metadata: 

185 attachment_meta["parent_file_path"] = merged_metadata["file_path"] 

186 all_attachments.append( 

187 ByteStream(data=attachment.data, meta=attachment_meta, mime_type=attachment.mime_type) 

188 ) 

189 

190 return {"documents": documents, "attachments": all_attachments}