Coverage for src/local_deep_research/web_search_engines/engines/search_engine

1"""Webz.io direct News Search API (not the MCP service).

3The provider's examples use HTTP. This adapter only uses HTTPS, refuses

4redirects and never falls back to HTTP; live TLS compatibility is unverified.

5The client caps k at 50 because the Parameters and Errors pages disagree.

6"""

8import copy

9import math

10from datetime import date

11from typing import Any, Dict, List, Optional

13import requests

14from langchain_core.language_models import BaseLLM

16from ...security.safe_requests import safe_post

17from ...security.secure_logging import logger

18from ..rate_limiting import RateLimitError

19from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity

22class WebzSearchEngine(BaseSearchEngine):

23 """Semantic news search with article excerpts and source metadata."""

25 is_public = True

26 is_news = True

27 egress_sensitivity = Sensitivity.NON_SENSITIVE

28 egress_exposure = Exposure.EXPOSING

30 SEARCH_URL = "https://api.webz.io/api/news/context"

31 MAX_RESULTS_CAP = 50

32 REQUEST_TIMEOUT = 30

33 # Client-side bounds for documented array filters, not provider limits.

34 MAX_FILTER_VALUES = 50

35 MAX_FILTER_VALUE_LENGTH = 256

36 ARRAY_FILTERS = frozenset(

37 {

38 "language",

39 "country",

40 "category",

41 "sentiment",

42 "domain",

43 "exclude_domain",

44 "topic",

45 "person",

46 "organization",

47 "location",

48 "ticker",

49 "political_bias",

50 }

51 )

52 DATE_FILTERS = frozenset({"published_from", "published_to"})

53 RANK_FILTERS = frozenset({"domain_rank_gte", "domain_rank_lte"})

54 CHOICE_FILTERS = {

55 "source_type": {"local_news", "newsroom", "gov_news"},

56 "trust_category": {"trusted_news", "fake_news", "satirical_news"},

57 }

59 def __init__(

60 self,

61 max_results: int = 10,

62 api_key: Optional[str] = None,

63 filters: Optional[Dict[str, Any]] = None,

64 score_gte: float = 4,

65 score_lte: Optional[float] = None,

66 allow_multiple_chunks_per_article: bool = False,

67 llm: Optional[BaseLLM] = None,

68 max_filtered_results: Optional[int] = None,

69 settings_snapshot: Optional[Dict[str, Any]] = None,

70 **kwargs,

71 ):

72 """Accept only documented direct-API fields and normal base kwargs.

74 ``filters`` uses absolute ISO calendar dates, with no implicit recency

75 conversion. The provider determines coverage when dates are omitted.

76 Array filters are bounded to 50 nonempty strings of 256 characters.

77 Credentials are resolved by the base class from parameters/snapshot.

78 """

79 if type(max_results) is not int:

80 raise ValueError("Webz max_results must be an integer")

81 max_results = max(1, min(max_results, self.MAX_RESULTS_CAP))

82 for score in (score_gte, score_lte):

83 if score is not None and (

84 type(score) not in (int, float)

85 or not math.isfinite(score)

86 or not 0 <= score <= 10

87 ):

88 raise ValueError(

89 "Webz scores must be finite numbers from 0 to 10"

90 )

91 if score_gte is None or (

92 score_lte is not None and score_gte > score_lte

93 ):

94 raise ValueError("Webz score_gte must not exceed score_lte")

95 if type(allow_multiple_chunks_per_article) is not bool:

96 raise ValueError(

97 "Webz allow_multiple_chunks_per_article must be boolean"

98 )

99 self.filters = self._validate_filters(filters)

100 self.score_gte = score_gte

101 self.score_lte = score_lte

102 self.allow_multiple_chunks_per_article = (

103 allow_multiple_chunks_per_article

104 )

105 if api_key is not None and not isinstance(api_key, str):

106 raise ValueError("Webz API key must be a string")

107 super().__init__(

108 llm=llm,

109 max_filtered_results=max_filtered_results,

110 max_results=max_results,

111 settings_snapshot=settings_snapshot,

112 **kwargs,

113 )

114 try:

115 self.api_key = self._resolve_api_key(

116 api_key,

117 "search.engine.web.webz.api_key",

118 engine_name="Webz.io",

119 settings_snapshot=settings_snapshot,

120 )

121 except (AttributeError, TypeError):

122 raise ValueError("Webz API key must be a string") from None

123 if any(ord(char) < 32 or ord(char) == 127 for char in self.api_key):

124 raise ValueError("Webz API key contains invalid header characters")

125

126 @classmethod

127 def _validate_filters(

128 cls, filters: Optional[Dict[str, Any]]

129 ) -> Dict[str, Any]:

130 if filters is None:

131 return {}

132 if not isinstance(filters, dict):

133 raise ValueError("Webz filters must be an object")

134 allowed = (

135 cls.ARRAY_FILTERS

136 | cls.DATE_FILTERS

137 | cls.RANK_FILTERS

138 | cls.CHOICE_FILTERS.keys()

139 )

140 if filters.keys() - allowed:

141 raise ValueError("Unsupported Webz filter field")

142 for key, value in filters.items():

143 if key in cls.ARRAY_FILTERS:

144 if (

145 not isinstance(value, list)

146 or not 1 <= len(value) <= cls.MAX_FILTER_VALUES

147 ):

148 raise ValueError("Webz array filters need 1-50 values")

149 if any(

150 not isinstance(item, str)

151 or not item.strip()

152 or len(item) > cls.MAX_FILTER_VALUE_LENGTH

153 for item in value

154 ):

155 raise ValueError(

156 "Webz array filter values need nonempty strings of at most 256 characters"

157 )

158 elif key in cls.DATE_FILTERS:

