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.2, created at 2026-10-22 21:20 +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 Google PSE query. Waiting {} s",
232 attempt + 1,
233 self.max_retries,
234 f"{sleep_time:.2f}",
235 )
236 time.sleep(sleep_time)
238 # Make the request
239 logger.debug(
240 "Making request to Google PSE API (start_index={})",
241 start_index,
242 )
243 # Apply rate limiting before request
244 adaptive_wait = self.rate_tracker.apply_rate_limit(
245 self.engine_type
246 )
247 minimum_interval_wait = self._respect_rate_limit()
248 self._last_wait_time = adaptive_wait + minimum_interval_wait
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("Retrieved {} search results", len(results))
367 return results