Coverage for src/local_deep_research/web_search_engines/search_engine_base.py: 95%
399 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1import json
2import time
3from abc import ABC, abstractmethod
4from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union
5from urllib.parse import urlparse
7from langchain_core.language_models import BaseLLM
8from ..security.secure_logging import logger
9from tenacity import (
10 RetryError,
11 retry,
12 retry_if_exception_type,
13 stop_after_attempt,
14)
15from tenacity.wait import wait_base
17from ..advanced_search_system.filters.base_filter import BaseFilter
18from ..utilities.type_utils import unwrap_setting
19from ..config.constants import DEFAULT_MAX_FILTERED_RESULTS
20from ..config.thread_settings import get_setting_from_snapshot
21from ..security.log_sanitizer import sanitize_error_message, scrub_error
22from ..security.egress.classification import Exposure, Sensitivity
23from ..utilities.thread_context import clear_search_context, set_search_context
25from .rate_limiting import RateLimitError, get_tracker
26from ..constants import DEFAULT_SEARCH_TOOL
28if TYPE_CHECKING:
29 from ..advanced_search_system.filters.journal_reputation_filter import (
30 JournalReputationFilter,
31 )
34# Common placeholder values that should never be treated as a real API key.
35# Three call sites previously had inconsistent subsets of this list, which
36# let invalid placeholders silently through the production path in
37# search_engines_config.py. Centralizing here so all three stay in sync.
38API_KEY_PLACEHOLDERS = frozenset(
39 {
40 "",
41 "None",
42 "null",
43 "PLACEHOLDER",
44 "YOUR_API_KEY_HERE",
45 "YOUR_API_KEY",
46 "API_KEY",
47 "your_api_key",
48 "your-api-key",
49 }
50)
53def _is_api_key_placeholder(api_key: Optional[str]) -> bool:
54 """Return True if ``api_key`` looks like a placeholder, not a real key.
56 Catches:
57 - Exact-match placeholders (see ``API_KEY_PLACEHOLDERS``)
58 - Environment-variable-style names ending in ``_API_KEY``
59 (e.g. ``BRAVE_API_KEY``)
60 - Templates starting with ``YOUR_``
61 - Angle-bracket templates: ``<key>`` or ``${KEY}``
62 """
63 if not api_key:
64 return True
65 api_key = api_key.strip()
66 if api_key in API_KEY_PLACEHOLDERS:
67 return True
68 if api_key.endswith("_API_KEY"):
69 return True
70 if api_key.startswith("YOUR_"):
71 return True
72 if api_key.startswith("<") and api_key.endswith(">"):
73 return True
74 if api_key.startswith("${") and api_key.endswith("}"):
75 return True
76 return False
79class AdaptiveWait(wait_base):
80 """Custom wait strategy that uses adaptive rate limiting."""
82 def __init__(self, get_wait_func):
83 self.get_wait_func = get_wait_func
85 def __call__(self, retry_state):
86 return self.get_wait_func()
89class BaseSearchEngine(ABC):
90 """
91 Abstract base class for search engines with two-phase retrieval capability.
92 Handles common parameters and implements the two-phase search approach.
94 Subclass contract for ``__init__``
95 ---------------------------------
96 Concrete engines should forward ``settings_snapshot`` and
97 ``programmatic_mode`` to ``super().__init__`` so the base class can wire
98 them up correctly. The cleanest way is to declare ``**kwargs`` and pass
99 it along::
101 def __init__(self, max_results=10, *, my_param=None, **kwargs):
102 super().__init__(max_results=max_results, **kwargs)
103 self.my_param = my_param
105 Many existing engines accept ``**kwargs`` but don't forward — that
106 silently drops ``programmatic_mode`` (and used to bind the wrong rate
107 tracker). The factory has a post-construction safety net that calls
108 ``_configure_programmatic_mode`` when the engine's mode doesn't match
109 what the caller asked for, but new engines should not rely on it: the
110 safety net only covers ``programmatic_mode``, not ``settings_snapshot``
111 or other base kwargs.
112 """
114 # Class attribute to indicate if this engine searches public internet sources
115 # Should be overridden by subclasses - defaults to False for safety
116 is_public = False
118 # Class attribute to indicate if this is a generic search engine (vs specialized)
119 # Generic engines are general web search (Google, Bing, etc) vs specialized (arXiv, PubMed).
120 # Note: generic does NOT imply good native ranking — see is_lexical.
121 is_generic = False
123 # Class attribute to indicate if this is a scientific/academic search engine
124 # Scientific engines include arXiv, PubMed, Semantic Scholar, etc.
125 is_scientific = False
127 # Class attribute to indicate if this is a local RAG/document search engine
128 # Local engines search private document collections stored locally
129 is_local = False
131 # Egress data classification (ADR-0007). Two orthogonal axes, declared
132 # explicitly per engine alongside is_public/is_local. The defaults fail
133 # closed: an engine that forgets to classify itself is treated as the most
134 # restrictive quadrant (sensitive data + exposing sink → usable only by
135 # itself). Engines whose label depends on configuration (a collection's
136 # is_public flag, a self-hosted store's URL) are refined by the classifier
137 # at run time; these constants are the static fallback.
138 egress_sensitivity = Sensitivity.SENSITIVE
139 egress_exposure = Exposure.EXPOSING
141 # Class attribute to indicate if this is a news search engine
142 # News engines specialize in news articles and current events
143 is_news = False
145 # Class attribute to indicate if this is a code search engine
146 # Code engines specialize in searching code repositories
147 is_code = False
149 # Class attribute to indicate if this is a book/literature search engine
150 # Book engines search libraries and literary archives
151 is_books = False
153 # Classification: does this engine use lexical/keyword-based search?
154 # Lexical engines (arXiv, PubMed, Wikipedia, Mojeek, etc.) match results by
155 # keywords without ML-based ranking. This is an informational flag that can
156 # drive multiple behaviors (query optimization, result deduplication, UI hints).
157 # For LLM relevance filtering specifically, see needs_llm_relevance_filter.
158 is_lexical = False
160 # Behavioral: should the factory auto-enable LLM relevance filtering?
161 # When True, the factory sets enable_llm_relevance_filter=True on the engine
162 # instance, causing _filter_for_relevance() to run after previews are fetched.
163 # Typically set alongside is_lexical=True, but can be set independently —
164 # e.g. a non-lexical engine with noisy results could opt in.
165 needs_llm_relevance_filter = False
167 # Tuning for the LLM relevance filter (only applies when the filter
168 # is active for this engine).
169 #
170 # relevance_filter_batch_size: split previews into chunks of this many
171 # before sending to the LLM. Smaller batches are faster per call and
172 # more reliable on weaker models which struggle with many indices in
173 # one context. None or 0 = single-call mode (no batching).
174 #
175 # relevance_filter_max_parallel_batches: number of batches to dispatch
176 # concurrently against the LLM. 1 = sequential. Most providers handle
177 # parallel requests fine (Ollama with OLLAMA_NUM_PARALLEL>1, OpenAI,
178 # Anthropic).
179 relevance_filter_batch_size: Optional[int] = 5
180 relevance_filter_max_parallel_batches: int = 10
182 # Class attribute for rate limit detection patterns
183 # Subclasses can override to add engine-specific patterns
184 rate_limit_patterns: Set[str] = {
185 "rate limit",
186 "rate_limit",
187 "ratelimit",
188 "too many requests",
189 "throttl",
190 "quota exceeded",
191 "quota_exceeded",
192 "limit exceeded",
193 "request limit",
194 "api limit",
195 "usage limit",
196 }
198 # Instance-attribute names holding credential values to redact from error
199 # messages/logs via _scrub_error(). Subclasses override when they store
200 # secrets under different names (e.g. Elasticsearch: _api_key/_password).
201 # Centralizing the list keeps every dual-scrub call site uniform and
202 # prevents the per-site drift that previously dropped a secret.
203 _secret_attrs: tuple[str, ...] = ("api_key",)
205 @staticmethod
206 def _ensure_list(value, *, default=None):
207 """Normalize a value that should be a list.
209 Handles JSON-encoded strings, comma-separated strings, and
210 already-parsed lists. Returns *default* (empty list when not
211 supplied) for ``None`` or empty/unparseable input.
212 """
213 if default is None:
214 default = []
215 if value is None:
216 return default
217 if isinstance(value, list):
218 return value
219 if isinstance(value, str):
220 stripped = value.strip()
221 if not stripped:
222 return default
223 if stripped.startswith("["):
224 try:
225 parsed = json.loads(stripped)
226 if isinstance(parsed, list): 226 ↛ 230line 226 didn't jump to line 230 because the condition on line 226 was always true
227 return [str(item) for item in parsed]
228 except (json.JSONDecodeError, ValueError, RecursionError):
229 pass
230 return [
231 item.strip() for item in stripped.split(",") if item.strip()
232 ]
233 return default
235 @classmethod
236 def _load_engine_class(cls, name: str, config: Dict[str, Any]):
237 """
238 Helper method to load an engine class dynamically.
240 Args:
241 name: Engine name
242 config: Engine configuration dict with module_path and class_name
244 Returns:
245 Tuple of (success: bool, engine_class or None, error_msg or None)
246 """
247 from ..security.module_whitelist import (
248 ModuleNotAllowedError,
249 get_safe_module_class,
250 )
252 try:
253 module_path = config.get("module_path")
254 class_name = config.get("class_name")
256 if not module_path or not class_name:
257 return (
258 False,
259 None,
260 f"Missing module_path or class_name for {name}",
261 )
263 # Use whitelist-validated safe import
264 engine_class = get_safe_module_class(module_path, class_name)
266 return True, engine_class, None
268 except ModuleNotAllowedError as e:
269 return (
270 False,
271 None,
272 f"Security error loading engine class for {name}: {e}",
273 )
274 except Exception as e:
275 return False, None, f"Could not load engine class for {name}: {e}"
277 def __init__(
278 self,
279 llm: Optional[BaseLLM] = None,
280 max_filtered_results: Optional[int] = None,
281 max_results: Optional[int] = 10, # Default value if not provided
282 preview_filters: List[BaseFilter] | None = None,
283 content_filters: List[BaseFilter] | None = None,
284 search_snippets_only: bool = True, # New parameter with default
285 include_full_content: bool = False,
286 settings_snapshot: Optional[Dict[str, Any]] = None,
287 programmatic_mode: bool = False,
288 **kwargs,
289 ):
290 """
291 Initialize the search engine with common parameters.
293 Args:
294 llm: Optional language model for relevance filtering
295 max_filtered_results: Maximum number of results to keep after filtering
296 max_results: Maximum number of search results to return
297 preview_filters: Filters that will be applied to all previews
298 produced by the search engine, before relevancy checks.
299 content_filters: Filters that will be applied to the full content
300 produced by the search engine, after relevancy checks.
301 search_snippets_only: Whether to return only snippets or full content
302 include_full_content: Whether to use FullSearchResults for full webpage content
303 settings_snapshot: Settings snapshot for configuration
304 programmatic_mode: If True, disables database operations and uses memory-only tracking
305 **kwargs: Additional engine-specific parameters
306 """
307 if max_filtered_results is None:
308 max_filtered_results = DEFAULT_MAX_FILTERED_RESULTS
309 if max_results is None:
310 max_results = 10
311 self._preview_filters: List[BaseFilter] = (
312 preview_filters if preview_filters is not None else []
313 )
314 self._content_filters: List[BaseFilter] = (
315 content_filters if content_filters is not None else []
316 )
318 self.llm = llm # LLM for relevance filtering
319 self._max_filtered_results = int(
320 max_filtered_results
321 ) # Ensure it's an integer
322 self._max_results = max(
323 1, int(max_results)
324 ) # Ensure it's a positive integer
325 self.search_snippets_only = search_snippets_only # Store the setting
326 self.include_full_content = include_full_content
327 self.settings_snapshot = (
328 settings_snapshot or {}
329 ) # Store settings snapshot
331 self.engine_type = self.__class__.__name__
332 self._engine_name: str = "" # set by the factory after construction
333 # The snapshot dict the runtime egress backstop last verified
334 # against (identity comparison; holding the reference keeps the
335 # memo valid — a verification re-runs when the caller assigns a
336 # NEW snapshot) plus the policy-relevant VALUES it carried at
337 # verification time, so an in-place mutation of the scope or
338 # primary key on the same dict is also caught.
339 self._egress_verified_snapshot: Optional[Dict[str, Any]] = None
340 self._egress_verified_policy_key: Optional[tuple] = None
341 self._egress_skip_warned = False
342 self._configure_programmatic_mode(programmatic_mode)
343 self._last_wait_time = (
344 0.0 # Default to 0 for successful searches without rate limiting
345 )
346 self._last_results_count = 0
348 def _create_journal_filter(
349 self,
350 engine_name: str,
351 llm: Optional[BaseLLM],
352 settings_snapshot: Optional[Dict[str, Any]],
353 ) -> Optional["JournalReputationFilter"]:
354 """Build the default :class:`JournalReputationFilter` for this engine.
356 Wraps the identical 8-line ``JournalReputationFilter.create_default``
357 boilerplate that previously lived in every academic subclass. The
358 ``engine_name`` is passed explicitly because the auto-derived class
359 name does not match the settings key for ``nasa_ads`` (would be
360 ``"nasaads"``) or ``semantic_scholar`` (would be
361 ``"semanticscholar"``).
363 ``llm`` and ``settings_snapshot`` are passed in rather than read from
364 ``self`` because subclasses build their preview filters *before*
365 calling ``super().__init__()`` (the filter is handed to the parent
366 constructor), so ``self.llm`` / ``self.settings_snapshot`` do not
367 exist yet at call time.
369 Args:
370 engine_name: Settings key identifying the engine (e.g.
371 ``"arxiv"``, ``"nasa_ads"``).
372 llm: Language model used for the filter's relevance pass.
373 settings_snapshot: Settings snapshot for configuration lookups.
375 Returns:
376 A configured :class:`JournalReputationFilter`, or ``None`` if
377 filtering is disabled in settings for this engine.
378 """
379 from ..advanced_search_system.filters.journal_reputation_filter import (
380 JournalReputationFilter,
381 )
383 return JournalReputationFilter.create_default(
384 model=llm, # type: ignore[arg-type]
385 engine_name=engine_name,
386 settings_snapshot=settings_snapshot,
387 )
389 def _configure_programmatic_mode(self, programmatic_mode: bool) -> None:
390 """Set ``programmatic_mode`` and (re)bind the matching rate tracker.
392 Called from ``__init__`` and from the factory as a fallback when an
393 engine subclass swallows the ``programmatic_mode`` kwarg without
394 forwarding it to ``super().__init__``. Safe to call after init —
395 rebinding ``rate_tracker`` discards the previous tracker (no
396 resources to release) and the new one starts with empty in-memory
397 caches.
398 """
399 self.programmatic_mode = programmatic_mode
400 if programmatic_mode:
401 from .rate_limiting.tracker import AdaptiveRateLimitTracker
403 self.rate_tracker = AdaptiveRateLimitTracker(
404 settings_snapshot=self.settings_snapshot,
405 programmatic_mode=programmatic_mode,
406 )
407 else:
408 self.rate_tracker = get_tracker()
410 @property
411 def max_filtered_results(self) -> int:
412 """Get the maximum number of filtered results."""
413 return self._max_filtered_results
415 @max_filtered_results.setter
416 def max_filtered_results(self, value: int) -> None:
417 """Set the maximum number of filtered results."""
418 if value is None:
419 value = DEFAULT_MAX_FILTERED_RESULTS
420 logger.warning(
421 f"Setting max_filtered_results to {DEFAULT_MAX_FILTERED_RESULTS}"
422 )
423 self._max_filtered_results = int(value)
425 @property
426 def max_results(self) -> int:
427 """Get the maximum number of search results."""
428 return self._max_results
430 @max_results.setter
431 def max_results(self, value: int) -> None:
432 """Set the maximum number of search results."""
433 if value is None:
434 value = 10
435 self._max_results = max(1, int(value))
437 def _get_adaptive_wait(self) -> float:
438 """Get adaptive wait time from tracker."""
439 wait_time = self.rate_tracker.get_wait_time(self.engine_type)
440 self._last_wait_time = wait_time
441 logger.debug(
442 f"{self.engine_type} waiting {wait_time:.2f}s before retry"
443 )
444 return wait_time
446 def _record_retry_outcome(self, retry_state) -> None:
447 """Record outcome after retry completes."""
448 success = (
449 not retry_state.outcome.failed if retry_state.outcome else False
450 )
451 self.rate_tracker.record_outcome(
452 self.engine_type,
453 self._last_wait_time or 0,
454 success,
455 retry_state.attempt_number,
456 error_type="RateLimitError" if not success else None,
457 search_result_count=self._last_results_count if success else 0,
458 )
460 def _verify_egress_scope(self) -> None:
461 """Runtime backstop: verify this engine is allowed under the egress
462 scope in ``self.settings_snapshot`` before executing a search.
464 No-op when ``settings_snapshot`` is missing or empty (programmatic
465 API callers) or when ``_engine_name`` has not been set. Raises
466 ``PolicyDeniedError`` when the policy denies the engine; any other
467 internal error in the policy evaluation is logged and ignored so a
468 broken backstop never takes down searches the factory PEP already
469 approved.
471 This is a defense-in-depth check behind the factory PEP, the
472 strategy-level filters, and the audit hook. It covers engines that
473 bypassed the factory (direct instantiation with a snapshot). Note
474 the limit: the snapshot is the one captured at construction, so a
475 scope change AFTER construction is only caught if the caller also
476 refreshes ``settings_snapshot`` (assigns a NEW dict — in-place
477 mutation of the existing dict is not detected, see below).
479 The verification is memoized per snapshot IDENTITY plus the
480 policy-relevant VALUES (scope, primary): the full policy inputs
481 are stable for a given snapshot object, so re-evaluating on every
482 ``run()`` would only repeat the same decision — and under
483 ADAPTIVE scope with a URL-configurable primary the evaluation can
484 include a DNS lookup (bounded at 2s), which must not become a
485 per-search stall. Assigning a refreshed snapshot OR mutating the
486 scope/primary keys in place invalidates the memo and re-verifies.
487 Denials are never memoized (they raise).
488 """
489 if not self.settings_snapshot:
490 return
491 if not self._engine_name:
492 # A snapshot WITHOUT a stamped engine name means a direct
493 # instantiation the factory never saw and no subclass
494 # self-stamps — the backstop cannot evaluate it. Surface the
495 # gap in the audit log (once per instance) instead of
496 # skipping silently.
497 if not self._egress_skip_warned: 497 ↛ 504line 497 didn't jump to line 504 because the condition on line 497 was always true
498 self._egress_skip_warned = True
499 logger.bind(policy_audit=True).warning(
500 "egress backstop skipped: engine has a settings "
501 "snapshot but no _engine_name stamped",
502 engine_type=self.__class__.__name__,
503 )
504 return
505 if (
506 self.settings_snapshot is self._egress_verified_snapshot
507 and self._egress_policy_key() == self._egress_verified_policy_key
508 ):
509 return
510 from ..security.egress.policy import PolicyDeniedError
512 try:
513 self._check_egress_policy()
514 except PolicyDeniedError:
515 logger.bind(policy_audit=True).warning(
516 "runtime egress backstop denied engine",
517 engine=self._engine_name,
518 )
519 raise
520 except Exception:
521 logger.bind(policy_audit=True).debug(
522 "egress runtime verify errored; "
523 "factory PEP still enforces at instantiation",
524 engine=self._engine_name,
525 )
526 else:
527 self._egress_verified_snapshot = self.settings_snapshot
528 self._egress_verified_policy_key = self._egress_policy_key()
530 def _egress_policy_key(self) -> tuple:
531 """The policy-relevant snapshot values the backstop memo guards on.
533 Cheap dict reads only — no context construction, no DNS. Not an
534 exhaustive input set (URL settings or DB-side collection flags can
535 also influence the decision), but it covers the keys an in-place
536 mutation would realistically target; the factory PEP remains the
537 primary enforcement for everything else.
539 MAINTENANCE: if a new snapshot key ever becomes policy-relevant in
540 ``context_from_snapshot``/``evaluate_engine``, add it here too —
541 otherwise an in-place mutation of that key returns a stale memo.
543 """
544 scope = unwrap_setting(
545 self.settings_snapshot.get("policy.egress_scope")
546 )
547 primary = unwrap_setting(self.settings_snapshot.get("search.tool"))
548 return (scope, primary)
550 def _check_egress_policy(self) -> None:
551 """Inner helper so the raise is not inside the except handler
552 (ruff TRY301). Raises PolicyDeniedError on scope mismatch."""
553 from ..security.egress.policy import (
554 PolicyDeniedError,
555 context_from_snapshot,
556 evaluate_engine,
557 )
559 primary = unwrap_setting(
560 self.settings_snapshot.get("search.tool", self._engine_name)
561 )
562 ctx = context_from_snapshot(
563 self.settings_snapshot,
564 primary or self._engine_name,
565 username=self.settings_snapshot.get("_username"),
566 )
567 decision = evaluate_engine(
568 self._engine_name,
569 ctx,
570 settings_snapshot=self.settings_snapshot,
571 )
572 if not decision.allowed:
573 raise PolicyDeniedError(decision, target=self._engine_name)
575 def run(
576 self, query: str, research_context: Dict[str, Any] | None = None
577 ) -> List[Dict[str, Any]]:
578 """
579 Run the search engine with a given query, retrieving and filtering results.
580 This implements a two-phase retrieval approach:
581 1. Get preview information for many results
582 2. Filter the previews for relevance
583 3. Get full content for only the relevant results
585 Args:
586 query: The search query
587 research_context: Context from previous research to use.
589 Returns:
590 List of search results with full content (if available)
591 """
592 logger.info(f"---Execute a search using {self.__class__.__name__}---")
594 # Runtime egress scope backstop: verify this engine is allowed
595 # before making any HTTP requests.
596 self._verify_egress_scope()
598 # Track search call for metrics (if available and not in programmatic mode)
599 should_record_metrics = False
600 context_was_set = False
601 if not self.programmatic_mode:
602 from ..metrics.search_tracker import SearchTracker
604 should_record_metrics = True
606 # For thread-safe context propagation: if we have research_context parameter, use it
607 # Otherwise, try to inherit from current thread context (normal case)
608 # This allows strategies running in threads to explicitly pass context when needed
609 if research_context:
610 # Explicit context provided - use it and set it for this thread
611 set_search_context(research_context)
612 context_was_set = True
614 engine_name = self.__class__.__name__.replace(
615 "SearchEngine", ""
616 ).lower()
617 start_time = time.time()
619 success = True
620 error_message = None
621 results_count = 0
623 # Define the core search function with retry logic
624 if self.rate_tracker.enabled:
625 # Rate limiting enabled - use retry with adaptive wait
626 @retry(
627 stop=stop_after_attempt(3),
628 wait=AdaptiveWait(lambda: self._get_adaptive_wait()),
629 retry=retry_if_exception_type((RateLimitError,)),
630 after=self._record_retry_outcome,
631 reraise=True,
632 )
633 def _run_with_retry():
634 nonlocal success, error_message, results_count
635 return _execute_search()
636 else:
637 # Rate limiting disabled - run without retry
638 def _run_with_retry():
639 nonlocal success, error_message, results_count
640 return _execute_search()
642 def _execute_search():
643 nonlocal success, error_message, results_count
645 try:
646 # Step 1: Get preview information for items
647 previews = self._get_previews(query)
648 if not previews:
649 logger.info(
650 f"Search engine {self.__class__.__name__} returned no preview results for query: {query}"
651 )
652 results_count = 0
653 return []
655 # Pre-filter enrichment: resolve DOIs to OpenAlex source
656 # IDs BEFORE the preview filters run. The
657 # JournalReputationFilter is registered as a preview
658 # filter and uses ``result["openalex_source_id"]`` for
659 # Tier 2 lookups; running enrichment afterwards (as the
660 # old pipeline did) left the field empty at filter time,
661 # silently forcing a fragile-name-match fallback. Only
662 # for scientific engines whose results carry DOIs.
663 if getattr(self, "is_scientific", False):
664 try:
665 from ..utilities.openalex_enrichment import (
666 enrich_results_with_source_ids,
667 )
669 email = getattr(self, "email", None)
670 previews = enrich_results_with_source_ids(
671 previews, email=email
672 )
673 except Exception:
674 logger.debug(
675 "DOI enrichment skipped (import or call failed)"
676 )
678 for preview_filter in self._preview_filters:
679 previews = preview_filter.filter_results(previews, query)
681 # Step 2: Filter previews for relevance with LLM
682 enable_llm_filter = getattr(
683 self, "enable_llm_relevance_filter", False
684 )
686 if enable_llm_filter and self.llm:
687 filtered_items = self._filter_for_relevance(previews, query)
688 else:
689 filtered_items = previews
690 logger.debug(
691 f"[{type(self).__name__}] Relevance filter skipped "
692 f"(enabled={enable_llm_filter}, "
693 f"llm={'yes' if self.llm else 'no'})"
694 )
696 # Step 3: Get full content for filtered items
697 if self.search_snippets_only:
698 logger.info("Returning snippet-only results as per config")
699 results = filtered_items
700 else:
701 results = self._get_full_content(filtered_items)
703 for content_filter in self._content_filters:
704 results = content_filter.filter_results(results, query)
706 results_count = len(results)
707 self._last_results_count = results_count
709 # Record success if we get here and rate limiting is enabled
710 if self.rate_tracker.enabled:
711 logger.info(
712 f"Recording successful search for {self.engine_type}: wait_time={self._last_wait_time}s, results={results_count}"
713 )
714 self.rate_tracker.record_outcome(
715 self.engine_type,
716 self._last_wait_time,
717 success=True,
718 retry_count=1, # First attempt succeeded
719 search_result_count=results_count,
720 )
721 else:
722 logger.info(
723 f"Rate limiting disabled, not recording search for {self.engine_type}"
724 )
726 return results
728 except RateLimitError:
729 # Only re-raise if rate limiting is enabled
730 if self.rate_tracker.enabled:
731 raise
732 # If rate limiting is disabled, treat as regular error
733 success = False
734 error_message = "Rate limit hit but rate limiting disabled"
735 logger.warning(
736 f"Rate limit hit on {self.__class__.__name__} but rate limiting is disabled"
737 )
738 results_count = 0
739 return []
740 except Exception as e:
741 # Other errors - don't retry
742 success = False
743 # Sanitize before it flows to SearchTracker.record_search
744 # (and the database) and to the log. Use logger.warning with
745 # the dual-scrubbed text instead of logger.exception: the
746 # cause chain frequently carries the request URL or auth
747 # header from upstream HTTP clients (see #4131).
748 error_message = safe_msg = self._scrub_error(e)
749 logger.warning(
750 f"Search engine {self.__class__.__name__} failed: {safe_msg}"
751 )
752 results_count = 0
753 return []
755 try:
756 return _run_with_retry() # type: ignore[no-any-return]
757 except RetryError as e:
758 # All retries exhausted
759 success = False
760 error_message = self._scrub_error(
761 f"Rate limited after all retries: {e}"
762 )
763 safe_msg = self._scrub_error(e)
764 logger.warning(
765 f"{self.__class__.__name__} failed after all retries: {safe_msg}"
766 )
767 return []
768 except Exception as e:
769 success = False
770 error_message = safe_msg = self._scrub_error(e)
771 logger.warning(
772 f"Search engine {self.__class__.__name__} error: {safe_msg}"
773 )
774 return []
775 finally:
776 try:
777 # Record search metrics BEFORE clearing context (record_search needs it)
778 if should_record_metrics:
779 response_time_ms = int((time.time() - start_time) * 1000)
780 SearchTracker.record_search(
781 engine_name=engine_name,
782 query=query,
783 results_count=results_count,
784 response_time_ms=response_time_ms,
785 success=success,
786 error_message=error_message,
787 )
788 finally:
789 # Clean up temporary search result storage
790 for attr in self._temp_attributes():
791 if hasattr(self, attr):
792 delattr(self, attr)
793 # ALWAYS clean up search context, even if recording fails
794 if context_was_set:
795 clear_search_context()
797 def invoke(self, query: str) -> List[Dict[str, Any]]:
798 """Compatibility method for LangChain tools"""
799 return self.run(query)
801 def _filter_for_relevance(
802 self, previews: List[Dict[str, Any]], query: str
803 ) -> List[Dict[str, Any]]:
804 """
805 Filter search results by relevance using the LLM.
807 Delegates to the ``relevance_filter`` module, which prompts the
808 LLM for a plain-text list of relevant indices and parses them
809 with a regex (no structured output).
811 Args:
812 previews: List of preview dictionaries
813 query: The original search query
815 Returns:
816 Filtered list of preview dictionaries
817 """
818 engine_name = type(self).__name__
820 if not self.llm or len(previews) <= 1:
821 logger.debug(
822 f"[{engine_name}] Skipping relevance filter "
823 f"(llm={'yes' if self.llm else 'no'}, "
824 f"previews={len(previews)})"
825 )
826 return previews
828 from .relevance_filter import filter_previews_for_relevance
830 return filter_previews_for_relevance(
831 llm=self.llm,
832 previews=previews,
833 query=query,
834 max_filtered_results=self.max_filtered_results,
835 engine_name=engine_name,
836 batch_size=self.relevance_filter_batch_size,
837 max_parallel_batches=self.relevance_filter_max_parallel_batches,
838 )
840 # =========================================================================
841 # Shared Helper Methods for Subclasses
842 # =========================================================================
844 @staticmethod
845 def _is_valid_api_key(api_key: Optional[str]) -> bool:
846 """
847 Check if an API key is valid (not a placeholder value).
849 Args:
850 api_key: The API key to validate
852 Returns:
853 True if the key appears to be a real API key, False if it's a placeholder
855 Example:
856 >>> BaseSearchEngine._is_valid_api_key("sk-abc123")
857 True
858 >>> BaseSearchEngine._is_valid_api_key("YOUR_API_KEY_HERE")
859 False
860 """
861 return not _is_api_key_placeholder(api_key)
863 @staticmethod
864 def _extract_display_link(url: str, fallback: str = "") -> str:
865 """Extract the netloc (domain) from a URL for display purposes.
867 Args:
868 url: The URL to extract from
869 fallback: Value to return on parse failure (default: empty string)
871 Returns:
872 The netloc portion of the URL, or *fallback* if parsing fails.
873 """
874 if not url:
875 return fallback
876 try:
877 parsed = urlparse(url)
878 return parsed.netloc or fallback
879 except Exception:
880 return fallback
882 @staticmethod
883 def _clean_result_url(value: Any) -> str:
884 """Normalize a raw search-result URL for the validity gate.
886 Coerces ``None`` (and any other falsy value) to ``""`` and any
887 truthy value to ``str`` before stripping surrounding whitespace.
889 The strip is **load-bearing**, not merely cosmetic: several engines
890 gate the URL on a ``url.lower().startswith(("http://", "https://"))``
891 prefix check (see ``_is_valid_search_result``) *before* it reaches
892 the SSRF validator's own internal strip, so leading/trailing
893 whitespace — common in HTML-scraped ``href`` attributes — would
894 silently drop an otherwise-valid result. Cleaning once at extraction
895 also keeps the URL tidy for logs and downstream consumers (preview
896 ``id``/``link`` fields, etc.).
898 Args:
899 value: The raw URL value from a search result, e.g.
900 ``result.get("url")`` or a parsed HTML ``href`` attribute.
902 Returns:
903 The whitespace-stripped URL string, or ``""`` if *value* is
904 falsy (``None``, empty, etc.).
905 """
906 if not value:
907 return ""
908 return str(value).strip()
910 def _resolve_api_key(
911 self,
912 api_key: Optional[str],
913 setting_key: str,
914 engine_name: str = "search engine",
915 settings_snapshot: Optional[Dict[str, Any]] = None,
916 ) -> str:
917 """
918 Resolve an API key from multiple sources with priority order.
920 Environment variables are handled automatically by SettingsManager
921 when building the settings snapshot, so they don't need to be
922 checked separately here.
924 Priority order:
925 1. Direct parameter (api_key argument)
926 2. Settings snapshot (via setting_key)
928 Args:
929 api_key: API key passed directly as parameter
930 setting_key: Key to look up in settings snapshot (e.g., "search.brave_api_key")
931 engine_name: Human-readable engine name for error messages
932 settings_snapshot: Optional settings snapshot dict (uses self.settings_snapshot if not provided)
934 Returns:
935 The resolved API key string
937 Raises:
938 ValueError: If no valid API key is found from any source
940 Example:
941 >>> engine._resolve_api_key(
942 ... api_key=None,
943 ... setting_key="search.brave_api_key",
944 ... engine_name="Brave Search"
945 ... )
946 "sk-abc123..."
947 """
948 # Use instance settings snapshot if not provided
949 if settings_snapshot is None:
950 settings_snapshot = self.settings_snapshot
952 # Priority 1: Direct parameter
953 if self._is_valid_api_key(api_key) and api_key is not None:
954 return api_key.strip()
956 # Priority 2: Settings snapshot (includes env var overrides via SettingsManager)
957 if settings_snapshot:
958 settings_value = get_setting_from_snapshot(
959 setting_key,
960 default=None,
961 settings_snapshot=settings_snapshot,
962 )
963 if self._is_valid_api_key(settings_value):
964 return settings_value.strip() if settings_value else ""
966 # No valid API key found
967 masked_key = self._mask_api_key(str(api_key)) if api_key else "None"
968 raise ValueError(
969 f"No valid API key found for {engine_name}. "
970 f"Checked: direct parameter ({masked_key}), "
971 f"settings key '{setting_key}'. "
972 f"Please provide a valid API key."
973 )
975 def _is_rate_limit_error(
976 self,
977 error: Union[Exception, str, int],
978 additional_patterns: Optional[Set[str]] = None,
979 ) -> bool:
980 """
981 Detect if an error is a rate limit error.
983 Checks multiple sources for rate limit indicators:
984 - HTTP status code 429
985 - HTTPError response objects
986 - Error messages containing rate limit phrases
988 Args:
989 error: The error to check (Exception, string, or HTTP status code)
990 additional_patterns: Optional set of additional patterns to match
992 Returns:
993 True if the error appears to be a rate limit error
995 Example:
996 >>> engine._is_rate_limit_error(429)
997 True
998 >>> engine._is_rate_limit_error("Rate limit exceeded")
999 True
1000 >>> engine._is_rate_limit_error(ValueError("Invalid input"))
1001 False
1002 """
1003 # Combine default and additional patterns
1004 patterns = self.rate_limit_patterns.copy()
1005 if additional_patterns:
1006 patterns.update(additional_patterns)
1008 # Check integer status code directly
1009 if isinstance(error, int):
1010 return error == 429
1012 # Convert to string for pattern matching
1013 error_str = ""
1014 status_code = None
1016 if isinstance(error, str):
1017 error_str = error
1018 elif isinstance(error, Exception):
1019 error_str = str(error)
1021 # Check for HTTP status code in common HTTP error types
1022 if hasattr(error, "status_code"):
1023 status_code = error.status_code
1024 elif hasattr(error, "response"):
1025 response = error.response
1026 if hasattr(response, "status_code"):
1027 status_code = response.status_code
1029 # Check status code first
1030 if status_code == 429:
1031 return True
1033 # Case-insensitive pattern matching
1034 error_lower = error_str.lower()
1035 for pattern in patterns:
1036 if pattern.lower() in error_lower:
1037 return True
1039 return False
1041 def _raise_if_rate_limit(
1042 self,
1043 error: Union[Exception, str, int],
1044 additional_patterns: Optional[Set[str]] = None,
1045 ) -> None:
1046 """
1047 Raise RateLimitError if the given error is a rate limit error.
1049 Convenience method that combines _is_rate_limit_error check with
1050 raising RateLimitError.
1052 Args:
1053 error: The error to check
1054 additional_patterns: Optional set of additional patterns to match
1056 Raises:
1057 RateLimitError: If the error is detected as a rate limit error
1059 Example:
1060 >>> try:
1061 ... response = make_api_call()
1062 ... except Exception as e:
1063 ... engine._raise_if_rate_limit(e)
1064 """
1065 if self._is_rate_limit_error(error, additional_patterns):
1066 error_msg = str(error) if not isinstance(error, str) else error
1067 raise RateLimitError(
1068 f"Rate limit detected: {self._sanitize_error_message(error_msg)}"
1069 )
1071 def _extract_full_result(self, item: Dict[str, Any]) -> Dict[str, Any]:
1072 """
1073 Extract the full result from an item that may contain a _full_result key.
1075 This is a helper for the default _get_full_content implementation.
1076 It extracts data from the _full_result key if present, otherwise uses
1077 the item directly, and removes the internal _full_result key.
1079 Args:
1080 item: A search result item that may contain a _full_result key
1082 Returns:
1083 A dictionary with the full result data, without the _full_result key
1085 Example:
1086 >>> engine._extract_full_result({"title": "A", "_full_result": {"title": "A", "content": "Full"}})
1087 {"title": "A", "content": "Full"}
1088 """
1089 source = item.get("_full_result")
1090 if source is None:
1091 source = item
1092 return {k: v for k, v in source.items() if k != "_full_result"}
1094 def _get_full_content(
1095 self, relevant_items: List[Dict[str, Any]]
1096 ) -> List[Dict[str, Any]]:
1097 """
1098 Get full content for the relevant items.
1100 Default implementation extracts data from _full_result keys if present.
1101 Subclasses can override this method to fetch additional content from
1102 external sources (e.g., web scraping, API calls).
1104 Args:
1105 relevant_items: List of relevant preview dictionaries
1107 Returns:
1108 List of result dictionaries with full content
1110 Example:
1111 >>> engine._get_full_content([
1112 ... {"title": "A", "_full_result": {"title": "A", "content": "Full A"}},
1113 ... {"title": "B"}
1114 ... ])
1115 [{"title": "A", "content": "Full A"}, {"title": "B"}]
1116 """
1117 if not relevant_items:
1118 return []
1119 return [self._extract_full_result(item) for item in relevant_items]
1121 def _build_full_search_egress_context(self):
1122 """Build an EgressContext from self.settings_snapshot for the
1123 per-URL scope gate inside FullSearchResults. Returns None when a
1124 context can't be built and disables full-content fetching when
1125 policy evaluation fails on a supplied snapshot.
1127 The previous "fail closed by returning None" comment was wrong:
1128 full_search.py treats ``egress_context is None`` as
1129 ``evaluate_url_fn = None`` and skips the per-URL scope check
1130 entirely, leaving only SSRF — which allows public hosts under
1131 PRIVATE_ONLY. When the snapshot was supplied but the policy is
1132 unevaluable, we now turn full-content fetching off for this
1133 engine instead.
1134 """
1135 if not getattr(self, "settings_snapshot", None):
1136 return None
1137 try:
1138 from ..security.egress.policy import (
1139 PolicyDeniedError,
1140 context_from_snapshot,
1141 )
1142 except ImportError:
1143 logger.debug(
1144 "egress_policy unavailable in search_engine_base; "
1145 "FullSearchResults fetches will be SSRF-only"
1146 )
1147 return None
1149 primary_raw = unwrap_setting(
1150 self.settings_snapshot.get("search.tool", DEFAULT_SEARCH_TOOL)
1151 )
1152 try:
1153 return context_from_snapshot(
1154 self.settings_snapshot, primary_raw or DEFAULT_SEARCH_TOOL
1155 )
1156 except (PolicyDeniedError, ValueError) as exc:
1157 safe_msg = self._scrub_error(exc)
1158 logger.bind(policy_audit=True).warning(
1159 "Disabling include_full_content: egress policy could "
1160 "not be evaluated for this engine",
1161 reason=safe_msg,
1162 )
1163 self.include_full_content = False
1164 return None
1166 def _init_full_search(
1167 self,
1168 web_search=None,
1169 language="en",
1170 max_results=10,
1171 region=None,
1172 time_period=None,
1173 safe_search=None,
1174 ):
1175 """Initialize FullSearchResults if include_full_content is True.
1177 Call this at the end of your __init__ after setting up your search wrapper.
1179 Args:
1180 web_search: The search wrapper/engine to pass to FullSearchResults
1181 language: Language for search results
1182 max_results: Maximum number of results
1183 region: Region/country code for results
1184 time_period: Time period filter
1185 safe_search: Safe search setting (string value for FullSearchResults)
1186 """
1187 if self.include_full_content and self.llm:
1188 try:
1189 from .engines.full_search import FullSearchResults
1191 # Egress context for per-URL scope gating. Built once
1192 # here (not per fetch) so the policy snapshot is the
1193 # same one used by the factory PEP that authorized this
1194 # engine.
1195 egress_context = self._build_full_search_egress_context()
1196 self.full_search = FullSearchResults(
1197 llm=self.llm,
1198 web_search=web_search,
1199 language=language,
1200 max_results=max_results,
1201 region=region,
1202 time=time_period,
1203 safesearch=safe_search,
1204 settings_snapshot=self.settings_snapshot,
1205 egress_context=egress_context,
1206 )
1207 except ImportError:
1208 logger.warning(
1209 "FullSearchResults not available. "
1210 "Full content retrieval disabled."
1211 )
1212 self.include_full_content = False
1214 def _temp_attributes(self):
1215 """Return list of temporary attribute names to clean up after run().
1217 Override in subclasses that store additional temporary data.
1218 """
1219 return ["_search_results"]
1221 def _sanitize_error_message(self, message: str) -> str:
1222 """
1223 Remove/mask API keys, tokens, and secrets from error messages.
1225 Uses pattern matching for common credential formats.
1227 Args:
1228 message: The error message to sanitize
1230 Returns:
1231 Sanitized message with sensitive data redacted
1233 Example:
1234 >>> engine._sanitize_error_message(
1235 ... "Error with key sk-abc123def456ghi789jkl012"
1236 ... )
1237 "Error with key [REDACTED_KEY]"
1238 """
1239 return sanitize_error_message(message)
1241 def _scrub_error(self, error: Union[BaseException, str]) -> str:
1242 """Return a log/DB-safe rendering of *error*.
1244 Delegates the "dual-scrub" to :func:`~..security.log_sanitizer.scrub_error`,
1245 resolving this engine's known literal secret values from
1246 ``_secret_attrs``. *error* may be an exception or a pre-built
1247 message string.
1249 Use this at every catch site that logs or persists an exception so
1250 the two scrub passes can never drift apart per-engine.
1251 """
1252 return scrub_error(
1253 error, *(getattr(self, name, None) for name in self._secret_attrs)
1254 )
1256 def _mask_api_key(self, api_key: str, visible_chars: int = 4) -> str:
1257 """
1258 Mask an API key for safe logging, showing only first and last characters.
1260 Args:
1261 api_key: The API key to mask
1262 visible_chars: Number of characters to show at start and end
1264 Returns:
1265 Masked API key in format "sk-1...nop" or "***" for short keys
1267 Example:
1268 >>> engine._mask_api_key("sk-abcdefghijklmnop123456")
1269 "sk-a...3456"
1270 >>> engine._mask_api_key("short")
1271 "***"
1272 """
1273 if not api_key:
1274 return "***"
1276 api_key = str(api_key).strip()
1278 if len(api_key) <= visible_chars * 2:
1279 return "***"
1281 return f"{api_key[:visible_chars]}...{api_key[-visible_chars:]}"
1283 @abstractmethod
1284 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
1285 """
1286 Get preview information (titles, summaries) for initial search results.
1288 Args:
1289 query: The search query
1291 Returns:
1292 List of preview dictionaries with at least 'id', 'title', and 'snippet' keys
1293 """
1294 pass
1296 def close(self) -> None:
1297 """
1298 Close any resources held by this search engine.
1300 Subclasses with HTTP sessions or other resources should override this.
1301 The base implementation safely closes any 'session' attribute if present
1302 and closes both preview and content filters that hold resources.
1303 """
1304 from ..utilities.resource_utils import safe_close
1306 if hasattr(self, "session") and self.session is not None:
1307 safe_close(self.session, "HTTP session")
1308 # Close preview filters as well as content filters — the journal
1309 # reputation filter is registered as a preview filter on academic
1310 # engines (arxiv, pubmed, openalex, nasa_ads, semantic_scholar)
1311 # and holds a SearXNG engine + LLM client that need releasing.
1312 if hasattr(self, "_preview_filters"):
1313 for preview_filter in self._preview_filters: 1313 ↛ 1314line 1313 didn't jump to line 1314 because the loop on line 1313 never started
1314 safe_close(preview_filter, "preview filter")
1315 if hasattr(self, "_content_filters"):
1316 for content_filter in self._content_filters: 1316 ↛ 1317line 1316 didn't jump to line 1317 because the loop on line 1316 never started
1317 safe_close(content_filter, "content filter")
1319 def __enter__(self):
1320 """Support context manager usage."""
1321 return self
1323 def __exit__(self, exc_type, exc_val, exc_tb):
1324 """Cleanup on context exit."""
1325 self.close()
1326 return False