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

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, default_from_dict, default_to_dict, logging 

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

11from haystack.dataclasses import ByteStream 

12from haystack.lazy_imports import LazyImport 

13 

14logger = logging.getLogger(__name__) 

15 

16with LazyImport("Run 'pip install trafilatura'") as trafilatura_import: 

17 from trafilatura import extract 

18 

19 

20@component 

21class HTMLToDocument: 

22 """ 

23 Converts an HTML file to a Document. 

24 

25 Usage example: 

26 ```python 

27 from haystack.components.converters import HTMLToDocument 

28 

29 converter = HTMLToDocument() 

30 results = converter.run(sources=["test/test_files/html/paul_graham_superlinear.html"]) 

31 documents = results["documents"] 

32 

33 print(documents[0].content) 

34 # >> 'This is a text from the HTML file.' 

35 ``` 

36 """ 

37 

38 def __init__( 

39 self, extraction_kwargs: dict[str, Any] | None = None, store_full_path: bool = False, encoding: str = "utf-8" 

40 ) -> None: 

41 """ 

42 Create an HTMLToDocument component. 

43 

44 :param extraction_kwargs: A dictionary containing keyword arguments to customize the extraction process. These 

45 are passed to the underlying Trafilatura `extract` function. For the full list of available arguments, see 

46 the [Trafilatura documentation](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#extract). 

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 :param encoding: 

51 The default encoding to use when converting HTML files. If the encoding is specified in the metadata of a 

52 source ByteStream, it overrides this value. 

53 """ 

54 trafilatura_import.check() 

55 

56 self.extraction_kwargs = extraction_kwargs or {} 

57 self.store_full_path = store_full_path 

58 self.encoding = encoding 

59 

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

61 """ 

62 Serializes the component to a dictionary. 

63 

64 :returns: 

65 Dictionary with serialized data. 

66 """ 

67 return default_to_dict( 

68 self, extraction_kwargs=self.extraction_kwargs, store_full_path=self.store_full_path, encoding=self.encoding 

69 ) 

70 

71 @classmethod 

72 def from_dict(cls, data: dict[str, Any]) -> "HTMLToDocument": 

73 """ 

74 Deserializes the component from a dictionary. 

75 

76 :param data: 

77 The dictionary to deserialize from. 

78 :returns: 

79 The deserialized component. 

80 """ 

81 return default_from_dict(cls, data) 

82 

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

84 def run( 

85 self, 

86 sources: list[str | Path | ByteStream], 

87 meta: dict[str, Any] | list[dict[str, Any]] | None = None, 

88 extraction_kwargs: dict[str, Any] | None = None, 

89 ) -> dict[str, Any]: 

90 """ 

91 Converts a list of HTML files to Documents. 

92 

93 :param sources: 

94 List of HTML file paths or ByteStream objects. 

95 :param meta: 

96 Optional metadata to attach to the Documents. 

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

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

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

100 be zipped. 

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

102 :param extraction_kwargs: 

103 Additional keyword arguments to customize the extraction process. 

104 

105 :returns: 

106 A dictionary with the following keys: 

107 - `documents`: Created Documents 

108 """ 

109 

110 merged_extraction_kwargs = {**self.extraction_kwargs, **(extraction_kwargs or {})} 

111 

112 documents = [] 

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

114 

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

116 try: 

117 bytestream = get_bytestream_from_source(source=source) 

118 except Exception as e: 

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

120 continue 

121 

122 if not bytestream.data: 

123 logger.warning("Skipping {source} because it is empty.", source=source) 

124 continue 

125 

126 try: 

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

128 text = extract(bytestream.data.decode(encoding), **merged_extraction_kwargs) 

129 except Exception as conversion_e: 

130 logger.warning( 

131 "Failed to extract text from {source}. Skipping it. Error: {error}", 

132 source=source, 

133 error=conversion_e, 

134 ) 

135 continue 

136 

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

138 

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

140 file_path = bytestream.meta.get("file_path") 

141 if file_path: # Ensure the value is not None for mypy 

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

143 

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

145 documents.append(document) 

146 

147 return {"documents": documents}