Coverage for src/local_deep_research/web_search_engines/engines/search_engine_stackexchange.py: 95%
237 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Stack Exchange search engine for Q&A content."""
3import html
4import re
5import time
6from typing import Any, Dict, List, Optional
8import requests
9from langchain_core.language_models import BaseLLM
11from ...constants import USER_AGENT
12from ...security.safe_requests import safe_get
13from ...security.secure_logging import logger
14from ..rate_limiting import RateLimitError
15from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
18class StackExchangeSearchEngine(BaseSearchEngine):
19 """
20 Stack Exchange search engine for Q&A content.
22 Provides access to Stack Overflow and other Stack Exchange sites.
23 No authentication required (300 requests/day without key).
24 """
26 is_public = True
27 egress_sensitivity = Sensitivity.NON_SENSITIVE
28 egress_exposure = Exposure.EXPOSING
29 is_generic = False
30 is_scientific = False
31 is_code = True
32 is_lexical = True
33 needs_llm_relevance_filter = True
35 # Common Stack Exchange sites
36 SITES = {
37 "stackoverflow": "Stack Overflow",
38 "serverfault": "Server Fault",
39 "superuser": "Super User",
40 "askubuntu": "Ask Ubuntu",
41 "unix": "Unix & Linux",
42 "math": "Mathematics",
43 "physics": "Physics",
44 "stats": "Cross Validated",
45 "security": "Information Security",
46 "dba": "Database Administrators",
47 }
49 # Sites with their own .com domains (not *.stackexchange.com)
50 SITE_DOMAINS = {
51 "stackoverflow": "stackoverflow.com",
52 "serverfault": "serverfault.com",
53 "superuser": "superuser.com",
54 "askubuntu": "askubuntu.com",
55 }
57 def __init__(
58 self,
59 max_results: int = 10,
60 site: str = "stackoverflow",
61 sort: str = "relevance",
62 accepted_only: bool = False,
63 has_answers: bool = False,
64 min_score: Optional[int] = None,
65 tagged: Optional[str] = None,
66 llm: Optional[BaseLLM] = None,
67 max_filtered_results: Optional[int] = None,
68 settings_snapshot: Optional[Dict[str, Any]] = None,
69 **kwargs,
70 ):
71 """
72 Initialize the Stack Exchange search engine.
74 Args:
75 max_results: Maximum number of search results
76 site: Stack Exchange site to search (stackoverflow, serverfault, etc.)
77 sort: Sort order (relevance, votes, creation, activity)
78 accepted_only: Only return questions with accepted answers
79 has_answers: Only return questions that have answers
80 min_score: Minimum score for questions
81 tagged: Filter by tags (semicolon separated)
82 llm: Language model for relevance filtering
83 max_filtered_results: Maximum results after filtering
84 settings_snapshot: Settings snapshot for thread context
85 """
86 super().__init__(
87 llm=llm,
88 max_filtered_results=max_filtered_results,
89 max_results=max_results,
90 settings_snapshot=settings_snapshot,
91 **kwargs,
92 )
94 # Validate site parameter
95 if site not in self.SITES:
96 valid_sites = ", ".join(self.SITES.keys())
97 raise ValueError(
98 f"Invalid site: '{site}'. Must be one of: {valid_sites}"
99 )
101 # Validate sort parameter
102 valid_sorts = ("relevance", "votes", "creation", "activity")
103 if sort not in valid_sorts:
104 raise ValueError(
105 f"Invalid sort: '{sort}'. Must be one of: {', '.join(valid_sorts)}"
106 )
108 # Validate sort/min_score combination: the StackExchange API's "min"
109 # parameter works with any sort except "relevance".
110 if min_score is not None and sort == "relevance":
111 raise ValueError(
112 "min_score requires a numeric sort order (votes, creation, or activity). "
113 "sort='relevance' does not support the 'min' parameter."
114 )
116 self.site = site
117 self.sort = sort
118 self.accepted_only = accepted_only
119 self.has_answers = has_answers
120 self.min_score = min_score
121 self.tagged = tagged
123 self.base_url = "https://api.stackexchange.com/2.3"
124 self.search_url = f"{self.base_url}/search/advanced"
126 # User-Agent and required headers for API requests
127 self.headers = {
128 "User-Agent": USER_AGENT,
129 "Accept-Encoding": "gzip, deflate",
130 }
132 # Track backoff requirement from API responses
133 self._backoff_until: float = 0
135 def _apply_backoff(self) -> None:
136 """Apply backoff if required by previous API response."""
137 if self._backoff_until > 0:
138 wait_time = self._backoff_until - time.time()
139 if wait_time > 0: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 logger.info(
141 f"Stack Exchange backoff: waiting {wait_time:.1f} seconds"
142 )
143 time.sleep(wait_time)
144 self._backoff_until = 0
146 def _handle_backoff(self, data: Dict[str, Any]) -> None:
147 """Handle backoff field in API response."""
148 backoff = data.get("backoff")
149 if backoff:
150 self._backoff_until = time.time() + min(int(backoff), 300)
151 logger.warning(
152 f"Stack Exchange API requested backoff of {backoff} seconds"
153 )
155 def _build_query_params(self, query: str) -> Dict[str, Any]:
156 """Build query parameters for the API request."""
157 params = {
158 "q": query,
159 "site": self.site,
160 "order": "desc",
161 "sort": self.sort,
162 "pagesize": min(self.max_results, 100),
163 "filter": "withbody", # Include question body
164 }
166 if self.accepted_only:
167 params["accepted"] = "True"
169 if self.has_answers: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true
170 params["answers"] = "1"
172 if self.min_score is not None:
173 params["min"] = self.min_score
175 if self.tagged:
176 params["tagged"] = self.tagged
178 return params
180 def _decode_html(self, text: str) -> str:
181 """Decode HTML entities in text."""
182 return html.unescape(text)
184 def _get_site_name(self) -> str:
185 """Get human-readable site name."""
186 return self.SITES.get(self.site, self.site.title())
188 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
189 """
190 Get preview information for Stack Exchange questions.
192 Args:
193 query: The search query
195 Returns:
196 List of preview dictionaries
197 """
198 logger.info(f"Getting Stack Exchange previews on {self.site}")
200 # Apply rate limiting
201 self._last_wait_time = self.rate_tracker.apply_rate_limit(
202 self.engine_type
203 )
205 # Apply backoff if required by previous API response
206 self._apply_backoff()
208 try:
209 params = self._build_query_params(query)
211 response = safe_get(
212 self.search_url,
213 params=params,
214 headers=self.headers,
215 timeout=30,
216 )
218 self._raise_if_rate_limit(response.status_code)
220 response.raise_for_status()
221 data = response.json()
223 # Handle backoff if present in response
224 self._handle_backoff(data)
226 # Check for API errors
227 if "error_id" in data:
228 error_msg = data.get("error_message", "Unknown error")
229 logger.error(f"Stack Exchange API error: {error_msg}")
230 return []
232 results = data.get("items", [])
233 quota_remaining = data.get("quota_remaining", 0)
234 logger.info(
235 f"Found {len(results)} Stack Exchange results, quota remaining: {quota_remaining}"
236 )
238 if quota_remaining < 10:
239 logger.warning(f"Stack Exchange quota low: {quota_remaining}")
241 previews = []
242 for question in results[: self.max_results]:
243 try:
244 question_id = question.get("question_id")
245 title = self._decode_html(question.get("title", "Untitled"))
247 # Get owner info
248 owner = question.get("owner", {})
249 author = self._decode_html(
250 owner.get("display_name", "Unknown")
251 )
252 author_link = owner.get("link", "")
253 author_reputation = owner.get("reputation", 0)
255 # Get question stats
256 score = question.get("score", 0)
257 view_count = question.get("view_count", 0)
258 answer_count = question.get("answer_count", 0)
259 is_answered = question.get("is_answered", False)
260 accepted_answer_id = question.get("accepted_answer_id")
262 # Get tags
263 tags = question.get("tags", [])
265 # Build answer status prefix
266 status_parts = []
267 if is_answered:
268 status = f"Answered ({answer_count} answer{'s' if answer_count != 1 else ''}"
269 if accepted_answer_id: 269 ↛ 271line 269 didn't jump to line 271 because the condition on line 269 was always true
270 status += ", accepted"
271 status += ")"
272 status_parts.append(status)
273 elif answer_count > 0: 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 status_parts.append(
275 f"{answer_count} answer{'s' if answer_count != 1 else ''}"
276 )
277 if tags:
278 status_parts.append(f"Tags: {', '.join(tags[:4])}")
279 prefix = " | ".join(status_parts)
281 # Get body (snippet)
282 body = question.get("body", "")
283 # Strip HTML for snippet
284 body_text = html.unescape(re.sub(r"<[^>]+>", " ", body))
285 body_text = " ".join(body_text.split())[:1000]
286 snippet = f"{prefix} | {body_text}" if prefix else body_text
288 # Get dates
289 creation_date = question.get("creation_date", 0)
290 last_activity = question.get("last_activity_date", 0)
292 # Build link
293 fallback_domain = self.SITE_DOMAINS.get(
294 self.site, f"{self.site}.stackexchange.com"
295 )
296 link = question.get(
297 "link",
298 f"https://{fallback_domain}/questions/{question_id}",
299 )
301 preview = {
302 "id": str(question_id),
303 "title": title,
304 "link": link,
305 "snippet": snippet,
306 "author": author,
307 "author_link": author_link,
308 "author_reputation": author_reputation,
309 "score": score,
310 "view_count": view_count,
311 "answer_count": answer_count,
312 "is_answered": is_answered,
313 "has_accepted_answer": accepted_answer_id is not None,
314 "tags": tags,
315 "creation_date": creation_date,
316 "last_activity_date": last_activity,
317 "site": self.site,
318 "source": self._get_site_name(),
319 "_raw": question,
320 }
322 previews.append(preview)
324 except Exception as e:
325 safe_msg = self._scrub_error(e)
326 logger.exception(
327 f"Error parsing Stack Exchange question ({type(e).__name__}): {safe_msg}"
328 )
329 continue
331 return previews
333 except (requests.RequestException, ValueError) as e:
334 safe_msg = self._scrub_error(e)
335 logger.exception(
336 f"Stack Exchange API request failed ({type(e).__name__}): {safe_msg}"
337 )
338 self._raise_if_rate_limit(e)
339 return []
341 def _get_full_content(
342 self, relevant_items: List[Dict[str, Any]]
343 ) -> List[Dict[str, Any]]:
344 """
345 Get full content for the relevant Stack Exchange questions.
347 Fetches the question body and top answers from the API.
349 Args:
350 relevant_items: List of relevant preview dictionaries
352 Returns:
353 List of result dictionaries with full content
354 """
355 logger.info(
356 f"Getting full content for {len(relevant_items)} Stack Exchange questions"
357 )
359 results = []
360 for item in relevant_items:
361 result = item.copy()
363 raw = item.get("_raw", {})
364 if raw:
365 # Get full body
366 body = raw.get("body", "")
367 clean_body = html.unescape(re.sub(r"<[^>]+>", " ", body))
368 clean_body = " ".join(clean_body.split())
370 # Build content with question + answers
371 content_parts = []
372 content_parts.append(
373 f"Question: {result.get('title', 'Untitled')}"
374 )
375 if result.get("tags"):
376 content_parts.append(f"Tags: {', '.join(result['tags'])}")
377 content_parts.append(f"\n{clean_body}")
379 # Fetch top answers
380 question_id = raw.get("question_id")
381 if question_id:
382 try:
383 question_id = int(question_id)
384 except (TypeError, ValueError):
385 question_id = None
386 if question_id:
387 answers = self._fetch_top_answers(
388 question_id, max_answers=3
389 )
390 if answers: 390 ↛ 407line 390 didn't jump to line 407 because the condition on line 390 was always true
391 content_parts.append(
392 f"\n--- Top Answers ({len(answers)}) ---"
393 )
394 for ans in answers:
395 ans_body = html.unescape(
396 re.sub(r"<[^>]+>", " ", ans.get("body", ""))
397 )
398 ans_body = " ".join(ans_body.split())[:3000]
399 score = ans.get("score", 0)
400 accepted = ans.get("is_accepted", False)
401 label = f"[Score: {score}"
402 if accepted: 402 ↛ 404line 402 didn't jump to line 404 because the condition on line 402 was always true
403 label += ", Accepted"
404 label += "]"
405 content_parts.append(f"\n{label}\n{ans_body}")
407 result["content"] = "\n".join(content_parts)
409 # Clean up internal fields
410 if "_raw" in result:
411 del result["_raw"]
413 results.append(result)
415 return results
417 def _fetch_top_answers(
418 self, question_id: int, max_answers: int = 3
419 ) -> List[Dict[str, Any]]:
420 """Fetch top answers for a question, sorted by votes."""
421 try:
422 self._apply_backoff()
423 url = f"{self.base_url}/questions/{question_id}/answers"
424 params = {
425 "site": self.site,
426 "order": "desc",
427 "sort": "votes",
428 "pagesize": max_answers,
429 "filter": "withbody",
430 }
431 response = safe_get(
432 url, params=params, headers=self.headers, timeout=30
433 )
434 self._raise_if_rate_limit(response.status_code)
435 response.raise_for_status()
436 data = response.json()
437 self._handle_backoff(data)
439 if "error_id" in data:
440 logger.warning(
441 f"Stack Exchange API error fetching answers for "
442 f"{question_id}: {data.get('error_message', 'Unknown')}"
443 )
444 return []
446 quota_remaining = data.get("quota_remaining")
447 if quota_remaining is not None and quota_remaining < 10:
448 logger.warning(f"Stack Exchange quota low: {quota_remaining}")
450 return data.get("items", []) # type: ignore[no-any-return]
451 except (RateLimitError, ValueError):
452 raise
453 except Exception:
454 logger.warning(
455 f"Failed to fetch answers for question {question_id}"
456 )
457 return []
459 def get_question(self, question_id: int) -> Optional[Dict[str, Any]]:
460 """
461 Get a specific question by ID.
463 Args:
464 question_id: The Stack Exchange question ID
466 Returns:
467 Question dictionary or None
468 """
469 try:
470 url = f"{self.base_url}/questions/{question_id}"
471 params = {"site": self.site, "filter": "withbody"}
472 response = safe_get(
473 url, params=params, headers=self.headers, timeout=30
474 )
475 self._raise_if_rate_limit(response.status_code)
476 response.raise_for_status()
477 data = response.json()
478 self._handle_backoff(data)
479 items = data.get("items", [])
480 return items[0] if items else None
481 except RateLimitError:
482 raise
483 except Exception as e:
484 safe_msg = self._scrub_error(e)
485 logger.exception(
486 f"Error fetching Stack Exchange question {question_id} ({type(e).__name__}): {safe_msg}"
487 )
488 return None
490 def get_answers(self, question_id: int) -> List[Dict[str, Any]]:
491 """
492 Get answers for a specific question.
494 Args:
495 question_id: The Stack Exchange question ID
497 Returns:
498 List of answer dictionaries
499 """
500 try:
501 url = f"{self.base_url}/questions/{question_id}/answers"
502 params = {
503 "site": self.site,
504 "order": "desc",
505 "sort": "votes",
506 "filter": "withbody",
507 }
508 response = safe_get(
509 url, params=params, headers=self.headers, timeout=30
510 )
511 self._raise_if_rate_limit(response.status_code)
512 response.raise_for_status()
513 data = response.json()
514 self._handle_backoff(data)
515 return data.get("items", []) # type: ignore[no-any-return]
516 except RateLimitError:
517 raise
518 except Exception as e:
519 safe_msg = self._scrub_error(e)
520 logger.exception(
521 f"Error fetching answers for question {question_id} ({type(e).__name__}): {safe_msg}"
522 )
523 return []
525 def search_by_tag(self, tag: str, query: str = "") -> List[Dict[str, Any]]:
526 """
527 Search questions by tag.
529 Args:
530 tag: The tag to filter by
531 query: Optional search query
533 Returns:
534 List of matching questions
535 """
536 original_tagged = self.tagged
537 try:
538 self.tagged = tag
539 return self.run(query)
540 finally:
541 self.tagged = original_tagged