Coverage for haystack/components/converters/txt.py: 91%

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

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import os 

6from pathlib import Path 

7from typing import Any 

8 

9from haystack import Document, component, logging 

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

11from haystack.dataclasses import ByteStream 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16@component 

17class TextFileToDocument: 

18 """ 

19 Converts text files to documents your pipeline can query. 

20 

21 By default, it uses UTF-8 encoding when converting files but 

22 you can also set custom encoding. 

23 It can attach metadata to the resulting documents. 

24 

25 ### Usage example 

26 

27 ```python 

28 from haystack.components.converters.txt import TextFileToDocument 

29 

30 converter = TextFileToDocument() 

31 results = converter.run(sources=["test/test_files/txt/doc_1.txt"]) 

32 documents = results["documents"] 

33 

34 print(documents[0].content) 

35 # >> 'This is the content from the txt file.' 

36 ``` 

37 """ 

38 

39 def __init__(self, encoding: str = "utf-8", store_full_path: bool = False) -> None: 

40 """ 

41 Creates a TextFileToDocument component. 

42 

43 :param encoding: 

44 The encoding of the text files to convert. 

45 If the encoding is specified in the metadata of a source ByteStream, 

46 it overrides this value. 

47 :param store_full_path: 

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

49 If False, only the file name is stored. 

50 """ 

51 self.encoding = encoding 

52 self.store_full_path = store_full_path 

53 

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

55 def run( 

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

57 ) -> dict[str, list[Document]]: 

58 """ 

59 Converts text files to documents. 

60 

61 :param sources: 

62 List of text file paths or ByteStream objects to convert. 

63 :param meta: 

64 Optional metadata to attach to the documents. 

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

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

67 If it's a list, its length must match the number of sources as they're zipped together. 

68 For ByteStream objects, their `meta` is added to the output documents. 

69 

70 :returns: 

71 A dictionary with the following keys: 

72 - `documents`: A list of converted documents. 

73 """ 

74 documents = [] 

75 

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

77 

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

79 try: 

80 bytestream = get_bytestream_from_source(source) 

81 except Exception as e: 

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

83 continue 

84 try: 

85 encoding = bytestream.meta.get("encoding", self.encoding) 

86 text = bytestream.data.decode(encoding) 

87 except Exception as e: 

88 logger.warning( 

89 "Could not convert file {source}. Skipping it. Error message: {error}", source=source, error=e 

90 ) 

91 continue 

92 

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

94 

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

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

97 document = Document(content=text, meta=merged_metadata) 

98 documents.append(document) 

99 

100 return {"documents": documents}