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 (`utf-8-sig`, which also strips a byte order mark if 

22 present) when converting files but 

23 you can also set custom encoding. 

24 It can attach metadata to the resulting documents. 

25 

26 ### Usage example 

27 

28 ```python 

29 from haystack.components.converters.txt import TextFileToDocument 

30 

31 converter = TextFileToDocument() 

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

33 documents = results["documents"] 

34 

35 print(documents[0].content) 

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

37 ``` 

38 """ 

39 

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

41 """ 

42 Creates a TextFileToDocument component. 

43 

44 :param encoding: 

45 The encoding of the text files to convert. 

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

47 it overrides this value. 

48 :param store_full_path: 

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

50 If False, only the file name is stored. 

51 """ 

52 self.encoding = encoding 

53 self.store_full_path = store_full_path 

54 

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

56 def run( 

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

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

59 """ 

60 Converts text files to documents. 

61 

62 :param sources: 

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

64 :param meta: 

65 Optional metadata to attach to the documents. 

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

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

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

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

70 

71 :returns: 

72 A dictionary with the following keys: 

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

74 """ 

75 documents = [] 

76 

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

78 

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

80 try: 

81 bytestream = get_bytestream_from_source(source) 

82 except Exception as e: 

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

84 continue 

85 try: 

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

87 text = bytestream.data.decode(encoding) 

88 except Exception as e: 

89 logger.warning( 

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

91 ) 

92 continue 

93 

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

95 

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

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

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

99 documents.append(document) 

100 

101 return {"documents": documents}