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