Coverage for src/local_deep_research/web_search_engines/engines/search_engine_scaleserp.py: 99%
78 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1from ...security.secure_logging import logger
2from typing import Any, Dict, List, Optional
3import requests
5from langchain_core.language_models import BaseLLM
7from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
8from ..rate_limiting import RateLimitError
9from ...security import safe_get
12class ScaleSerpSearchEngine(BaseSearchEngine):
13 """Google search engine implementation using ScaleSerp API with caching support"""
15 # Mark as public search engine
16 is_public = True
17 egress_sensitivity = Sensitivity.NON_SENSITIVE
18 egress_exposure = Exposure.EXPOSING
19 # Mark as generic search engine (general web search via Google)
20 is_generic = True
22 def __init__(
23 self,
24 max_results: int = 10,
25 location: str = "United States",
26 language: str = "en",
27 device: str = "desktop",
28 safe_search: bool = True,
29 api_key: Optional[str] = None,
30 llm: Optional[BaseLLM] = None,
31 include_full_content: bool = False,
32 max_filtered_results: Optional[int] = None,
33 settings_snapshot: Optional[Dict[str, Any]] = None,
34 enable_cache: bool = True,
35 **kwargs,
36 ):
37 """
38 Initialize the ScaleSerp search engine.
40 Args:
41 max_results: Maximum number of search results (default 10, max 100)
42 location: Location for localized results (e.g., 'United States', 'London,England,United Kingdom')
43 language: Language code for results (e.g., 'en', 'es', 'fr')
44 device: Device type for search ('desktop' or 'mobile')
45 safe_search: Whether to enable safe search
46 api_key: ScaleSerp API key (can also be set in settings)
47 llm: Language model for relevance filtering
48 include_full_content: Retained for backward compatibility; full-content
49 retrieval is handled via the factory wrapper.
50 max_filtered_results: Maximum number of results to keep after filtering
51 settings_snapshot: Settings snapshot for thread context
52 enable_cache: Whether to use ScaleSerp's 1-hour caching (saves costs for repeated searches)
53 **kwargs: Additional parameters (ignored but accepted for compatibility)
54 """
55 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
56 super().__init__(
57 llm=llm,
58 max_filtered_results=max_filtered_results,
59 max_results=max_results,
60 include_full_content=include_full_content,
61 settings_snapshot=settings_snapshot,
62 )
63 self.location = location
64 self.language = language
65 self.device = device
66 self.safe_search = safe_search
67 self.enable_cache = enable_cache # ScaleSerp's unique caching feature
69 # Get API key - check params, settings, or env vars
70 scaleserp_api_key = self._resolve_api_key(
71 api_key,
72 "search.engine.web.scaleserp.api_key",
73 engine_name="ScaleSerp",
74 settings_snapshot=settings_snapshot,
75 )
77 self.api_key = scaleserp_api_key
78 self.base_url = "https://api.scaleserp.com/search"
80 # Initialize per-query attributes (reset in _get_previews per search)
81 self._knowledge_graph = None
83 # Full-content retrieval is the factory wrapper's job — the
84 # wrapper fetches pages via ``batch_fetch_and_extract`` and
85 # populates ``full_content`` on each result; the inner engine
86 # only emits snippets from ``_get_previews``.
88 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
89 """
90 Get preview information from ScaleSerp API.
92 Args:
93 query: The search query
95 Returns:
96 List of preview dictionaries
97 """
98 logger.info("Getting search results from ScaleSerp API")
100 # Reset per-query attributes to prevent leakage between searches
101 self._knowledge_graph = None
103 try:
104 # Build request parameters
105 params = {
106 "api_key": self.api_key,
107 "q": query,
108 "num": min(self.max_results, 100), # ScaleSerp max is 100
109 "location": self.location,
110 "hl": self.language,
111 "device": self.device,
112 }
114 # Add safe search if enabled
115 if self.safe_search:
116 params["safe"] = "on"
118 # ScaleSerp automatically caches identical queries for 1 hour
119 # Cached results are served instantly and don't consume API credits
120 if self.enable_cache:
121 params["output"] = (
122 "json" # Ensure JSON output for cache detection
123 )
124 logger.debug(
125 "ScaleSerp caching enabled - identical searches within 1 hour are free"
126 )
128 # Apply rate limiting before request
129 self._last_wait_time = self.rate_tracker.apply_rate_limit(
130 self.engine_type
131 )
133 # Make API request
134 response = safe_get(self.base_url, params=params, timeout=30)
136 # Check for rate limits
137 self._raise_if_rate_limit(response.status_code)
139 response.raise_for_status()
141 data = response.json()
143 # Extract organic results
144 organic_results = data.get("organic_results", [])
146 # Format results as previews
147 previews = []
149 # Check if results were served from cache for monitoring
150 from_cache = data.get("request_info", {}).get("cached", False)
152 for idx, result in enumerate(organic_results):
153 # Extract display link — ScaleSerp uses truncated URL as fallback
154 # `or ""` guards against a JSON null link: the fallback slice
155 # is evaluated eagerly and would raise TypeError on None.
156 link = result.get("link") or ""
157 display_link = self._extract_display_link(
158 link, fallback=link[:50]
159 )
161 preview = {
162 "id": idx,
163 "title": result.get("title", ""),
164 "link": link,
165 "snippet": result.get("snippet", ""),
166 "displayed_link": display_link,
167 "position": result.get("position", idx + 1),
168 "from_cache": from_cache, # Add cache status for monitoring
169 }
171 # Store full ScaleSerp result for later
172 preview["_full_result"] = result
174 # Include rich snippets if available
175 if "rich_snippet" in result:
176 preview["rich_snippet"] = result["rich_snippet"]
178 # Include date if available
179 if "date" in result:
180 preview["date"] = result["date"]
182 # Include sitelinks if available
183 if "sitelinks" in result:
184 preview["sitelinks"] = result["sitelinks"]
186 previews.append(preview)
188 # Store the previews for potential full content retrieval
189 self._search_results = previews
191 # Store knowledge graph if available
192 if "knowledge_graph" in data:
193 self._knowledge_graph = data["knowledge_graph"]
194 logger.info(
195 f"Found knowledge graph for query: {data['knowledge_graph'].get('title', 'Unknown')}"
196 )
198 # Log if result was served from cache
199 if from_cache:
200 logger.debug(
201 "Result served from ScaleSerp cache - no API credit used!"
202 )
204 return previews
206 except RateLimitError:
207 raise # Re-raise rate limit errors
208 except requests.exceptions.RequestException as e:
209 safe_msg = self._scrub_error(e)
210 logger.warning(
211 f"Error getting ScaleSerp API results: {safe_msg}. "
212 "Check API docs: https://docs.scaleserp.com"
213 )
214 self._raise_if_rate_limit(e)
215 return []
216 except Exception as e:
217 safe_msg = self._scrub_error(e)
218 logger.warning(
219 f"Unexpected error getting ScaleSerp API results: {safe_msg}"
220 )
221 return []
223 def _get_full_content(
224 self, relevant_items: List[Dict[str, Any]]
225 ) -> List[Dict[str, Any]]:
226 """
227 Get full content for the relevant search results.
228 Extends base implementation to include knowledge graph data.
230 Args:
231 relevant_items: List of relevant preview dictionaries
233 Returns:
234 List of result dictionaries with full content if requested
235 """
236 results = super()._get_full_content(relevant_items)
238 # Include knowledge graph if available
239 if results and hasattr(self, "_knowledge_graph"): 239 ↛ 242line 239 didn't jump to line 242 because the condition on line 239 was always true
240 results[0]["knowledge_graph"] = self._knowledge_graph
242 return results
244 def _temp_attributes(self):
245 """Return list of temporary attribute names to clean up after run()."""
246 return super()._temp_attributes() + [
247 "_knowledge_graph",
248 ]