Coverage for src/local_deep_research/web_search_engines/engines/search_engine_wikipedia.py: 100%
139 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import json
2from typing import Any, Dict, List, Optional
4import requests
5import wikipedia
6from langchain_core.language_models import BaseLLM
8from ..engine_availability import SearchFailure
9from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
10from ...security import sanitize_for_log
11from ...security.secure_logging import logger
14# Wikipedia / MediaWiki returns a non-JSON HTML body with HTTP 429 when a
15# client is rate-limited. The `wikipedia` PyPI library calls
16# ``Response.json()`` unconditionally, so the symptom we observe is a
17# ``JSONDecodeError`` on every subsequent ``wikipedia.summary()`` call.
18# Catch that family explicitly so we can short-circuit instead of
19# emitting a full traceback per title (the original behaviour spammed
20# the log with 14+ stack traces for a single rate-limited query).
21_TRANSIENT_DECODE_ERRORS: tuple = (
22 json.JSONDecodeError,
23 requests.exceptions.JSONDecodeError,
24)
26# The ``wikipedia`` library discards the response, so the 429 status is not
27# available here; classify the decode error as the throttle it almost always
28# is instead of letting it fall back to a generic one-minute failure.
29_THROTTLED = SearchFailure(
30 "Rate limit reached (Wikipedia returned a non-JSON response)", 300
31)
34class WikipediaSearchEngine(BaseSearchEngine):
35 """Wikipedia search engine implementation with two-phase approach"""
37 # Mark as public search engine
38 is_public = True
39 egress_sensitivity = Sensitivity.NON_SENSITIVE
40 egress_exposure = Exposure.EXPOSING
41 is_lexical = True
42 needs_llm_relevance_filter = True
44 def __init__(
45 self,
46 max_results: int = 10,
47 language: str = "en",
48 include_content: bool = True,
49 sentences: int = 5,
50 llm: Optional[BaseLLM] = None,
51 max_filtered_results: Optional[int] = None,
52 settings_snapshot: Optional[Dict[str, Any]] = None,
53 **kwargs,
54 ):
55 """
56 Initialize the Wikipedia search engine.
58 Args:
59 max_results: Maximum number of search results
60 language: Language code for Wikipedia (e.g., 'en', 'fr', 'es')
61 include_content: Whether to include full page content in results
62 sentences: Number of sentences to include in summary
63 llm: Language model for relevance filtering
64 max_filtered_results: Maximum number of results to keep after filtering
65 **kwargs: Additional parameters (ignored but accepted for compatibility)
66 """
67 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
68 super().__init__(
69 llm=llm,
70 max_filtered_results=max_filtered_results,
71 max_results=max_results,
72 settings_snapshot=settings_snapshot,
73 )
74 self.include_content = include_content
75 self.sentences = sentences
77 # Set the Wikipedia language
78 wikipedia.set_lang(language)
80 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
81 """
82 Get preview information (titles and summaries) for Wikipedia pages.
84 Args:
85 query: The search query
87 Returns:
88 List of preview dictionaries
89 """
90 logger.info("Getting Wikipedia page previews")
92 try:
93 # Apply rate limiting before search request
94 self._last_wait_time = self.rate_tracker.apply_rate_limit(
95 self.engine_type
96 )
98 # Get search results (just titles)
99 search_results = wikipedia.search(query, results=self.max_results)
100 safe_search_results = [
101 sanitize_for_log(title) for title in search_results
102 ]
104 logger.info(
105 f"Found {len(search_results)} Wikipedia results: "
106 f"{safe_search_results}"
107 )
109 if not search_results:
110 logger.info("No Wikipedia results found")
111 return []
113 # Generate previews with summaries.
114 # NOTE: This loop is intentionally sequential. Do NOT parallelize with
115 # ThreadPoolExecutor because:
116 # 1. The `wikipedia` PyPI library is not thread-safe — it uses global
117 # mutable state (API_URL, RATE_LIMIT_LAST_CALL) and an unlocked cache.
118 # Concurrent threads would corrupt the library's built-in rate limiting.
119 # 2. self._last_wait_time is a shared instance attribute with no lock —
120 # concurrent writes would feed incorrect data to record_outcome().
121 # 3. Downstream _filter_for_relevance uses positional indices — random
122 # completion order would cause the LLM to select wrong articles.
123 previews = []
124 for title in search_results:
125 safe_title = sanitize_for_log(title)
126 try:
127 # Get just the summary, with auto_suggest=False to be more precise
128 summary = None
129 try:
130 # Apply rate limiting before summary request
131 self._last_wait_time = (
132 self.rate_tracker.apply_rate_limit(self.engine_type)
133 )
135 summary = wikipedia.summary(
136 title, sentences=self.sentences, auto_suggest=False
137 )
138 except wikipedia.exceptions.DisambiguationError as e:
139 # If disambiguation error, try the first option
140 if e.options and len(e.options) > 0:
141 # Page titles are external content; sanitize
142 # before interpolation to prevent log injection.
143 first_option = e.options[0]
144 safe_first_option = sanitize_for_log(first_option)
145 logger.info(
146 f"Disambiguation for '{safe_title}', trying "
147 f"first option: {safe_first_option}"
148 )
149 try:
150 summary = wikipedia.summary(
151 first_option,
152 sentences=self.sentences,
153 auto_suggest=False,
154 )
155 title = first_option # Use the new title
156 except Exception as inner_e:
157 safe_msg = self._scrub_error(inner_e)
158 logger.exception(
159 f"Error with disambiguation option ({type(inner_e).__name__}): {safe_msg}"
160 )
161 continue
162 else:
163 logger.warning(
164 f"Disambiguation with no options for '{safe_title}'"
165 )
166 continue
168 if summary:
169 preview = {
170 "id": title, # Use title as ID
171 "title": title,
172 "snippet": summary,
173 "link": f"https://en.wikipedia.org/wiki/{title.replace(' ', '_')}",
174 "source": "Wikipedia",
175 }
177 previews.append(preview)
179 except (
180 wikipedia.exceptions.PageError,
181 wikipedia.exceptions.WikipediaException,
182 ):
183 # Skip pages with errors
184 logger.warning(f"Error getting summary for '{safe_title}'")
185 continue
186 except _TRANSIENT_DECODE_ERRORS:
187 self.last_search_failure = _THROTTLED
188 # MediaWiki almost certainly returned a 429 (or other
189 # non-JSON page) — every remaining title in this batch
190 # will hit the same throttle. Bail out with one warning
191 # instead of a per-title traceback storm.
192 logger.warning(
193 "Wikipedia rate-limited (non-JSON response while "
194 "fetching '{}'); returning {} previews collected so far",
195 safe_title,
196 len(previews),
197 )
198 break
199 except Exception as e:
200 self._record_search_failure(e)
201 safe_msg = self._scrub_error(e)
202 logger.exception(
203 f"Unexpected error for '{safe_title}' "
204 f"({type(e).__name__}): {safe_msg}"
205 )
206 continue
208 logger.info(
209 f"Successfully created {len(previews)} previews from Wikipedia"
210 )
211 return previews
213 except _TRANSIENT_DECODE_ERRORS:
214 self.last_search_failure = _THROTTLED
215 # Same 429-style failure on the outer wikipedia.search() call —
216 # log once at warning level and return an empty list.
217 logger.warning(
218 "Wikipedia rate-limited on search; returning no previews"
219 )
220 return []
221 except Exception as e:
222 self._record_search_failure(e)
223 safe_msg = self._scrub_error(e)
224 logger.exception(
225 f"Error getting Wikipedia previews ({type(e).__name__}): {safe_msg}"
226 )
227 return []
229 def _get_full_content(
230 self, relevant_items: List[Dict[str, Any]]
231 ) -> List[Dict[str, Any]]:
232 """
233 Get full content for the relevant Wikipedia pages.
235 Args:
236 relevant_items: List of relevant preview dictionaries
238 Returns:
239 List of result dictionaries with full content
240 """
241 logger.info(
242 f"Getting full content for {len(relevant_items)} relevant Wikipedia pages"
243 )
245 results = []
246 for item in relevant_items:
247 title = item.get("id") # Title stored as ID
249 if not title:
250 results.append(item)
251 continue
253 safe_title = sanitize_for_log(str(title))
254 try:
255 # Apply rate limiting before page request
256 self._last_wait_time = self.rate_tracker.apply_rate_limit(
257 self.engine_type
258 )
260 # Get the full page
261 page = wikipedia.page(title, auto_suggest=False)
263 # Create a full result with all information
264 result = {
265 "title": page.title,
266 "link": page.url,
267 "snippet": item.get("snippet", ""), # Keep existing snippet
268 "source": "Wikipedia",
269 }
271 # Add additional information
272 result["content"] = page.content
273 result["full_content"] = page.content
274 result["full_content_is_complete"] = True
275 result["categories"] = page.categories
276 result["references"] = page.references
277 result["links"] = page.links
278 result["images"] = page.images
279 result["sections"] = page.sections
281 results.append(result)
283 except (
284 wikipedia.exceptions.DisambiguationError,
285 wikipedia.exceptions.PageError,
286 wikipedia.exceptions.WikipediaException,
287 ):
288 # If error, use the preview
289 logger.warning(f"Error getting full content for '{safe_title}'")
290 results.append(item)
291 except Exception as e:
292 safe_msg = self._scrub_error(e)
293 logger.exception(
294 f"Unexpected error getting full content for "
295 f"'{safe_title}' ({type(e).__name__}): {safe_msg}"
296 )
297 results.append(item)
299 return results
301 def get_summary(self, title: str, sentences: Optional[int] = None) -> str:
302 """
303 Get a summary of a specific Wikipedia page.
305 Args:
306 title: Title of the Wikipedia page
307 sentences: Number of sentences to include (defaults to self.sentences)
309 Returns:
310 Summary of the page
311 """
312 sentences = sentences or self.sentences
313 try:
314 return str(
315 wikipedia.summary(
316 title, sentences=sentences, auto_suggest=False
317 )
318 )
319 except wikipedia.exceptions.DisambiguationError as e:
320 if e.options and len(e.options) > 0:
321 return str(
322 wikipedia.summary(
323 e.options[0], sentences=sentences, auto_suggest=False
324 )
325 )
326 raise
328 def get_page(self, title: str) -> Dict[str, Any]:
329 """
330 Get detailed information about a specific Wikipedia page.
332 Args:
333 title: Title of the Wikipedia page
335 Returns:
336 Dictionary with page information
337 """
338 include_content = self.include_content
340 try:
341 page = wikipedia.page(title, auto_suggest=False)
343 result = {
344 "title": page.title,
345 "link": page.url,
346 "snippet": self.get_summary(title, self.sentences),
347 "source": "Wikipedia",
348 }
350 # Add additional information if requested
351 if include_content:
352 result["content"] = page.content
353 result["full_content"] = page.content
354 result["full_content_is_complete"] = True
355 result["categories"] = page.categories
356 result["references"] = page.references
357 result["links"] = page.links
358 result["images"] = page.images
359 result["sections"] = page.sections
361 return result
362 except wikipedia.exceptions.DisambiguationError as e:
363 if e.options and len(e.options) > 0:
364 return self.get_page(e.options[0])
365 raise
367 def set_language(self, language: str) -> None:
368 """
369 Change the Wikipedia language.
371 Args:
372 language: Language code (e.g., 'en', 'fr', 'es')
373 """
374 wikipedia.set_lang(language)