159 if not isinstance(value, str): 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 raise ValueError(

161 "Webz date filters need YYYY-MM-DD strings"

162 )

163 try:

164 parsed = date.fromisoformat(value)

165 except ValueError:

166 raise ValueError(

167 "Webz date filters need valid YYYY-MM-DD dates"

168 ) from None

169 if parsed.isoformat() != value: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true

170 raise ValueError("Webz date filters need YYYY-MM-DD dates")

171 elif key in cls.RANK_FILTERS:

172 if type(value) is not int or not 1 <= value <= 1000000:

173 raise ValueError(

174 "Webz domain rank filters need integers from 1 to 1000000"

175 )

176 elif (

177 not isinstance(value, str)

178 or value not in cls.CHOICE_FILTERS[key]

179 ):

180 raise ValueError(

181 "Invalid Webz source_type or trust_category filter"

182 )

183 for lower, upper in (

184 ("published_from", "published_to"),

185 ("domain_rank_gte", "domain_rank_lte"),

186 ):

187 if (

188 lower in filters

189 and upper in filters

190 and filters[lower] > filters[upper]

191 ):

192 raise ValueError(

193 "Webz filter lower bound must not exceed upper bound"

194 )

195 included = {item.lower() for item in filters.get("domain", [])}

196 excluded = {item.lower() for item in filters.get("exclude_domain", [])}

197 if included & excluded:

198 raise ValueError(

199 "Webz domain filters cannot include and exclude the same domain"

200 )

201 return copy.deepcopy(filters)

202

203 @staticmethod

204 def _check_response(response: requests.Response) -> None:

205 response.raise_for_status()

206 if response.status_code != 200:

207 raise requests.HTTPError(

208 f"Unexpected Webz HTTP status {response.status_code}",

209 response=response,

210 )

211

212 def _get_previews(self, query: str) -> List[Dict[str, Any]]:

213 if not isinstance(query, str):

214 raise ValueError("Webz query must be a string")

215 if not query.strip():

216 return []

217 if len(query) > 750 or len(query.split()) > 100:

218 raise ValueError("Webz query exceeds 750 characters or 100 words")

219 body: Dict[str, Any] = {

220 "query": query,

221 "k": self.max_results,

222 "score_gte": self.score_gte,

223 "allow_multiple_chunks_per_article": self.allow_multiple_chunks_per_article,

224 }

225 if self.score_lte is not None:

226 body["score_lte"] = self.score_lte

227 if self.filters:

228 body["filters"] = copy.deepcopy(self.filters)

229 try:

230 self._last_wait_time = self.rate_tracker.apply_rate_limit(

231 self.engine_type

232 )

233 response = safe_post(

234 self.SEARCH_URL,

235 json=body,

236 headers={

237 "Authorization": f"Bearer {self.api_key}",

238 "Accept": "application/json",

239 },

240 timeout=self.REQUEST_TIMEOUT,

241 allow_redirects=False,

242 )

243 self._check_response(response)

244 data = response.json()

245 rows = data.get("results") if isinstance(data, dict) else None

246 if not isinstance(rows, list):

247 return []

248 previews: List[Dict[str, Any]] = []

249 for row in rows:

250 if not isinstance(row, dict) or not isinstance(

251 row.get("article"), dict

252 ):

253 continue

254 article = row["article"]

255 link = article.get("url")

256 if not isinstance(link, str) or not link.strip():

257 continue

258 link = self._clean_result_url(link)

259 if not link.lower().startswith(("https://", "http://")):

260 continue

261 if not self._extract_display_link(link):

262 continue

263 chunk = row.get("chunk")

264 text = chunk.get("text") if isinstance(chunk, dict) else None

265 summary = article.get("summary")

266 snippet = (

267 text if isinstance(text, str) and text.strip() else summary

268 )

269 metadata = row.get("metadata")

270 metadata = metadata if isinstance(metadata, dict) else {}

271 domain = metadata.get("domain")

272 preview = {

273 "id": len(previews),

274 "title": article.get("title")

275 if isinstance(article.get("title"), str)

276 else "",

277 "link": link,

278 "snippet": snippet if isinstance(snippet, str) else "",

279 "displayed_link": self._extract_display_link(link),

280 "source": domain

281 if isinstance(domain, str)

282 else self._extract_display_link(link),

283 "metadata": copy.deepcopy(metadata),

284 "position": len(previews) + 1,

285 }

286 for field, target in (

287 ("published_at", "date"),

288 ("article_id", "article_id"),

289 ("main_image", "main_image"),

290 ):

291 if isinstance(article.get(field), str) and article[field]:

292 preview[target] = article[field]

293 score = row.get("score")

294 if (

295 isinstance(score, (int, float))

296 and not isinstance(score, bool)

297 and math.isfinite(score)

298 ):

299 preview["score"] = score

300 previews.append(preview)

301 if len(previews) == self.max_results:

302 break

303 self._search_results = previews

304 return previews

305 except RateLimitError:

306 raise

307 except requests.RequestException as error:

308 self._record_search_failure(error)

309 safe_msg = self._scrub_error(error)

310 # Preserve Retry-After while passing only scrubbed text to the

311 # base rate-limit detector (which can include error text).

312 self._raise_if_rate_limit(

313 requests.RequestException(

314 safe_msg, response=getattr(error, "response", None)

315 )

316 )

317 logger.warning(f"Error getting Webz.io news results: {safe_msg}")

318 return []

319 except Exception as error:

320 self._record_search_failure(error)

321 safe_msg = self._scrub_error(error)

322 logger.warning(

323 f"Unexpected error getting Webz.io news results: {safe_msg}"

324 )

325 return []

Coverage for src/local_deep_research/web_search_engines/engines/search_engine_webz.py: 98%

155 statements