import json
import time
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union
from urllib.parse import urlparse

from langchain_core.language_models import BaseLLM
from loguru import logger
from tenacity import (
    RetryError,
    retry,
    retry_if_exception_type,
    stop_after_attempt,
)
from tenacity.wait import wait_base

from ..advanced_search_system.filters.base_filter import BaseFilter
from ..utilities.type_utils import unwrap_setting
from ..config.constants import DEFAULT_MAX_FILTERED_RESULTS
from ..config.thread_settings import get_setting_from_snapshot
from ..security.log_sanitizer import redact_secrets, sanitize_error_message
from ..utilities.thread_context import clear_search_context, set_search_context

from .rate_limiting import RateLimitError, get_tracker
from ..constants import DEFAULT_SEARCH_TOOL

if TYPE_CHECKING:
    from ..advanced_search_system.filters.journal_reputation_filter import (
        JournalReputationFilter,
    )


# Common placeholder values that should never be treated as a real API key.
# Three call sites previously had inconsistent subsets of this list, which
# let invalid placeholders silently through the production path in
# search_engines_config.py. Centralizing here so all three stay in sync.
API_KEY_PLACEHOLDERS = frozenset(
    {
        "",
        "None",
        "null",
        "PLACEHOLDER",
        "YOUR_API_KEY_HERE",
        "YOUR_API_KEY",
        "API_KEY",
        "your_api_key",
        "your-api-key",
    }
)


def _is_api_key_placeholder(api_key: Optional[str]) -> bool:
    """Return True if ``api_key`` looks like a placeholder, not a real key.

    Catches:
    - Exact-match placeholders (see ``API_KEY_PLACEHOLDERS``)
    - Environment-variable-style names ending in ``_API_KEY``
      (e.g. ``BRAVE_API_KEY``)
    - Templates starting with ``YOUR_``
    - Angle-bracket templates: ``<key>`` or ``${KEY}``
    """
    if not api_key:
        return True
    api_key = api_key.strip()
    if api_key in API_KEY_PLACEHOLDERS:
        return True
    if api_key.endswith("_API_KEY"):
        return True
    if api_key.startswith("YOUR_"):
        return True
    if api_key.startswith("<") and api_key.endswith(">"):
        return True
    if api_key.startswith("${") and api_key.endswith("}"):
        return True
    return False


class AdaptiveWait(wait_base):
    """Custom wait strategy that uses adaptive rate limiting."""

    def __init__(self, get_wait_func):
        self.get_wait_func = get_wait_func

    def __call__(self, retry_state):
        return self.get_wait_func()


