Coverage for src/local_deep_research/research_library/downloaders/html.py: 94%

1""" 

2HTML Content Downloader for web pages. 

3 

4Downloads and extracts clean text content from HTML web pages. 

5Extraction is handled by the shared pipeline in extraction/pipeline.py. 

6""" 

7 

8from typing import Optional, Dict, Any 

9from urllib.parse import urlparse 

10from loguru import logger 

11from bs4 import BeautifulSoup 

12 

13from .base import BaseDownloader, ContentType, DownloadResult 

14from .extraction.pipeline import extract_content_with_metadata 

15from ...constants import BROWSER_USER_AGENT 

16from ...security import sanitize_error_for_client 

17 

18 

19class HTMLDownloader(BaseDownloader): 

20 """Downloader for HTML web pages - extracts clean text content.""" 

21 

22 def __init__( 

23 self, 

24 timeout: int = 30, 

25 language: str = "English", 

26 **kwargs, 

27 ): 

28 super().__init__(timeout) 

29 self.session.headers.update({"User-Agent": BROWSER_USER_AGENT}) 

30 self.language = language 

31 

32 def can_handle(self, url: str) -> bool: 

33 """ 

34 Check if this downloader can handle the given URL. 

35 

36 Returns True for any HTTP/HTTPS URL (fallback downloader for web content). 

37 """ 

38 try: 

39 parsed = urlparse(url) 

40 return parsed.scheme in ("http", "https") 

41 except Exception: 

42 return False 

43 

44 def download( 

45 self, url: str, content_type: ContentType = ContentType.TEXT 

46 ) -> Optional[bytes]: 

47 """ 

48 Download and extract text content from HTML page. 

49 

50 Args: 

51 url: The URL to download 

52 content_type: Type of content (TEXT for HTML extraction) 

53 

54 Returns: 

55 Extracted text as UTF-8 bytes, or None if failed 

56 """ 

57 if content_type == ContentType.PDF: 

58 logger.warning(f"HTML downloader cannot download PDFs: {url}") 

59 return None 

60 

61 try: 

62 html_content = self._fetch_html(url) 

63 if not html_content: 

64 return None 

65 

66 extracted = self._extract_content(html_content, url) 

67 if extracted: 

68 text = self._format_extracted_content(extracted) 

69 return text.encode("utf-8") 

70 

71 return None 

72 

73 except Exception: 

74 logger.exception(f"Failed to download HTML from {url}") 

75 return None 

76 

77 def download_with_result( 

78 self, url: str, content_type: ContentType = ContentType.TEXT 

79 ) -> DownloadResult: 

80 """Download content and return detailed result with skip reason.""" 

81 if content_type == ContentType.PDF: 

82 return DownloadResult( 

83 skip_reason="HTML downloader does not support PDF downloads" 

84 ) 

85 

86 try: 

87 html_content = self._fetch_html(url) 

88 if not html_content: 

89 return DownloadResult( 

90 skip_reason="Failed to fetch HTML content from URL" 

91 ) 

92 

93 extracted = self._extract_content(html_content, url) 

94 if not extracted: 

95 return DownloadResult( 

96 skip_reason="Could not extract meaningful content from page" 

97 ) 

98 

99 text = self._format_extracted_content(extracted) 

100 if not text.strip(): 

101 return DownloadResult(skip_reason="Extracted content is empty") 

102 

103 return DownloadResult( 

104 content=text.encode("utf-8"), 

105 is_success=True, 

106 ) 

107 

108 except Exception as e: 

109 logger.exception(f"Failed to download HTML from {url}") 

110 # skip_reason propagates to the browser via the download SSE 

111 # stream; the fetch URL can carry credentials — scrub before 

112 # returning (full detail stays in the server log above). 

113 return DownloadResult( 

114 skip_reason=sanitize_error_for_client(f"Error: {str(e)}") 

115 ) 

116 

117 def _fetch_html(self, url: str) -> Optional[str]: 

118 """Fetch raw HTML content from URL.""" 

119 logger.debug(f"Static fetch: {url}") 

120 domain = urlparse(url).netloc 

121 engine_type = f"html_download_{domain}" 

122 

123 wait_time = self.rate_tracker.apply_rate_limit(engine_type) 

