Coverage for src/local_deep_research/web_search_engines/engines/search_engine_scaleserp.py: 99%
79 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +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: Whether to include full webpage content in results
49 max_filtered_results: Maximum number of results to keep after filtering
50 settings_snapshot: Settings snapshot for thread context
51 enable_cache: Whether to use ScaleSerp's 1-hour caching (saves costs for repeated searches)
52 **kwargs: Additional parameters (ignored but accepted for compatibility)
53 """
54 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
55 super().__init__(
56 llm=llm,
57 max_filtered_results=max_filtered_results,
58 max_results=max_results,
59 include_full_content=include_full_content,
60 settings_snapshot=settings_snapshot,
61 )
62 self.location = location
63 self.language = language
64 self.device = device
65 self.safe_search = safe_search
66 self.enable_cache = enable_cache # ScaleSerp's unique caching feature
68 # Get API key - check params, settings, or env vars
69 scaleserp_api_key = self._resolve_api_key(
70 api_key,
71 "search.engine.web.scaleserp.api_key",
72 engine_name="ScaleSerp",
73 settings_snapshot=settings_snapshot,
74 )
76 self.api_key = scaleserp_api_key
77 self.base_url = "https://api.scaleserp.com/search"
79 # Initialize per-query attributes (reset in _get_previews per search)
80 self._knowledge_graph = None
82 # If full content is requested, initialize FullSearchResults
83 self._init_full_search(
84 web_search=None, # We'll handle the search ourselves
85 language=language,
86 max_results=max_results,
87 region=location,
88 time_period=None,
89 safe_search="Moderate" if safe_search else "Off",
90 )
92 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
93 """
94 Get preview information from ScaleSerp API.
96 Args:
97 query: The search query
99 Returns:
100 List of preview dictionaries
101 """
102 logger.info("Getting search results from ScaleSerp API")
104 # Reset per-query attributes to prevent leakage between searches
105 self._knowledge_graph = None
107 try:
108 # Build request parameters
109 params = {
110 "api_key": self.api_key,
111 "q": query,
112 "num": min(self.max_results, 100), # ScaleSerp max is 100
113 "location": self.location,
114 "hl": self.language,
115 "device": self.device,
116 }
118 # Add safe search if enabled
119 if self.safe_search:
120 params["safe"] = "on"
122 # ScaleSerp automatically caches identical queries for 1 hour
123 # Cached results are served instantly and don't consume API credits
124 if self.enable_cache:
125 params["output"] = (
126 "json" # Ensure JSON output for cache detection
127 )
128 logger.debug(
129 "ScaleSerp caching enabled - identical searches within 1 hour are free"
130 )
132 # Apply rate limiting before request
133 self._last_wait_time = self.rate_tracker.apply_rate_limit(
134 self.engine_type
135 )
137 # Make API request
138 response = safe_get(self.base_url, params=params, timeout=30)
140 # Check for rate limits
141 self._raise_if_rate_limit(response.status_code)
143 response.raise_for_status()
145 data = response.json()
147 # Extract organic results
148 organic_results = data.get("organic_results", [])
150 # Format results as previews
151 previews = []
153 # Check if results were served from cache for monitoring
154 from_cache = data.get("request_info", {}).get("cached", False)
156 for idx, result in enumerate(organic_results):
157 # Extract display link — ScaleSerp uses truncated URL as fallback
158 # `or ""` guards against a JSON null link: the fallback slice
159 # is evaluated eagerly and would raise TypeError on None.
160 link = result.get("link") or ""
161 display_link = self._extract_display_link(
162 link, fallback=link[:50]
163 )
165 preview = {
166 "id": idx,
167 "title": result.get("title", ""),
168 "link": link,
169 "snippet": result.get("snippet", ""),
170 "displayed_link": display_link,
171 "position": result.get("position", idx + 1),
172 "from_cache": from_cache, # Add cache status for monitoring
173 }
175 # Store full ScaleSerp result for later
176 preview["_full_result"] = result
178 # Include rich snippets if available
179 if "rich_snippet" in result:
180 preview["rich_snippet"] = result["rich_snippet"]
182 # Include date if available
183 if "date" in result:
184 preview["date"] = result["date"]
186 # Include sitelinks if available
187 if "sitelinks" in result:
188 preview["sitelinks"] = result["sitelinks"]
190 previews.append(preview)
192 # Store the previews for potential full content retrieval
193 self._search_results = previews
195 # Store knowledge graph if available
196 if "knowledge_graph" in data:
197 self._knowledge_graph = data["knowledge_graph"]
198 logger.info(
199 f"Found knowledge graph for query: {data['knowledge_graph'].get('title', 'Unknown')}"
200 )
202 # Log if result was served from cache
203 if from_cache:
204 logger.debug(
205 "Result served from ScaleSerp cache - no API credit used!"
206 )
208 return previews
210 except RateLimitError:
211 raise # Re-raise rate limit errors
212 except requests.exceptions.RequestException as e:
213 safe_msg = self._scrub_error(e)
214 logger.warning(
215 f"Error getting ScaleSerp API results: {safe_msg}. "
216 "Check API docs: https://docs.scaleserp.com"
217 )
218 self._raise_if_rate_limit(e)
219 return []
220 except Exception as e:
221 safe_msg = self._scrub_error(e)
222 logger.warning(
223 f"Unexpected error getting ScaleSerp API results: {safe_msg}"
224 )
225 return []
227 def _get_full_content(
228 self, relevant_items: List[Dict[str, Any]]
229 ) -> List[Dict[str, Any]]:
230 """
231 Get full content for the relevant search results.
232 Extends base implementation to include knowledge graph data.
234 Args:
235 relevant_items: List of relevant preview dictionaries
237 Returns:
238 List of result dictionaries with full content if requested
239 """
240 results = super()._get_full_content(relevant_items)
242 # Include knowledge graph if available
243 if results and hasattr(self, "_knowledge_graph"): 243 ↛ 246line 243 didn't jump to line 246 because the condition on line 243 was always true
244 results[0]["knowledge_graph"] = self._knowledge_graph
246 return results
248 def _temp_attributes(self):
249 """Return list of temporary attribute names to clean up after run()."""
250 return super()._temp_attributes() + [
251 "_knowledge_graph",
252 ]