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, 

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

41 store_full_path: bool = False, 

42 encoding: str | None = None, 

43 ) -> None: 

44 """ 

45 Create an HTMLToDocument component. 

46 

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

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

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

50 :param store_full_path: 

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

52 If False, only the file name is stored. 

53 :param encoding: 

54 The encoding to use when decoding HTML files. If None (the default), Trafilatura detects the encoding. 

55 If detection produces incorrect characters, install `trafilatura[all]` for additional encoding detection 

56 support, or set this parameter if you know the encoding. 

57 """ 

58 trafilatura_import.check() 

59 

60 self.extraction_kwargs = extraction_kwargs or {} 

61 self.store_full_path = store_full_path 

62 self.encoding = encoding 

63 

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

65 """ 

66 Serializes the component to a dictionary. 

67 

68 :returns: 

69 Dictionary with serialized data. 

70 """ 

71 return default_to_dict( 

72 self, extraction_kwargs=self.extraction_kwargs, store_full_path=self.store_full_path, encoding=self.encoding 

73 ) 

74 

75 @classmethod 

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

77 """ 

78 Deserializes the component from a dictionary. 

79 

80 :param data: 

81 The dictionary to deserialize from. 

82 :returns: 

83 The deserialized component. 

84 """ 

85 return default_from_dict(cls, data) 

86 

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

88 def run( 

89 self, 

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

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

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

93 ) -> dict[str, Any]: 

94 """ 

95 Converts a list of HTML files to Documents. 

96 

97 :param sources: 

98 List of HTML file paths or ByteStream objects. 

99 :param meta: 

100 Optional metadata to attach to the Documents. 

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

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

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

104 be zipped. 

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

106 :param extraction_kwargs: 

107 Additional keyword arguments to customize the extraction process. 

108 

109 :returns: 

110 A dictionary with the following keys: 

111 - `documents`: Created Documents 

112 """ 

113 

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

115 

116 documents = [] 

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

118 

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

120 try: 

121 bytestream = get_bytestream_from_source(source=source) 

122 except Exception as e: 

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

124 continue 

125 

126 if not bytestream.data: 

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

128 continue 

129 

130 try: 

131 html_bytes = bytestream.data 

132 html_content = html_bytes.decode(self.encoding) if self.encoding is not None else html_bytes 

133 text = extract(html_content, **merged_extraction_kwargs) 

134 except Exception as conversion_e: 

135 logger.warning( 

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

137 source=source, 

138 error=conversion_e, 

139 ) 

140 continue 

141 

142 if text is None: 

143 logger.warning("No text could be extracted from {source}. Skipping it.", source=source) 

144 continue 

145 

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

147 

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

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

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

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

152 

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

154 documents.append(document) 

155 

156 return {"documents": documents}