_core.py

Go to the documentation of this file.

# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=too-few-public-methods,missing-module-docstring
 
from __future__ import annotations
 
__all__ = ["PluginInfo", "Plugin", "PluginCfg", "PluginStorage"]
 
import abc
import importlib
import inspect
import logging
import re
import typing
 
from dataclasses import dataclass, field
 
from searx.extended_types import SXNG_Request
from searx.result_types import Result
 
if typing.TYPE_CHECKING:
    from searx.search import SearchWithPlugins
    import flask
 
log: logging.Logger = logging.getLogger("searx.plugins")
 
 
@dataclass
class PluginInfo:
    """Object that holds informations about a *plugin*, these infos are shown to
    the user in the Preferences menu.
 
    To be able to translate the information into other languages, the text must
    be written in English and translated with :py:obj:`flask_babel.gettext`.
    """
 
    id: str
    """The ID-selector in HTML/CSS `#<id>`."""
 
    name: str
    """Name of the *plugin*."""
 
    description: str
    """Short description of the *answerer*."""
 
    preference_section: typing.Literal["general", "ui", "privacy", "query"] | None = "general"
    """Section (tab/group) in the preferences where this plugin is shown to the
    user.
 
    The value ``query`` is reserved for plugins that are activated via a
    *keyword* as part of a search query, see:
 
    - :py:obj:`PluginInfo.examples`
    - :py:obj:`Plugin.keywords`
 
    Those plugins are shown in the preferences in tab *Special Queries*.
    """
 
    examples: list[str] = field(default_factory=list)
    """List of short examples of the usage / of query terms."""
 
    keywords: list[str] = field(default_factory=list)
    """See :py:obj:`Plugin.keywords`"""
 
 
ID_REGXP = re.compile("[a-z][a-z0-9].*")
 
 
class Plugin(abc.ABC):
    """Abstract base class of all Plugins."""
 
    id: str = ""
    """The ID (suffix) in the HTML form."""
 
    active: typing.ClassVar[bool]
    """Plugin is enabled/disabled by default (:py:obj:`PluginCfg.active`)."""
 
    keywords: list[str] = []
    """Keywords in the search query that activate the plugin.  The *keyword* is
    the first word in a search query.  If a plugin should be executed regardless
    of the search query, the list of keywords should be empty (which is also the
    default in the base class for Plugins)."""
 
    log: logging.Logger
    """A logger object, is automatically initialized when calling the
    constructor (if not already set in the subclass)."""
 
    info: PluginInfo
    """Informations about the *plugin*, see :py:obj:`PluginInfo`."""
 
    fqn: str = ""
 
    def __init__(self, plg_cfg: PluginCfg) -> None:
        super().__init__()
        if not self.fqn:
            self.fqn = self.__class__.__mro__[0].__module__
 
        # names from the configuration
        for n, v in plg_cfg.__dict__.items():
            setattr(self, n, v)
 
        # names that must be set by the plugin implementation
        for attr in [
            "id",
        ]:
            if getattr(self, attr, None) is None:
                raise NotImplementedError(f"plugin {self} is missing attribute {attr}")
 
        if not ID_REGXP.match(self.id):
            raise ValueError(f"plugin ID {self.id} contains invalid character (use lowercase ASCII)")
 
        if not getattr(self, "log", None):
            pkg_name = inspect.getmodule(self.__class__).__package__  # type: ignore
            self.log = logging.getLogger(f"{pkg_name}.{self.id}")
 
    def __hash__(self) -> int:
        """The hash value is used in :py:obj:`set`, for example, when an object
        is added to the set.  The hash value is also used in other contexts,
        e.g. when checking for equality to identify identical plugins from
        different sources (name collisions)."""
 
        return id(self)
 
    def __eq__(self, other):
        """py:obj:`Plugin` objects are equal if the hash values of the two
        objects are equal."""
 
        return hash(self) == hash(other)
 
    def init(self, app: "flask.Flask") -> bool:  # pylint: disable=unused-argument
        """Initialization of the plugin, the return value decides whether this
        plugin is active or not.  Initialization only takes place once, at the
        time the WEB application is set up.  The base methode always returns
        ``True``, the methode can be overwritten in the inheritances,
 
        - ``True`` plugin is active
        - ``False`` plugin is inactive
        """
        return True
 
    # pylint: disable=unused-argument
    def pre_search(self, request: SXNG_Request, search: "SearchWithPlugins") -> bool:
        """Runs BEFORE the search request and returns a boolean:
 
        - ``True`` to continue the search
        - ``False`` to stop the search
        """
        return True
 
    def on_result(self, request: SXNG_Request, search: "SearchWithPlugins", result: Result) -> bool:
        """Runs for each result of each engine and returns a boolean:
 
        - ``True`` to keep the result
        - ``False`` to remove the result from the result list
 
        The ``result`` can be modified to the needs.
 
        .. hint::
 
           If :py:obj:`Result.url <searx.result_types._base.Result.url>` is modified,
           :py:obj:`Result.parsed_url <searx.result_types._base.Result.parsed_url>` must
           be changed accordingly:
 
           .. code:: python
 
              result["parsed_url"] = urlparse(result["url"])
        """
        return True
 
    def post_search(self, request: SXNG_Request, search: "SearchWithPlugins") -> None | typing.Sequence[Result]:
        """Runs AFTER the search request.  Can return a list of
        :py:obj:`Result <searx.result_types._base.Result>` objects to be added to the
        final result list."""
        return
 
 
