Coverage for haystack/components/converters/file_to_file_content.py: 100%

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

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import base64 

6from pathlib import Path 

7from typing import Any 

8 

9from haystack import component, logging 

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

11from haystack.dataclasses import ByteStream, FileContent 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16_EMPTY_BYTE_STRING = b"" 

17 

18 

19@component 

20class FileToFileContent: 

21 """ 

22 Converts files to FileContent objects to be included in ChatMessage objects. 

23 

24 ### Usage example 

25 <!-- test-ignore --> 

26 ```python 

27 from haystack.components.converters import FileToFileContent 

28 

29 converter = FileToFileContent() 

30 sources = ["test/test_files/pdf/react_paper.pdf", "test/test_files/images/haystack-logo.png"] 

31 file_contents = converter.run(sources=sources)["file_contents"] 

32 

33 print(file_contents) 

34 # >> [FileContent(base64_data='...', mime_type='application/pdf', filename='react_paper.pdf', extra={}), 

35 # >> FileContent(base64_data='...', mime_type='image/png', filename='haystack-logo.png', extra={}) 

36 # >>] 

37 ``` 

38 """ 

39 

40 @component.output_types(file_contents=list[FileContent]) 

41 def run( 

42 self, sources: list[str | Path | ByteStream], *, extra: dict[str, Any] | list[dict[str, Any]] | None = None 

43 ) -> dict[str, list[FileContent]]: 

44 """ 

45 Converts files to FileContent objects. 

46 

47 :param sources: 

48 List of file paths or ByteStream objects to convert. 

49 :param extra: 

50 Optional extra information to attach to the FileContent objects. Can be used to store provider-specific 

51 information. 

52 To avoid serialization issues, values should be JSON serializable. 

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

54 If it's a single dictionary, its content is added to the extra of all produced FileContent objects. 

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

56 

57 :returns: 

58 A dictionary with the following keys: 

59 - `file_contents`: A list of FileContent objects. 

60 """ 

61 if not sources: 

62 return {"file_contents": []} 

63 

64 file_contents = [] 

65 

66 extra_list = normalize_metadata(extra, sources_count=len(sources)) 

67 

68 for source, extra_dict in zip(sources, extra_list, strict=True): 

69 if isinstance(source, str): 

70 source = Path(source) 

71 

72 filename = source.name if isinstance(source, Path) else None 

73 

74 try: 

75 bytestream = get_bytestream_from_source(source, guess_mime_type=True) 

76 except Exception as e: 

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

78 continue 

79 

80 if bytestream.data == _EMPTY_BYTE_STRING: 

81 logger.warning("File {source} is empty. Skipping it.", source=source) 

82 continue 

83 

84 base64_data = base64.b64encode(bytestream.data).decode("utf-8") 

85 file_content = FileContent( 

86 base64_data=base64_data, mime_type=bytestream.mime_type, filename=filename, extra=extra_dict 

87 ) 

88 file_contents.append(file_content) 

89 

90 return {"file_contents": file_contents}