class BaseSearchEngine(ABC):
    """
    Abstract base class for search engines with two-phase retrieval capability.
    Handles common parameters and implements the two-phase search approach.

    Subclass contract for ``__init__``
    ---------------------------------
    Concrete engines should forward ``settings_snapshot`` and
    ``programmatic_mode`` to ``super().__init__`` so the base class can wire
    them up correctly. The cleanest way is to declare ``**kwargs`` and pass
    it along::

        def __init__(self, max_results=10, *, my_param=None, **kwargs):
            super().__init__(max_results=max_results, **kwargs)
            self.my_param = my_param

    Many existing engines accept ``**kwargs`` but don't forward — that
    silently drops ``programmatic_mode`` (and used to bind the wrong rate
    tracker). The factory has a post-construction safety net that calls
    ``_configure_programmatic_mode`` when the engine's mode doesn't match
    what the caller asked for, but new engines should not rely on it: the
    safety net only covers ``programmatic_mode``, not ``settings_snapshot``
    or other base kwargs.
    """

    # Class attribute to indicate if this engine searches public internet sources
    # Should be overridden by subclasses - defaults to False for safety
    is_public = False

    # Class attribute to indicate if this is a generic search engine (vs specialized)
    # Generic engines are general web search (Google, Bing, etc) vs specialized (arXiv, PubMed).
    # Note: generic does NOT imply good native ranking — see is_lexical.
    is_generic = False

    # Class attribute to indicate if this is a scientific/academic search engine
    # Scientific engines include arXiv, PubMed, Semantic Scholar, etc.
    is_scientific = False

    # Class attribute to indicate if this is a local RAG/document search engine
    # Local engines search private document collections stored locally
    is_local = False

    # Class attribute to indicate if this is a news search engine
    # News engines specialize in news articles and current events
    is_news = False

    # Class attribute to indicate if this is a code search engine
    # Code engines specialize in searching code repositories
    is_code = False

    # Class attribute to indicate if this is a book/literature search engine
    # Book engines search libraries and literary archives
    is_books = False

    # Classification: does this engine use lexical/keyword-based search?
    # Lexical engines (arXiv, PubMed, Wikipedia, Mojeek, etc.) match results by
    # keywords without ML-based ranking. This is an informational flag that can
    # drive multiple behaviors (query optimization, result deduplication, UI hints).
    # For LLM relevance filtering specifically, see needs_llm_relevance_filter.
    is_lexical = False

    # Behavioral: should the factory auto-enable LLM relevance filtering?
    # When True, the factory sets enable_llm_relevance_filter=True on the engine
    # instance, causing _filter_for_relevance() to run after previews are fetched.
    # Typically set alongside is_lexical=True, but can be set independently —
    # e.g. a non-lexical engine with noisy results could opt in.
    needs_llm_relevance_filter = False

    # Tuning for the LLM relevance filter (only applies when the filter
    # is active for this engine).
    #
    # relevance_filter_batch_size: split previews into chunks of this many
    # before sending to the LLM. Smaller batches are faster per call and
    # more reliable on weaker models which struggle with many indices in
    # one context. None or 0 = single-call mode (no batching).
    #
    # relevance_filter_max_parallel_batches: number of batches to dispatch
    # concurrently against the LLM. 1 = sequential. Most providers handle
    # parallel requests fine (Ollama with OLLAMA_NUM_PARALLEL>1, OpenAI,
    # Anthropic).
    relevance_filter_batch_size: Optional[int] = 5
    relevance_filter_max_parallel_batches: int = 10

    # Class attribute for rate limit detection patterns
    # Subclasses can override to add engine-specific patterns
    rate_limit_patterns: Set[str] = {
        "rate limit",
        "rate_limit",
        "ratelimit",
        "too many requests",
        "throttl",
        "quota exceeded",
        "quota_exceeded",
        "limit exceeded",
        "request limit",
        "api limit",
        "usage limit",
    }

    # Instance-attribute names holding credential values to redact from error
    # messages/logs via _scrub_error(). Subclasses override when they store
    # secrets under different names (e.g. Elasticsearch: _api_key/_password).
    # Centralizing the list keeps every dual-scrub call site uniform and
    # prevents the per-site drift that previously dropped a secret.
    _secret_attrs: tuple[str, ...] = ("api_key",)

    @staticmethod
    def _ensure_list(value, *, default=None):
        """Normalize a value that should be a list.

        Handles JSON-encoded strings, comma-separated strings, and
        already-parsed lists.  Returns *default* (empty list when not
        supplied) for ``None`` or empty/unparseable input.
        """
        if default is None:
            default = []
        if value is None:
            return default
        if isinstance(value, list):
            return value
        if isinstance(value, str):
            stripped = value.strip()
            if not stripped:
                return default
            if stripped.startswith("["):
                try:
                    parsed = json.loads(stripped)
                    if isinstance(parsed, list):
                        return [str(item) for item in parsed]
                except (json.JSONDecodeError, ValueError, RecursionError):
                    pass
            return [
                item.strip() for item in stripped.split(",") if item.strip()
            ]
        return default

    @classmethod
    def _load_engine_class(cls, name: str, config: Dict[str, Any]):
        """
        Helper method to load an engine class dynamically.

        Args:
            name: Engine name
            config: Engine configuration dict with module_path and class_name

        Returns:
            Tuple of (success: bool, engine_class or None, error_msg or None)
        """
        from ..security.module_whitelist import (
            ModuleNotAllowedError,
            get_safe_module_class,
        )

        try:
            module_path = config.get("module_path")
            class_name = config.get("class_name")

            if not module_path or not class_name:
                return (
                    False,
                    None,
                    f"Missing module_path or class_name for {name}",
                )

            # Use whitelist-validated safe import
            engine_class = get_safe_module_class(module_path, class_name)

            return True, engine_class, None

        except ModuleNotAllowedError as e:
            return (
                False,
                None,
                f"Security error loading engine class for {name}: {e}",
            )
        except Exception as e:
            return False, None, f"Could not load engine class for {name}: {e}"

    @classmethod
    def _check_api_key_availability(
        cls, name: str, config: Dict[str, Any]
    ) -> bool:
        """
        Helper method to check if an engine's API key is available and valid.

        Args:
            name: Engine name
            config: Engine configuration dict

        Returns:
            True if API key is not required or is available and valid
        """
        from loguru import logger

        if not config.get("requires_api_key", False):
            return True

        api_key = config.get("api_key", "").strip()

        # Check for common placeholder values
        if _is_api_key_placeholder(api_key):
            logger.debug(
                f"Skipping {name} - requires API key but none configured"
            )
            return False

        return True

    def __init__(
        self,
        llm: Optional[BaseLLM] = None,
        max_filtered_results: Optional[int] = None,
        max_results: Optional[int] = 10,  # Default value if not provided
        preview_filters: List[BaseFilter] | None = None,
        content_filters: List[BaseFilter] | None = None,
        search_snippets_only: bool = True,  # New parameter with default
        include_full_content: bool = False,
        settings_snapshot: Optional[Dict[str, Any]] = None,
        programmatic_mode: bool = False,
        **kwargs,
    ):
        """
        Initialize the search engine with common parameters.

        Args:
            llm: Optional language model for relevance filtering
            max_filtered_results: Maximum number of results to keep after filtering
            max_results: Maximum number of search results to return
            preview_filters: Filters that will be applied to all previews
                produced by the search engine, before relevancy checks.
            content_filters: Filters that will be applied to the full content
                produced by the search engine, after relevancy checks.
            search_snippets_only: Whether to return only snippets or full content
            include_full_content: Whether to use FullSearchResults for full webpage content
            settings_snapshot: Settings snapshot for configuration
            programmatic_mode: If True, disables database operations and uses memory-only tracking
            **kwargs: Additional engine-specific parameters
        """
        if max_filtered_results is None:
            max_filtered_results = DEFAULT_MAX_FILTERED_RESULTS
        if max_results is None:
            max_results = 10
        self._preview_filters: List[BaseFilter] = (
            preview_filters if preview_filters is not None else []
        )
        self._content_filters: List[BaseFilter] = (
            content_filters if content_filters is not None else []
        )

        self.llm = llm  # LLM for relevance filtering
        self._max_filtered_results = int(
            max_filtered_results
        )  # Ensure it's an integer
        self._max_results = max(
            1, int(max_results)
        )  # Ensure it's a positive integer
        self.search_snippets_only = search_snippets_only  # Store the setting
        self.include_full_content = include_full_content
        self.settings_snapshot = (
            settings_snapshot or {}
        )  # Store settings snapshot

        self.engine_type = self.__class__.__name__
        self._engine_name: str = ""  # set by the factory after construction
        # The snapshot dict the runtime egress backstop last verified
        # against (identity comparison; holding the reference keeps the
        # memo valid — a verification re-runs when the caller assigns a
        # NEW snapshot) plus the policy-relevant VALUES it carried at
        # verification time, so an in-place mutation of the scope or
        # primary key on the same dict is also caught.
        self._egress_verified_snapshot: Optional[Dict[str, Any]] = None
        self._egress_verified_policy_key: Optional[tuple] = None
        self._egress_skip_warned = False
        self._configure_programmatic_mode(programmatic_mode)
        self._last_wait_time = (
            0.0  # Default to 0 for successful searches without rate limiting
        )
        self._last_results_count = 0

    def _create_journal_filter(
        self,
        engine_name: str,
        llm: Optional[BaseLLM],
        settings_snapshot: Optional[Dict[str, Any]],
    ) -> Optional["JournalReputationFilter"]:
        """Build the default :class:`JournalReputationFilter` for this engine.

        Wraps the identical 8-line ``JournalReputationFilter.create_default``
        boilerplate that previously lived in every academic subclass. The
        ``engine_name`` is passed explicitly because the auto-derived class
        name does not match the settings key for ``nasa_ads`` (would be
        ``"nasaads"``) or ``semantic_scholar`` (would be
        ``"semanticscholar"``).

        ``llm`` and ``settings_snapshot`` are passed in rather than read from
        ``self`` because subclasses build their preview filters *before*
        calling ``super().__init__()`` (the filter is handed to the parent
        constructor), so ``self.llm`` / ``self.settings_snapshot`` do not
        exist yet at call time.

        Args:
            engine_name: Settings key identifying the engine (e.g.
                ``"arxiv"``, ``"nasa_ads"``).
            llm: Language model used for the filter's relevance pass.
            settings_snapshot: Settings snapshot for configuration lookups.

        Returns:
            A configured :class:`JournalReputationFilter`, or ``None`` if
            filtering is disabled in settings for this engine.
        """
        from ..advanced_search_system.filters.journal_reputation_filter import (
            JournalReputationFilter,
        )

        return JournalReputationFilter.create_default(
            model=llm,  # type: ignore[arg-type]
            engine_name=engine_name,
            settings_snapshot=settings_snapshot,
        )

    def _configure_programmatic_mode(self, programmatic_mode: bool) -> None:
        """Set ``programmatic_mode`` and (re)bind the matching rate tracker.

        Called from ``__init__`` and from the factory as a fallback when an
        engine subclass swallows the ``programmatic_mode`` kwarg without
        forwarding it to ``super().__init__``. Safe to call after init —
        rebinding ``rate_tracker`` discards the previous tracker (no
        resources to release) and the new one starts with empty in-memory
        caches.
        """
        self.programmatic_mode = programmatic_mode
        if programmatic_mode:
            from .rate_limiting.tracker import AdaptiveRateLimitTracker

            self.rate_tracker = AdaptiveRateLimitTracker(
                settings_snapshot=self.settings_snapshot,
                programmatic_mode=programmatic_mode,
            )
        else:
            self.rate_tracker = get_tracker()

    @property
    def max_filtered_results(self) -> int:
        """Get the maximum number of filtered results."""
        return self._max_filtered_results

    @max_filtered_results.setter
    def max_filtered_results(self, value: int) -> None:
        """Set the maximum number of filtered results."""
        if value is None:
            value = DEFAULT_MAX_FILTERED_RESULTS
            logger.warning(
                f"Setting max_filtered_results to {DEFAULT_MAX_FILTERED_RESULTS}"
            )
        self._max_filtered_results = int(value)

    @property
    def max_results(self) -> int:
        """Get the maximum number of search results."""
        return self._max_results

    @max_results.setter
    def max_results(self, value: int) -> None:
        """Set the maximum number of search results."""
        if value is None:
            value = 10
        self._max_results = max(1, int(value))

    def _get_adaptive_wait(self) -> float:
        """Get adaptive wait time from tracker."""
        wait_time = self.rate_tracker.get_wait_time(self.engine_type)
        self._last_wait_time = wait_time
        logger.debug(
            f"{self.engine_type} waiting {wait_time:.2f}s before retry"
        )
        return wait_time

    def _record_retry_outcome(self, retry_state) -> None:
        """Record outcome after retry completes."""
        success = (
            not retry_state.outcome.failed if retry_state.outcome else False
        )
        self.rate_tracker.record_outcome(
            self.engine_type,
            self._last_wait_time or 0,
            success,
            retry_state.attempt_number,
            error_type="RateLimitError" if not success else None,
            search_result_count=self._last_results_count if success else 0,
        )

    def _verify_egress_scope(self) -> None:
        """Runtime backstop: verify this engine is allowed under the egress
        scope in ``self.settings_snapshot`` before executing a search.

        No-op when ``settings_snapshot`` is missing or empty (programmatic
        API callers) or when ``_engine_name`` has not been set.  Raises
        ``PolicyDeniedError`` when the policy denies the engine; any other
        internal error in the policy evaluation is logged and ignored so a
        broken backstop never takes down searches the factory PEP already
        approved.

        This is a defense-in-depth check behind the factory PEP, the
        strategy-level filters, and the audit hook.  It covers engines that
        bypassed the factory (direct instantiation with a snapshot).  Note
        the limit: the snapshot is the one captured at construction, so a
        scope change AFTER construction is only caught if the caller also
        refreshes ``settings_snapshot`` (assigns a NEW dict — in-place
        mutation of the existing dict is not detected, see below).

        The verification is memoized per snapshot IDENTITY plus the
        policy-relevant VALUES (scope, primary): the full policy inputs
        are stable for a given snapshot object, so re-evaluating on every
        ``run()`` would only repeat the same decision — and under
        ADAPTIVE scope with a URL-configurable primary the evaluation can
        include a DNS lookup (bounded at 2s), which must not become a
        per-search stall. Assigning a refreshed snapshot OR mutating the
        scope/primary keys in place invalidates the memo and re-verifies.
        Denials are never memoized (they raise).
        """
        if not self.settings_snapshot:
            return
        if not self._engine_name:
            # A snapshot WITHOUT a stamped engine name means a direct
            # instantiation the factory never saw and no subclass
            # self-stamps — the backstop cannot evaluate it. Surface the
            # gap in the audit log (once per instance) instead of
            # skipping silently.
            if not self._egress_skip_warned:
                self._egress_skip_warned = True
                logger.bind(policy_audit=True).warning(
                    "egress backstop skipped: engine has a settings "
                    "snapshot but no _engine_name stamped",
                    engine_type=self.__class__.__name__,
                )
            return
        if (
            self.settings_snapshot is self._egress_verified_snapshot
            and self._egress_policy_key() == self._egress_verified_policy_key
        ):
            return
        from ..security.egress.policy import PolicyDeniedError

        try:
            self._check_egress_policy()
        except PolicyDeniedError:
            logger.bind(policy_audit=True).warning(
                "runtime egress backstop denied engine",
                engine=self._engine_name,
            )
            raise
        except Exception:
            logger.bind(policy_audit=True).debug(
                "egress runtime verify errored; "
                "factory PEP still enforces at instantiation",
                engine=self._engine_name,
            )
        else:
            self._egress_verified_snapshot = self.settings_snapshot
            self._egress_verified_policy_key = self._egress_policy_key()

    def _egress_policy_key(self) -> tuple:
        """The policy-relevant snapshot values the backstop memo guards on.

        Cheap dict reads only — no context construction, no DNS. Not an
        exhaustive input set (URL settings or DB-side collection flags can
        also influence the decision), but it covers the keys an in-place
        mutation would realistically target; the factory PEP remains the
        primary enforcement for everything else.

        MAINTENANCE: if a new snapshot key ever becomes policy-relevant in
        ``context_from_snapshot``/``evaluate_engine``, add it here too —
        otherwise an in-place mutation of that key returns a stale memo.

        """
        scope = unwrap_setting(
            self.settings_snapshot.get("policy.egress_scope")
        )
        primary = unwrap_setting(self.settings_snapshot.get("search.tool"))
        return (scope, primary)

    def _check_egress_policy(self) -> None:
        """Inner helper so the raise is not inside the except handler
        (ruff TRY301).  Raises PolicyDeniedError on scope mismatch."""
        from ..security.egress.policy import (
            PolicyDeniedError,
            context_from_snapshot,
            evaluate_engine,
        )

        primary = unwrap_setting(
            self.settings_snapshot.get("search.tool", self._engine_name)
        )
        ctx = context_from_snapshot(
            self.settings_snapshot,
            primary or self._engine_name,
            username=self.settings_snapshot.get("_username"),
        )
        decision = evaluate_engine(
            self._engine_name,
            ctx,
            settings_snapshot=self.settings_snapshot,
        )
        if not decision.allowed:
            raise PolicyDeniedError(decision, target=self._engine_name)

    def run(
        self, query: str, research_context: Dict[str, Any] | None = None
    ) -> List[Dict[str, Any]]:
        """
        Run the search engine with a given query, retrieving and filtering results.
        This implements a two-phase retrieval approach:
        1. Get preview information for many results
        2. Filter the previews for relevance
        3. Get full content for only the relevant results

        Args:
            query: The search query
            research_context: Context from previous research to use.

        Returns:
            List of search results with full content (if available)
        """
        logger.info(f"---Execute a search using {self.__class__.__name__}---")

        # Runtime egress scope backstop: verify this engine is allowed
        # before making any HTTP requests.
        self._verify_egress_scope()

        # Track search call for metrics (if available and not in programmatic mode)
        should_record_metrics = False
        context_was_set = False
        if not self.programmatic_mode:
            from ..metrics.search_tracker import SearchTracker

            should_record_metrics = True

            # For thread-safe context propagation: if we have research_context parameter, use it
            # Otherwise, try to inherit from current thread context (normal case)
            # This allows strategies running in threads to explicitly pass context when needed
            if research_context:
                # Explicit context provided - use it and set it for this thread
                set_search_context(research_context)
                context_was_set = True

        engine_name = self.__class__.__name__.replace(
            "SearchEngine", ""
        ).lower()
        start_time = time.time()

        success = True
        error_message = None
        results_count = 0

        # Define the core search function with retry logic
        if self.rate_tracker.enabled:
            # Rate limiting enabled - use retry with adaptive wait
            @retry(
                stop=stop_after_attempt(3),
                wait=AdaptiveWait(lambda: self._get_adaptive_wait()),
                retry=retry_if_exception_type((RateLimitError,)),
                after=self._record_retry_outcome,
                reraise=True,
            )
            def _run_with_retry():
                nonlocal success, error_message, results_count
                return _execute_search()
        else:
            # Rate limiting disabled - run without retry
            def _run_with_retry():
                nonlocal success, error_message, results_count
                return _execute_search()

        def _execute_search():
            nonlocal success, error_message, results_count

            try:
                # Step 1: Get preview information for items
                previews = self._get_previews(query)
                if not previews:
                    logger.info(
                        f"Search engine {self.__class__.__name__} returned no preview results for query: {query}"
                    )
                    results_count = 0
                    return []

                # Pre-filter enrichment: resolve DOIs to OpenAlex source
                # IDs BEFORE the preview filters run. The
                # JournalReputationFilter is registered as a preview
                # filter and uses ``result["openalex_source_id"]`` for
                # Tier 2 lookups; running enrichment afterwards (as the
                # old pipeline did) left the field empty at filter time,
                # silently forcing a fragile-name-match fallback. Only
                # for scientific engines whose results carry DOIs.
                if getattr(self, "is_scientific", False):
                    try:
                        from ..utilities.openalex_enrichment import (
                            enrich_results_with_source_ids,
                        )

                        email = getattr(self, "email", None)
                        previews = enrich_results_with_source_ids(
                            previews, email=email
                        )
                    except Exception:
                        logger.debug(
                            "DOI enrichment skipped (import or call failed)"
                        )

                for preview_filter in self._preview_filters:
                    previews = preview_filter.filter_results(previews, query)

                # Step 2: Filter previews for relevance with LLM
                enable_llm_filter = getattr(
                    self, "enable_llm_relevance_filter", False
                )

                if enable_llm_filter and self.llm:
                    filtered_items = self._filter_for_relevance(previews, query)
                else:
                    filtered_items = previews
                    logger.debug(
                        f"[{type(self).__name__}] Relevance filter skipped "
                        f"(enabled={enable_llm_filter}, "
                        f"llm={'yes' if self.llm else 'no'})"
                    )

                # Step 3: Get full content for filtered items
                if self.search_snippets_only:
                    logger.info("Returning snippet-only results as per config")
                    results = filtered_items
                else:
                    results = self._get_full_content(filtered_items)

                for content_filter in self._content_filters:
                    results = content_filter.filter_results(results, query)

                results_count = len(results)
                self._last_results_count = results_count

                # Record success if we get here and rate limiting is enabled
                if self.rate_tracker.enabled:
                    logger.info(
                        f"Recording successful search for {self.engine_type}: wait_time={self._last_wait_time}s, results={results_count}"
                    )
                    self.rate_tracker.record_outcome(
                        self.engine_type,
                        self._last_wait_time,
                        success=True,
                        retry_count=1,  # First attempt succeeded
                        search_result_count=results_count,
                    )
                else:
                    logger.info(
                        f"Rate limiting disabled, not recording search for {self.engine_type}"
                    )

                return results

            except RateLimitError:
                # Only re-raise if rate limiting is enabled
                if self.rate_tracker.enabled:
                    raise
                # If rate limiting is disabled, treat as regular error
                success = False
                error_message = "Rate limit hit but rate limiting disabled"
                logger.warning(
                    f"Rate limit hit on {self.__class__.__name__} but rate limiting is disabled"
                )
                results_count = 0
                return []
            except Exception as e:
                # Other errors - don't retry
                success = False
                # Sanitize before it flows to SearchTracker.record_search
                # (and the database) and to the log. Use logger.warning with
                # the dual-scrubbed text instead of logger.exception: the
                # cause chain frequently carries the request URL or auth
                # header from upstream HTTP clients (see #4131).
                error_message = safe_msg = self._scrub_error(e)
                logger.warning(
                    f"Search engine {self.__class__.__name__} failed: {safe_msg}"
                )
                results_count = 0
                return []

        try:
            return _run_with_retry()  # type: ignore[no-any-return]
        except RetryError as e:
            # All retries exhausted
            success = False
            error_message = self._scrub_error(
                f"Rate limited after all retries: {e}"
            )
            safe_msg = self._scrub_error(e)
            logger.warning(
                f"{self.__class__.__name__} failed after all retries: {safe_msg}"
            )
            return []
        except Exception as e:
            success = False
            error_message = safe_msg = self._scrub_error(e)
            logger.warning(
                f"Search engine {self.__class__.__name__} error: {safe_msg}"
            )
            return []
        finally:
            try:
                # Record search metrics BEFORE clearing context (record_search needs it)
                if should_record_metrics:
                    response_time_ms = int((time.time() - start_time) * 1000)
                    SearchTracker.record_search(
                        engine_name=engine_name,
                        query=query,
                        results_count=results_count,
                        response_time_ms=response_time_ms,
                        success=success,
                        error_message=error_message,
                    )
            finally:
                # Clean up temporary search result storage
                for attr in self._temp_attributes():
                    if hasattr(self, attr):
                        delattr(self, attr)
                # ALWAYS clean up search context, even if recording fails
                if context_was_set:
                    clear_search_context()

    def invoke(self, query: str) -> List[Dict[str, Any]]:
        """Compatibility method for LangChain tools"""
        return self.run(query)

    def _filter_for_relevance(
        self, previews: List[Dict[str, Any]], query: str
    ) -> List[Dict[str, Any]]:
        """
        Filter search results by relevance using the LLM.

        Delegates to the ``relevance_filter`` module, which prompts the
        LLM for a plain-text list of relevant indices and parses them
        with a regex (no structured output).

        Args:
            previews: List of preview dictionaries
            query: The original search query

        Returns:
            Filtered list of preview dictionaries
        """
        engine_name = type(self).__name__

        if not self.llm or len(previews) <= 1:
            logger.debug(
                f"[{engine_name}] Skipping relevance filter "
                f"(llm={'yes' if self.llm else 'no'}, "
                f"previews={len(previews)})"
            )
            return previews

        from .relevance_filter import filter_previews_for_relevance

        return filter_previews_for_relevance(
            llm=self.llm,
            previews=previews,
            query=query,
            max_filtered_results=self.max_filtered_results,
            engine_name=engine_name,
            batch_size=self.relevance_filter_batch_size,
            max_parallel_batches=self.relevance_filter_max_parallel_batches,
        )

    # =========================================================================
    # Shared Helper Methods for Subclasses
    # =========================================================================

    @staticmethod
    def _is_valid_api_key(api_key: Optional[str]) -> bool:
        """
        Check if an API key is valid (not a placeholder value).

        Args:
            api_key: The API key to validate

        Returns:
            True if the key appears to be a real API key, False if it's a placeholder

        Example:
            >>> BaseSearchEngine._is_valid_api_key("sk-abc123")
            True
            >>> BaseSearchEngine._is_valid_api_key("YOUR_API_KEY_HERE")
            False
        """
        return not _is_api_key_placeholder(api_key)

    @staticmethod
    def _extract_display_link(url: str, fallback: str = "") -> str:
        """Extract the netloc (domain) from a URL for display purposes.

        Args:
            url: The URL to extract from
            fallback: Value to return on parse failure (default: empty string)

        Returns:
            The netloc portion of the URL, or *fallback* if parsing fails.
        """
        if not url:
            return fallback
        try:
            parsed = urlparse(url)
            return parsed.netloc or fallback
        except Exception:
            return fallback

    @staticmethod
    def _clean_result_url(value: Any) -> str:
        """Normalize a raw search-result URL for the validity gate.

        Coerces ``None`` (and any other falsy value) to ``""`` and any
        truthy value to ``str`` before stripping surrounding whitespace.

        The strip is **load-bearing**, not merely cosmetic: several engines
        gate the URL on a ``url.lower().startswith(("http://", "https://"))``
        prefix check (see ``_is_valid_search_result``) *before* it reaches
        the SSRF validator's own internal strip, so leading/trailing
        whitespace — common in HTML-scraped ``href`` attributes — would
        silently drop an otherwise-valid result. Cleaning once at extraction
        also keeps the URL tidy for logs and downstream consumers (preview
        ``id``/``link`` fields, etc.).

        Args:
            value: The raw URL value from a search result, e.g.
                ``result.get("url")`` or a parsed HTML ``href`` attribute.

        Returns:
            The whitespace-stripped URL string, or ``""`` if *value* is
            falsy (``None``, empty, etc.).
        """
        if not value:
            return ""
        return str(value).strip()

    def _resolve_api_key(
        self,
        api_key: Optional[str],
        setting_key: str,
        engine_name: str = "search engine",
        settings_snapshot: Optional[Dict[str, Any]] = None,
    ) -> str:
        """
        Resolve an API key from multiple sources with priority order.

        Environment variables are handled automatically by SettingsManager
        when building the settings snapshot, so they don't need to be
        checked separately here.

        Priority order:
        1. Direct parameter (api_key argument)
        2. Settings snapshot (via setting_key)

        Args:
            api_key: API key passed directly as parameter
            setting_key: Key to look up in settings snapshot (e.g., "search.brave_api_key")
            engine_name: Human-readable engine name for error messages
            settings_snapshot: Optional settings snapshot dict (uses self.settings_snapshot if not provided)

        Returns:
            The resolved API key string

        Raises:
            ValueError: If no valid API key is found from any source

        Example:
            >>> engine._resolve_api_key(
            ...     api_key=None,
            ...     setting_key="search.brave_api_key",
            ...     engine_name="Brave Search"
            ... )
            "sk-abc123..."
        """
        # Use instance settings snapshot if not provided
        if settings_snapshot is None:
            settings_snapshot = self.settings_snapshot

        # Priority 1: Direct parameter
        if self._is_valid_api_key(api_key) and api_key is not None:
            return api_key.strip()

        # Priority 2: Settings snapshot (includes env var overrides via SettingsManager)
        if settings_snapshot:
            settings_value = get_setting_from_snapshot(
                setting_key,
                default=None,
                settings_snapshot=settings_snapshot,
            )
            if self._is_valid_api_key(settings_value):
                return settings_value.strip() if settings_value else ""

        # No valid API key found
        masked_key = self._mask_api_key(str(api_key)) if api_key else "None"
        raise ValueError(
            f"No valid API key found for {engine_name}. "
            f"Checked: direct parameter ({masked_key}), "
            f"settings key '{setting_key}'. "
            f"Please provide a valid API key."
        )

    def _is_rate_limit_error(
        self,
        error: Union[Exception, str, int],
        additional_patterns: Optional[Set[str]] = None,
    ) -> bool:
        """
        Detect if an error is a rate limit error.

        Checks multiple sources for rate limit indicators:
        - HTTP status code 429
        - HTTPError response objects
        - Error messages containing rate limit phrases

        Args:
            error: The error to check (Exception, string, or HTTP status code)
            additional_patterns: Optional set of additional patterns to match

        Returns:
            True if the error appears to be a rate limit error

        Example:
            >>> engine._is_rate_limit_error(429)
            True
            >>> engine._is_rate_limit_error("Rate limit exceeded")
            True
            >>> engine._is_rate_limit_error(ValueError("Invalid input"))
            False
        """
        # Combine default and additional patterns
        patterns = self.rate_limit_patterns.copy()
        if additional_patterns:
            patterns.update(additional_patterns)

        # Check integer status code directly
        if isinstance(error, int):
            return error == 429

        # Convert to string for pattern matching
        error_str = ""
        status_code = None

        if isinstance(error, str):
            error_str = error
        elif isinstance(error, Exception):
            error_str = str(error)

            # Check for HTTP status code in common HTTP error types
            if hasattr(error, "status_code"):
                status_code = error.status_code
            elif hasattr(error, "response"):
                response = error.response
                if hasattr(response, "status_code"):
                    status_code = response.status_code

        # Check status code first
        if status_code == 429:
            return True

        # Case-insensitive pattern matching
        error_lower = error_str.lower()
        for pattern in patterns:
            if pattern.lower() in error_lower:
                return True

        return False

    def _raise_if_rate_limit(
        self,
        error: Union[Exception, str, int],
        additional_patterns: Optional[Set[str]] = None,
    ) -> None:
        """
        Raise RateLimitError if the given error is a rate limit error.

        Convenience method that combines _is_rate_limit_error check with
        raising RateLimitError.

        Args:
            error: The error to check
            additional_patterns: Optional set of additional patterns to match

        Raises:
            RateLimitError: If the error is detected as a rate limit error

        Example:
            >>> try:
            ...     response = make_api_call()
            ... except Exception as e:
            ...     engine._raise_if_rate_limit(e)
        """
        if self._is_rate_limit_error(error, additional_patterns):
            error_msg = str(error) if not isinstance(error, str) else error
            raise RateLimitError(
                f"Rate limit detected: {self._sanitize_error_message(error_msg)}"
            )

    def _extract_full_result(self, item: Dict[str, Any]) -> Dict[str, Any]:
        """
        Extract the full result from an item that may contain a _full_result key.

        This is a helper for the default _get_full_content implementation.
        It extracts data from the _full_result key if present, otherwise uses
        the item directly, and removes the internal _full_result key.

        Args:
            item: A search result item that may contain a _full_result key

        Returns:
            A dictionary with the full result data, without the _full_result key

        Example:
            >>> engine._extract_full_result({"title": "A", "_full_result": {"title": "A", "content": "Full"}})
            {"title": "A", "content": "Full"}
        """
        source = item.get("_full_result")
        if source is None:
            source = item
        return {k: v for k, v in source.items() if k != "_full_result"}

    def _get_full_content(
        self, relevant_items: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """
        Get full content for the relevant items.

        Default implementation extracts data from _full_result keys if present.
        Subclasses can override this method to fetch additional content from
        external sources (e.g., web scraping, API calls).

        Args:
            relevant_items: List of relevant preview dictionaries

        Returns:
            List of result dictionaries with full content

        Example:
            >>> engine._get_full_content([
            ...     {"title": "A", "_full_result": {"title": "A", "content": "Full A"}},
            ...     {"title": "B"}
            ... ])
            [{"title": "A", "content": "Full A"}, {"title": "B"}]
        """
        if not relevant_items:
            return []
        return [self._extract_full_result(item) for item in relevant_items]

    def _build_full_search_egress_context(self):
        """Build an EgressContext from self.settings_snapshot for the
        per-URL scope gate inside FullSearchResults. Returns None when a
        context can't be built and disables full-content fetching when
        policy evaluation fails on a supplied snapshot.

        The previous "fail closed by returning None" comment was wrong:
        full_search.py treats ``egress_context is None`` as
        ``evaluate_url_fn = None`` and skips the per-URL scope check
        entirely, leaving only SSRF — which allows public hosts under
        PRIVATE_ONLY. When the snapshot was supplied but the policy is
        unevaluable, we now turn full-content fetching off for this
        engine instead.
        """
        if not getattr(self, "settings_snapshot", None):
            return None
        try:
            from ..security.egress.policy import (
                PolicyDeniedError,
                context_from_snapshot,
            )
        except ImportError:
            logger.debug(
                "egress_policy unavailable in search_engine_base; "
                "FullSearchResults fetches will be SSRF-only"
            )
            return None

        primary_raw = unwrap_setting(
            self.settings_snapshot.get("search.tool", DEFAULT_SEARCH_TOOL)
        )
        try:
            return context_from_snapshot(
                self.settings_snapshot, primary_raw or DEFAULT_SEARCH_TOOL
            )
        except (PolicyDeniedError, ValueError) as exc:
            logger.bind(policy_audit=True).warning(
                "Disabling include_full_content: egress policy could "
                "not be evaluated for this engine",
                reason=str(exc),
            )
            self.include_full_content = False
            return None

    def _init_full_search(
        self,
        web_search=None,
        language="en",
        max_results=10,
        region=None,
        time_period=None,
        safe_search=None,
    ):
        """Initialize FullSearchResults if include_full_content is True.

        Call this at the end of your __init__ after setting up your search wrapper.

        Args:
            web_search: The search wrapper/engine to pass to FullSearchResults
            language: Language for search results
            max_results: Maximum number of results
            region: Region/country code for results
            time_period: Time period filter
            safe_search: Safe search setting (string value for FullSearchResults)
        """
        if self.include_full_content and self.llm:
            try:
                from .engines.full_search import FullSearchResults

                # Egress context for per-URL scope gating. Built once
                # here (not per fetch) so the policy snapshot is the
                # same one used by the factory PEP that authorized this
                # engine.
                egress_context = self._build_full_search_egress_context()
                self.full_search = FullSearchResults(
                    llm=self.llm,
                    web_search=web_search,
                    language=language,
                    max_results=max_results,
                    region=region,
                    time=time_period,
                    safesearch=safe_search,
                    settings_snapshot=self.settings_snapshot,
                    egress_context=egress_context,
                )
            except ImportError:
                logger.warning(
                    "FullSearchResults not available. "
                    "Full content retrieval disabled."
                )
                self.include_full_content = False

    def _temp_attributes(self):
        """Return list of temporary attribute names to clean up after run().

        Override in subclasses that store additional temporary data.
        """
        return ["_search_results"]

    def _sanitize_error_message(self, message: str) -> str:
        """
        Remove/mask API keys, tokens, and secrets from error messages.

        Uses pattern matching for common credential formats.

        Args:
            message: The error message to sanitize

        Returns:
            Sanitized message with sensitive data redacted

        Example:
            >>> engine._sanitize_error_message(
            ...     "Error with key sk-abc123def456ghi789jkl012"
            ... )
            "Error with key [REDACTED_KEY]"
        """
        return sanitize_error_message(message)

    def _scrub_error(self, error: Union[BaseException, str]) -> str:
        """Return a log/DB-safe rendering of *error*.

        Applies the "dual-scrub": the regex-based
        :func:`sanitize_error_message` pass (catches credential *shapes* —
        Bearer tokens, URL-embedded credentials, ``sk-``/``pk-`` keys)
        followed by :func:`redact_secrets` with this engine's known literal
        secret values (resolved from ``_secret_attrs``).

        Use this at every catch site that logs or persists an exception so
        the two scrub passes can never drift apart per-engine. *error* may be
        an exception or a pre-built message string.

        Defensive by design: this runs inside ``except`` blocks, so it must
        never raise. ``str(error)`` is guarded (a custom exception whose
        ``__str__`` raises won't crash the handler) and each secret is coerced
        to ``str`` (a misconfigured non-string secret, e.g. an int from
        settings, won't trip ``redact_secrets``' ``len()`` check).
        """
        try:
            message = str(error)
        except Exception:
            message = f"<unprintable {type(error).__name__}>"
        # Coerce truthy non-str secrets to str; keep None/falsy as-is
        # (redact_secrets filters those out).
        secrets = [
            (v and str(v))
            for v in (getattr(self, name, None) for name in self._secret_attrs)
        ]
        return redact_secrets(sanitize_error_message(message), *secrets)

    def _mask_api_key(self, api_key: str, visible_chars: int = 4) -> str:
        """
        Mask an API key for safe logging, showing only first and last characters.

        Args:
            api_key: The API key to mask
            visible_chars: Number of characters to show at start and end

        Returns:
            Masked API key in format "sk-1...nop" or "***" for short keys

        Example:
            >>> engine._mask_api_key("sk-abcdefghijklmnop123456")
            "sk-a...3456"
            >>> engine._mask_api_key("short")
            "***"
        """
        if not api_key:
            return "***"

        api_key = str(api_key).strip()

        if len(api_key) <= visible_chars * 2:
            return "***"

        return f"{api_key[:visible_chars]}...{api_key[-visible_chars:]}"

    @abstractmethod
    def _get_previews(self, query: str) -> List[Dict[str, Any]]:
        """
        Get preview information (titles, summaries) for initial search results.

        Args:
            query: The search query

        Returns:
            List of preview dictionaries with at least 'id', 'title', and 'snippet' keys
        """
        pass

    def close(self) -> None:
        """
        Close any resources held by this search engine.

        Subclasses with HTTP sessions or other resources should override this.
        The base implementation safely closes any 'session' attribute if present
        and closes both preview and content filters that hold resources.
        """
        from ..utilities.resource_utils import safe_close

        if hasattr(self, "session") and self.session is not None:
            safe_close(self.session, "HTTP session")
        # Close preview filters as well as content filters — the journal
        # reputation filter is registered as a preview filter on academic
        # engines (arxiv, pubmed, openalex, nasa_ads, semantic_scholar)
        # and holds a SearXNG engine + LLM client that need releasing.
        if hasattr(self, "_preview_filters"):
            for preview_filter in self._preview_filters:
                safe_close(preview_filter, "preview filter")
        if hasattr(self, "_content_filters"):
            for content_filter in self._content_filters:
                safe_close(content_filter, "content filter")

    def __enter__(self):
        """Support context manager usage."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Cleanup on context exit."""
        self.close()
        return False