@dataclass
class PluginCfg:
    """Settings of a plugin.
 
    .. code:: yaml
 
       mypackage.mymodule.MyPlugin:
         active: true
    """
 
    active: bool = False
    """Plugin is active by default and the user can *opt-out* in the preferences."""
 
 
class PluginStorage:
    """A storage for managing the *plugins* of SearXNG."""
 
    plugin_list: set[Plugin]
    """The list of :py:obj:`Plugins` in this storage."""
 
    def __init__(self):
        self.plugin_list = set()
 
    def __iter__(self):
        yield from self.plugin_list
 
    def __len__(self):
        return len(self.plugin_list)
 
    @property
    def info(self) -> list[PluginInfo]:
 
        return [p.info for p in self.plugin_list]
 
    def load_settings(self, cfg: dict[str, dict]):
        """Load plugins configured in SearXNG's settings :ref:`settings
        plugins`."""
 
        for fqn, plg_settings in cfg.items():
            cls = None
            mod_name, cls_name = fqn.rsplit('.', 1)
            try:
                mod = importlib.import_module(mod_name)
                cls = getattr(mod, cls_name, None)
            except Exception as exc:  # pylint: disable=broad-exception-caught
                log.exception(exc)
 
            if cls is None:
                msg = f"plugin {fqn} is not implemented"
                raise ValueError(msg)
            plg = cls(PluginCfg(**plg_settings))
            self.register(plg)
 
    def register(self, plugin: Plugin):
        """Register a :py:obj:`Plugin`.  In case of name collision (if two
        plugins have same ID) a :py:obj:`KeyError` exception is raised.
        """
 
        if plugin in [p.id for p in self.plugin_list]:
            msg = f"name collision '{plugin.id}'"
            plugin.log.critical(msg)
            raise KeyError(msg)
 
        self.plugin_list.add(plugin)
        plugin.log.debug("plugin has been loaded")
 
    def init(self, app: "flask.Flask") -> None:
        """Calls the method :py:obj:`Plugin.init` of each plugin in this
        storage.  Depending on its return value, the plugin is removed from
        *this* storage or not."""
 
        for plg in self.plugin_list.copy():
            if not plg.init(app):
                self.plugin_list.remove(plg)
 
    def pre_search(self, request: SXNG_Request, search: "SearchWithPlugins") -> bool:
 
        ret = True
        for plugin in [p for p in self.plugin_list if p.id in search.user_plugins]:
            try:
                ret = bool(plugin.pre_search(request=request, search=search))
            except Exception:  # pylint: disable=broad-except
                plugin.log.exception("Exception while calling pre_search")
                continue
            if not ret:
                # skip this search on the first False from a plugin
                break
        return ret
 
    def on_result(self, request: SXNG_Request, search: "SearchWithPlugins", result: Result) -> bool:
 
        ret = True
        for plugin in [p for p in self.plugin_list if p.id in search.user_plugins]:
            try:
                ret = bool(plugin.on_result(request=request, search=search, result=result))
            except Exception:  # pylint: disable=broad-except
                plugin.log.exception("Exception while calling on_result")
                continue
            if not ret:
                # ignore this result item on the first False from a plugin
                break
 
        return ret
 
    def post_search(self, request: SXNG_Request, search: "SearchWithPlugins") -> None:
        """Extend :py:obj:`search.result_container
        <searx.results.ResultContainer`> with result items from plugins listed
        in :py:obj:`search.user_plugins <SearchWithPlugins.user_plugins>`.
        """
 
        keyword = None
        for keyword in search.search_query.query.split():
            if keyword:
                break
 
        for plugin in [p for p in self.plugin_list if p.id in search.user_plugins]:
 
            if plugin.keywords:
                # plugin with keywords: skip plugin if no keyword match
                if keyword and keyword not in plugin.keywords:
                    continue
            try:
                results = plugin.post_search(request=request, search=search) or []
            except Exception:  # pylint: disable=broad-except
                plugin.log.exception("Exception while calling post_search")
                continue
 
            # In case of *plugins* prefix ``plugin:`` is set, see searx.result_types.Result
            search.result_container.extend(f"plugin: {plugin.id}", results)