Coverage for src/local_deep_research/web_search_engines/engines/search_engine_google_pse.py: 98%
132 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1from ...security.secure_logging import logger
2import random
3import time
4from typing import Any, Dict, List, Optional
6from langchain_core.language_models import BaseLLM
7from requests.exceptions import RequestException
9from ...security.safe_requests import safe_get
10from ..rate_limiting import RateLimitError
11from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
14class GooglePSESearchEngine(BaseSearchEngine):
15 """Google Programmable Search Engine implementation"""
17 # Mark as public search engine
18 is_public = True
19 egress_sensitivity = Sensitivity.NON_SENSITIVE
20 egress_exposure = Exposure.EXPOSING
21 # Mark as generic search engine (general web search)
22 is_generic = True
24 def __init__(
25 self,
26 max_results: int = 10,
27 region: str = "us",
28 safe_search: bool = True,
29 search_language: str = "English",
30 api_key: Optional[str] = None,
31 search_engine_id: Optional[str] = None,
32 llm: Optional[BaseLLM] = None,
33 include_full_content: bool = False,
34 max_filtered_results: Optional[int] = None,
35 settings_snapshot: Optional[Dict[str, Any]] = None,
36 max_retries: int = 3,
37 retry_delay: float = 2.0,
38 **kwargs,
39 ):
40 """
41 Initialize the Google Programmable Search Engine.
43 Args:
44 max_results: Maximum number of search results
45 region: Region code for search results
46 safe_search: Whether to enable safe search
47 search_language: Language for search results
48 api_key: Google API key (can also be set via LDR_SEARCH_ENGINE_WEB_GOOGLE_PSE_API_KEY env var or in UI settings)
49 search_engine_id: Google CSE ID (can also be set via LDR_SEARCH_ENGINE_WEB_GOOGLE_PSE_ENGINE_ID env var or in UI settings)
50 llm: Language model for relevance filtering
51 include_full_content: Whether to include full webpage content in results
52 max_filtered_results: Maximum number of results to keep after filtering
53 max_retries: Maximum number of retry attempts for API requests
54 retry_delay: Base delay in seconds between retry attempts
55 **kwargs: Additional parameters (ignored but accepted for compatibility)
56 """
57 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
58 super().__init__(
59 llm=llm,
60 max_filtered_results=max_filtered_results,
61 max_results=max_results,
62 include_full_content=include_full_content,
63 settings_snapshot=settings_snapshot,
64 **kwargs,
65 )
67 # Google PSE returns full content via its API (snippet + htmlSnippet),
68 # so _init_full_search() is intentionally not called here.
70 # Retry configuration
71 self.max_retries = max_retries
72 self.retry_delay = retry_delay
74 # Rate limiting - keep track of last request time
75 self.last_request_time: float = 0.0
76 self.min_request_interval = (
77 0.5 # Minimum time between requests in seconds
78 )
80 # Language code mapping — Google PSE uses "zh-CN" for Chinese
81 from ...utilities.search_utilities import LANGUAGE_CODE_MAP
83 language_code_mapping = {**LANGUAGE_CODE_MAP, "chinese": "zh-CN"}
85 # Get language code
86 search_language = search_language.lower()
87 self.language = language_code_mapping.get(search_language, "en")
89 # Safe search setting
90 self.safe = "active" if safe_search else "off"
92 # Region/Country setting
93 self.region = region
95 # API key and Search Engine ID - check params, env vars, or database
96 from ...config.thread_settings import (
97 get_setting_from_snapshot,
98 NoSettingsContextError,
99 )
101 self.api_key = api_key
102 if not self.api_key:
103 try:
104 self.api_key = get_setting_from_snapshot(
105 "search.engine.web.google_pse.api_key",
106 default=None,
107 settings_snapshot=self.settings_snapshot,
108 )
109 except NoSettingsContextError:
110 # No settings context available
111 logger.debug(
112 "No settings context available for Google PSE API key"
113 )
114 pass
116 self.search_engine_id = search_engine_id
117 if not self.search_engine_id:
118 try:
119 self.search_engine_id = get_setting_from_snapshot(
120 "search.engine.web.google_pse.engine_id",
121 default=None,
122 settings_snapshot=self.settings_snapshot,
123 )
124 except NoSettingsContextError:
125 # No settings context available
126 logger.debug(
127 "No settings context available for Google PSE engine ID"
128 )
129 pass
131 if not self.api_key:
132 raise ValueError(
133 "Google API key is required. Set it in the UI settings, use the api_key parameter, or set the LDR_SEARCH_ENGINE_WEB_GOOGLE_PSE_API_KEY environment variable."
134 )
135 if not self.search_engine_id:
136 raise ValueError(
137 "Google Search Engine ID is required. Set it in the UI settings, use the search_engine_id parameter, or set the LDR_SEARCH_ENGINE_WEB_GOOGLE_PSE_ENGINE_ID environment variable."
138 )
140 # Validate connection and credentials
141 self._validate_connection()
143 def _validate_connection(self):
144 """Test the connection to ensure API key and Search Engine ID are valid"""
145 try:
146 # Make a minimal test query
147 response = self._make_request("test")
149 # Check if we got a valid response
150 if response.get("error"):
151 error_msg = response["error"].get("message", "Unknown error")
152 raise ValueError(f"Google PSE API error: {error_msg}") # noqa: TRY301 — except only adds logging before re-raise
154 # If we get here, the connection is valid
155 logger.info("Google PSE connection validated successfully")
156 return True
158 except Exception as e:
159 # Log the error and re-raise a sanitized exception. Use
160 # logger.warning with the api_key redacted from str(e) so
161 # the upstream exception message — which can embed the key
162 # in the URL — does not leak. The re-raised exception uses
163 # `type(e)(safe_msg) from None` to preserve the original
164 # exception type (for callers that dispatch on it) while
165 # replacing the args with the redacted message and
166 # suppressing the traceback chain (which carries the URL
167 # in earlier frames).
168 safe_msg = self._scrub_error(e)
169 logger.warning(
170 f"Error validating Google PSE connection: {safe_msg}"
171 )
172 raise type(e)(safe_msg) from None
174 def _respect_rate_limit(self):
175 """Ensure we don't exceed rate limits by adding appropriate delay between requests"""
176 current_time = time.time()
177 elapsed = current_time - self.last_request_time
179 # If we've made a request recently, wait until the minimum interval has passed
180 if elapsed < self.min_request_interval:
181 sleep_time = self.min_request_interval - elapsed
182 logger.debug("Rate limiting: sleeping for {:.2f} s", sleep_time)
183 time.sleep(sleep_time)
185 # Update the last request time
186 self.last_request_time = time.time()
188 def _make_request(self, query: str, start_index: int = 1) -> Dict:
189 """
190 Make a request to the Google PSE API with retry logic and rate limiting
192 Args:
193 query: Search query string
194 start_index: Starting index for pagination
196 Returns:
197 JSON response from the API
199 Raises:
200 RequestException: If all retry attempts fail
201 """
202 # Base URL for the API
203 url = "https://www.googleapis.com/customsearch/v1"
205 # Parameters for the request
206 params = {
207 "key": self.api_key,
208 "cx": self.search_engine_id,
209 "q": query,
210 "num": min(10, self.max_results), # Max 10 per request
211 "start": start_index,
212 "safe": self.safe,
213 "lr": f"lang_{self.language}",
214 "gl": self.region,
215 }
217 # Implement retry logic with exponential backoff
218 attempt = 0
219 last_exception: Exception | None = None
221 while attempt < self.max_retries:
222 try:
223 # Add jitter to retries after the first attempt
224 if attempt > 0:
225 # Security: random jitter for exponential backoff retry, not security-sensitive
226 jitter = random.uniform(0.5, 1.5)
227 sleep_time = (
228 self.retry_delay * (2 ** (attempt - 1)) * jitter
229 )
230 logger.info(
231 "Retry attempt {} / {} for query '{}'. Waiting {} s",
232 attempt + 1,
233 self.max_retries,
234 query,
235 f"{sleep_time:.2f}",
236 )
237 time.sleep(sleep_time)
239 # Make the request
240 logger.debug(
241 "Making request to Google PSE API: {} (start_index={})",
242 query,
243 start_index,
244 )
245 # Apply rate limiting before request
246 self._last_wait_time = self.rate_tracker.apply_rate_limit(
247 self.engine_type
248 )
250 response = safe_get(url, params=params, timeout=10)
252 # Check for HTTP errors
253 response.raise_for_status()
255 # Return the JSON response
256 return response.json() # type: ignore[no-any-return]
258 except RequestException as e:
259 error_msg = str(e)
260 sanitized = self._sanitize_error_message(error_msg)
261 logger.warning(
262 "Request error on attempt {} / {}: {}",
263 attempt + 1,
264 self.max_retries,
265 sanitized,
266 )
268 # Check for rate limiting patterns
269 if (
270 "quota" in error_msg.lower()
271 or "quotaExceeded" in error_msg
272 or "dailyLimitExceeded" in error_msg
273 or "rateLimitExceeded" in error_msg
274 or "429" in error_msg
275 or "403" in error_msg
276 ):
277 raise RateLimitError(
278 f"Google PSE rate limit/quota exceeded: {sanitized}"
279 )
281 last_exception = e
282 except Exception as e:
283 error_msg = str(e)
284 sanitized = self._sanitize_error_message(error_msg)
285 logger.warning(
286 "Error on attempt {} / {}: {}",
287 attempt + 1,
288 self.max_retries,
289 sanitized,
290 )
292 # Check for rate limiting patterns in general errors
293 if "quota" in error_msg.lower() or "limit" in error_msg.lower():
294 raise RateLimitError(
295 f"Google PSE error (possible rate limit): {sanitized}"
296 )
298 last_exception = e
300 attempt += 1
302 # If we get here, all retries failed
303 error_msg = f"Failed to get response from Google PSE API after {self.max_retries} attempts"
304 logger.error(error_msg)
306 if last_exception: 306 ↛ 310line 306 didn't jump to line 310 because the condition on line 306 was always true
307 raise RequestException(
308 f"{error_msg}: {self._sanitize_error_message(str(last_exception))}"
309 )
310 raise RequestException(error_msg)
312 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
313 """Get search result previews/snippets"""
314 results = []
316 # Google PSE API returns a maximum of 10 results per request
317 # We may need to make multiple requests to get the desired number
318 start_index = 1
319 total_results = 0
321 while total_results < self.max_results: 321 ↛ 366line 321 didn't jump to line 366 because the condition on line 321 was always true
322 try:
323 response = self._make_request(query, start_index)
325 # Break if no items
326 if "items" not in response:
327 break
329 items = response.get("items", [])
331 # Process each result
332 for item in items:
333 title = item.get("title", "")
334 snippet = item.get("snippet", "")
335 url = item.get("link", "")
337 # Skip results without URL
338 if not url:
339 continue
341 results.append(
342 {
343 "title": title,
344 "snippet": snippet,
345 "link": url,
346 "source": "Google Programmable Search",
347 }
348 )
350 total_results += 1
351 if total_results >= self.max_results:
352 break
354 # Check if there are more results
355 if not items or total_results >= self.max_results:
356 break
358 # Update start index for next request
359 start_index += len(items)
361 except Exception as e:
362 safe_msg = self._scrub_error(e)
363 logger.warning(f"Error getting search results: {safe_msg}")
364 break
366 logger.info(
367 "Retrieved {} search results for query: '{}'", len(results), query
368 )
369 return results