# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=too-few-public-methods,missing-module-docstring
from __future__ import annotations
__all__ = ["PluginInfo", "Plugin", "PluginStorage"]
import abc
import importlib
import logging
import pathlib
import types
import typing
import warnings
from dataclasses import dataclass, field
import flask
import searx
from searx.utils import load_module
from searx.extended_types import SXNG_Request
from searx.result_types import Result
if typing.TYPE_CHECKING:
from searx.search import SearchWithPlugins
_default = pathlib.Path(__file__).parent
log: logging.Logger = logging.getLogger("searx.plugins")
[docs]
@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`"""
[docs]
class Plugin(abc.ABC):
"""Abstract base class of all Plugins."""
id: typing.ClassVar[str]
"""The ID (suffix) in the HTML form."""
default_on: typing.ClassVar[bool]
"""Plugin is enabled/disabled by default."""
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`."""
def __init__(self) -> None:
super().__init__()
for attr in ["id", "default_on"]:
if getattr(self, attr, None) is None:
raise NotImplementedError(f"plugin {self} is missing attribute {attr}")
if not self.id:
self.id = f"{self.__class__.__module__}.{self.__class__.__name__}"
if not getattr(self, "log", None):
self.log = log.getChild(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)
[docs]
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
[docs]
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
[docs]
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` is modified, :py:obj:`Result.parsed_url` must
be changed accordingly:
.. code:: python
result["parsed_url"] = urlparse(result["url"])
"""
return True
[docs]
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`
objects to be added to the final result list."""
return
[docs]
class ModulePlugin(Plugin):
"""A wrapper class for legacy *plugins*.
.. note::
For internal use only!
In a module plugin, the follwing names are mapped:
- `module.query_keywords` --> :py:obj:`Plugin.keywords`
- `module.plugin_id` --> :py:obj:`Plugin.id`
- `module.logger` --> :py:obj:`Plugin.log`
"""
_required_attrs = (("name", str), ("description", str), ("default_on", bool))
def __init__(self, mod: types.ModuleType):
"""In case of missing attributes in the module or wrong types are given,
a :py:obj:`TypeError` exception is raised."""
self.module = mod
self.id = getattr(self.module, "plugin_id", self.module.__name__)
self.log = logging.getLogger(self.module.__name__)
self.keywords = getattr(self.module, "query_keywords", [])
for attr, attr_type in self._required_attrs:
if not hasattr(self.module, attr):
msg = f"missing attribute {attr}, cannot load plugin"
self.log.critical(msg)
raise TypeError(msg)
if not isinstance(getattr(self.module, attr), attr_type):
msg = f"attribute {attr} is not of type {attr_type}"
self.log.critical(msg)
raise TypeError(msg)
self.default_on = mod.default_on
self.info = PluginInfo(
id=self.id,
name=self.module.name,
description=self.module.description,
preference_section=getattr(self.module, "preference_section", None),
examples=getattr(self.module, "query_examples", []),
keywords=self.keywords,
)
# monkeypatch module
self.module.logger = self.log # type: ignore
super().__init__()
[docs]
def init(self, app: flask.Flask) -> bool:
if not hasattr(self.module, "init"):
return True
return self.module.init(app)
[docs]
def pre_search(self, request: SXNG_Request, search: "SearchWithPlugins") -> bool:
if not hasattr(self.module, "pre_search"):
return True
return self.module.pre_search(request, search)
[docs]
def on_result(self, request: SXNG_Request, search: "SearchWithPlugins", result: Result) -> bool:
if not hasattr(self.module, "on_result"):
return True
return self.module.on_result(request, search, result)
[docs]
def post_search(self, request: SXNG_Request, search: "SearchWithPlugins") -> None | list[Result]:
if not hasattr(self.module, "post_search"):
return None
return self.module.post_search(request, search)
[docs]
class PluginStorage:
"""A storage for managing the *plugins* of SearXNG."""
plugin_list: set[Plugin]
"""The list of :py:obj:`Plugins` in this storage."""
legacy_plugins = [
"ahmia_filter",
"calculator",
"hostnames",
"oa_doi_rewrite",
"tor_check",
"tracker_url_remover",
"unit_converter",
]
"""Internal plugins implemented in the legacy style (as module / deprecated!)."""
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]
[docs]
def load_builtins(self):
"""Load plugin modules from:
- the python packages in :origin:`searx/plugins` and
- the external plugins from :ref:`settings plugins`.
"""
for f in _default.iterdir():
if f.name.startswith("_"):
continue
if f.stem not in self.legacy_plugins:
self.register_by_fqn(f"searx.plugins.{f.stem}.SXNGPlugin")
continue
# for backward compatibility
mod = load_module(f.name, str(f.parent))
self.register(ModulePlugin(mod))
for fqn in searx.get_setting("plugins"): # type: ignore
self.register_by_fqn(fqn)
[docs]
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 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")
[docs]
def register_by_fqn(self, fqn: str):
"""Register a :py:obj:`Plugin` via its fully qualified class name (FQN).
The FQNs of external plugins could be read from a configuration, for
example, and registered using this method
"""
mod_name, _, obj_name = fqn.rpartition('.')
if not mod_name:
# for backward compatibility
code_obj = importlib.import_module(fqn)
else:
mod = importlib.import_module(mod_name)
code_obj = getattr(mod, obj_name, None)
if code_obj is None:
msg = f"plugin {fqn} is not implemented"
log.critical(msg)
raise ValueError(msg)
if isinstance(code_obj, types.ModuleType):
# for backward compatibility
warnings.warn(
f"plugin {fqn} is implemented in a legacy module / migrate to searx.plugins.Plugin", DeprecationWarning
)
self.register(ModulePlugin(code_obj))
return
self.register(code_obj())
[docs]
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
[docs]
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)
