Coverage for src/local_deep_research/web_search_engines/engines/search_engine_google_pse.py: 94%
135 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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
12from ._google_pse_rate_limiter import respect_rate_limit
15class GooglePSESearchEngine(BaseSearchEngine):
16 """Google Programmable Search Engine implementation"""
18 # Mark as public search engine
19 is_public = True
20 egress_sensitivity = Sensitivity.NON_SENSITIVE
21 egress_exposure = Exposure.EXPOSING
22 # Mark as generic search engine (general web search)
23 is_generic = True
25 def __init__(
26 self,
27 max_results: int = 10,
28 region: str = "us",
29 safe_search: bool = True,
30 search_language: str = "English",
31 api_key: Optional[str] = None,
32 search_engine_id: Optional[str] = None,
33 llm: Optional[BaseLLM] = None,
34 include_full_content: bool = False,
35 max_filtered_results: Optional[int] = None,
36 settings_snapshot: Optional[Dict[str, Any]] = None,
37 max_retries: int = 3,
38 retry_delay: float = 2.0,
39 **kwargs,
40 ):
41 """
42 Initialize the Google Programmable Search Engine.
44 Args:
45 max_results: Maximum number of search results
46 region: Region code for search results
47 safe_search: Whether to enable safe search
48 search_language: Language for search results
49 api_key: Google API key (can also be set via LDR_SEARCH_ENGINE_WEB_GOOGLE_PSE_API_KEY env var or in UI settings)
50 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)
51 llm: Language model for relevance filtering
52 include_full_content: Whether to include full webpage content in results
53 max_filtered_results: Maximum number of results to keep after filtering
54 max_retries: Maximum number of retry attempts for API requests
55 retry_delay: Base delay in seconds between retry attempts
56 **kwargs: Additional parameters (ignored but accepted for compatibility)
57 """
58 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
59 super().__init__(
60 llm=llm,
61 max_filtered_results=max_filtered_results,
62 max_results=max_results,
63 include_full_content=include_full_content,
64 settings_snapshot=settings_snapshot,
65 **kwargs,
66 )
68 # Google PSE returns full content via its API (snippet + htmlSnippet),
69 # so _init_full_search() is intentionally not called here.
71 # Retry configuration
72 self.max_retries = max_retries
73 self.retry_delay = retry_delay
75 self.min_request_interval = (
76 0.5 # Minimum time between requests in seconds
77 )
79 # Language code mapping — Google PSE uses "zh-CN" for Chinese
80 from ...utilities.search_utilities import LANGUAGE_CODE_MAP
82 language_code_mapping = {**LANGUAGE_CODE_MAP, "chinese": "zh-CN"}
84 # Get language code
85 search_language = search_language.lower()
86 self.language = language_code_mapping.get(search_language, "en")
88 # Safe search setting
89 self.safe = "active" if safe_search else "off"
91 # Region/Country setting
92 self.region = region
94 # API key and Search Engine ID - check params, env vars, or database
95 from ...config.thread_settings import (
96 get_setting_from_snapshot,
97 NoSettingsContextError,
98 )
100 self.api_key = api_key
101 if not self.api_key:
102 try:
103 self.api_key = get_setting_from_snapshot(
104 "search.engine.web.google_pse.api_key",
105 default=None,
106 settings_snapshot=self.settings_snapshot,
107 )
108 except NoSettingsContextError:
109 # No settings context available
110 logger.debug(
111 "No settings context available for Google PSE API key"
112 )
113 pass
115 self.search_engine_id = search_engine_id
116 if not self.search_engine_id:
117 try:
118 self.search_engine_id = get_setting_from_snapshot(
119 "search.engine.web.google_pse.engine_id",
120 default=None,
121 settings_snapshot=self.settings_snapshot,
122 )
123 except NoSettingsContextError:
124 # No settings context available
125 logger.debug(
126 "No settings context available for Google PSE engine ID"
127 )
128 pass
130 if not self.api_key:
131 raise ValueError(
132 "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."
133 )
134 if not self.search_engine_id:
135 raise ValueError(
136 "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."
137 )
139 # Validate connection and credentials
140 self._validate_connection()
142 def _validate_connection(self):
143 """Test the connection to ensure API key and Search Engine ID are valid"""
144 try:
145 # Make a minimal test query
146 response = self._make_request("test")
148 # Check if we got a valid response
149 if response.get("error"):
150 error_msg = response["error"].get("message", "Unknown error")
151 raise ValueError(f"Google PSE API error: {error_msg}") # noqa: TRY301 — except only adds logging before re-raise
153 # If we get here, the connection is valid
154 logger.info("Google PSE connection validated successfully")
155 return True
157 except Exception as e:
158 # Log the error and re-raise a sanitized exception. Use
159 # logger.warning with the api_key redacted from str(e) so
160 # the upstream exception message — which can embed the key
161 # in the URL — does not leak. The re-raised exception uses
162 # `type(e)(safe_msg) from None` to preserve the original
163 # exception type (for callers that dispatch on it) while
164 # replacing the args with the redacted message and
165 # suppressing the traceback chain (which carries the URL
166 # in earlier frames).
167 safe_msg = self._scrub_error(e)
168 logger.warning(
169 f"Error validating Google PSE connection: {safe_msg}"
170 )
171 raise type(e)(safe_msg) from None
173 def _respect_rate_limit(self) -> float:
174 """Enforce minimum spacing across engines using the same credentials."""
175 api_key = self.api_key
176 search_engine_id = self.search_engine_id
177 if api_key is None or search_engine_id is None: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true
178 raise ValueError("Google PSE credentials are required")
179 wait_time = respect_rate_limit(
180 api_key,
181 search_engine_id,
182 self.min_request_interval,
183 )
184 if wait_time > 0:
185 logger.debug("Rate limiting: sleeping for {:.2f} s", wait_time)
186 return wait_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 adaptive_wait = self.rate_tracker.apply_rate_limit(
247 self.engine_type
248 )
249 minimum_interval_wait = self._respect_rate_limit()
250 self._last_wait_time = adaptive_wait + minimum_interval_wait
252 response = safe_get(url, params=params, timeout=10)
254 # Check for HTTP errors
255 response.raise_for_status()
257 # Return the JSON response
258 return response.json() # type: ignore[no-any-return]
260 except RequestException as e:
261 error_msg = str(e)
262 sanitized = self._sanitize_error_message(error_msg)
263 logger.warning(
264 "Request error on attempt {} / {}: {}",
265 attempt + 1,
266 self.max_retries,
267 sanitized,
268 )
270 # Check for rate limiting patterns
271 if (
272 "quota" in error_msg.lower()
273 or "quotaExceeded" in error_msg
274 or "dailyLimitExceeded" in error_msg
275 or "rateLimitExceeded" in error_msg
276 or "429" in error_msg
277 or "403" in error_msg
278 ):
279 raise RateLimitError(
280 f"Google PSE rate limit/quota exceeded: {sanitized}"
281 )
283 last_exception = e
284 except Exception as e:
285 error_msg = str(e)
286 sanitized = self._sanitize_error_message(error_msg)
287 logger.warning(
288 "Error on attempt {} / {}: {}",
289 attempt + 1,
290 self.max_retries,
291 sanitized,
292 )
294 # Check for rate limiting patterns in general errors
295 if "quota" in error_msg.lower() or "limit" in error_msg.lower():
296 raise RateLimitError(
297 f"Google PSE error (possible rate limit): {sanitized}"
298 )
300 last_exception = e
302 attempt += 1
304 # If we get here, all retries failed
305 error_msg = f"Failed to get response from Google PSE API after {self.max_retries} attempts"
306 logger.error(error_msg)
308 if last_exception: 308 ↛ 312line 308 didn't jump to line 312 because the condition on line 308 was always true
309 raise RequestException(
310 f"{error_msg}: {self._sanitize_error_message(str(last_exception))}"
311 )
312 raise RequestException(error_msg)
314 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
315 """Get search result previews/snippets"""
316 results = []
318 # Google PSE API returns a maximum of 10 results per request
319 # We may need to make multiple requests to get the desired number
320 start_index = 1
321 total_results = 0
323 while total_results < self.max_results: 323 ↛ 368line 323 didn't jump to line 368 because the condition on line 323 was always true
324 try:
325 response = self._make_request(query, start_index)
327 # Break if no items
328 if "items" not in response:
329 break
331 items = response.get("items", [])
333 # Process each result
334 for item in items:
335 title = item.get("title", "")
336 snippet = item.get("snippet", "")
337 url = item.get("link", "")
339 # Skip results without URL
340 if not url:
341 continue
343 results.append(
344 {
345 "title": title,
346 "snippet": snippet,
347 "link": url,
348 "source": "Google Programmable Search",
349 }
350 )
352 total_results += 1
353 if total_results >= self.max_results:
354 break
356 # Check if there are more results
357 if not items or total_results >= self.max_results:
358 break
360 # Update start index for next request
361 start_index += len(items)
363 except Exception as e:
364 safe_msg = self._scrub_error(e)
365 logger.warning(f"Error getting search results: {safe_msg}")
366 break
368 logger.info(
369 "Retrieved {} search results for query: '{}'", len(results), query
370 )
371 return results