124 

125 try: 

126 response = self.session.get( 

127 url, 

128 timeout=self.timeout, 

129 allow_redirects=True, 

130 ) 

131 

132 if response.status_code == 200: 

133 content_type = response.headers.get("content-type", "").lower() 

134 if ( 

135 "text/html" in content_type 

136 or "application/xhtml" in content_type 

137 ): 

138 self.rate_tracker.record_outcome( 

139 engine_type=engine_type, 

140 wait_time=wait_time, 

141 success=True, 

142 retry_count=1, 

143 search_result_count=1, 

144 ) 

145 return response.text 

146 logger.warning( 

147 f"Unexpected content type for HTML download: {content_type}" 

148 ) 

149 return None 

150 logger.warning(f"HTTP {response.status_code} fetching {url}") 

151 self.rate_tracker.record_outcome( 

152 engine_type=engine_type, 

153 wait_time=wait_time, 

154 success=False, 

155 retry_count=1, 

156 error_type=f"HTTP_{response.status_code}", 

157 ) 

158 return None 

159 

160 except Exception as e: 

161 logger.exception(f"Error fetching HTML from {url}") 

162 self.rate_tracker.record_outcome( 

163 engine_type=engine_type, 

164 wait_time=wait_time, 

165 success=False, 

166 retry_count=1, 

167 error_type=type(e).__name__, 

168 ) 

169 return None 

170 

171 def _extract_content(self, html: str, url: str) -> Optional[Dict[str, Any]]: 

172 """Extract clean content and metadata from HTML. 

173 

174 Delegates to the shared extraction pipeline which handles 

175 trafilatura, readability, justext, and metadata enrichment. 

176 """ 

177 try: 

178 result = extract_content_with_metadata(html, language=self.language) 

179 if not result: 

180 return None 

181 

182 title = result.get("title") 

183 content = result["content"] 

184 

185 logger.info( 

186 f"Extracted {len(content)} chars from {url} " 

187 f"(title: {title[:50] + '...' if title and len(title) > 50 else title})" 

188 ) 

189 return { 

190 "title": title, 

191 "description": result.get("description"), 

192 "content": content, 

193 "url": url, 

194 } 

195 

196 except Exception: 

197 logger.exception("Error extracting content from HTML") 

198 return None 

199 

200 def _format_extracted_content(self, extracted: Dict[str, Any]) -> str: 

201 """Format extracted content as readable text.""" 

202 parts = [] 

203 

204 if extracted.get("title"): 

205 parts.append(f"# {extracted['title']}") 

206 parts.append("") 

207 

208 if extracted.get("description"): 

209 parts.append(f"*{extracted['description']}*") 

210 parts.append("") 

211 

212 if extracted.get("url"): 

213 parts.append(f"Source: {extracted['url']}") 

214 parts.append("") 

215 

216 if extracted.get("content"): 

217 parts.append(extracted["content"]) 

218 

219 return "\n".join(parts) 

220 

221 def get_metadata(self, url: str) -> Dict[str, Any]: 

222 """Get metadata about the page.""" 

223 html_content = self._fetch_html(url) 

224 if not html_content: 

225 return {} 

226 

227 try: 

228 soup = BeautifulSoup(html_content, "html.parser") 

229 

230 metadata = {"url": url} 

231 

232 if soup.title and soup.title.string: 232 ↛ 235line 232 didn't jump to line 235 because the condition on line 232 was always true

233 metadata["title"] = soup.title.string.strip() 

234 

235 meta_desc = soup.find("meta", attrs={"name": "description"}) 

236 if meta_desc and meta_desc.get("content"): 

237 metadata["description"] = str(meta_desc["content"]) 

238 

239 author = soup.find("meta", attrs={"name": "author"}) 

240 if author and author.get("content"): 

241 metadata["author"] = str(author["content"]) 

242 

243 for prop in ["article:published_time", "datePublished"]: 

244 date_tag = soup.find("meta", property=prop) 

245 if date_tag and date_tag.get("content"): 

246 metadata["published_date"] = str(date_tag["content"]) 

247 break 

248 

249 return metadata 

250 

251 except Exception: 

252 logger.exception(f"Error extracting metadata from {url}") 

253 return {"url": url}