.oO SearXNG Developer Documentation Oo.
Loading...
Searching...
No Matches
_core.py
Go to the documentation of this file.
1# SPDX-License-Identifier: AGPL-3.0-or-later
2# pylint: disable=too-few-public-methods, missing-module-docstring
3
4
5import abc
6import importlib
7import logging
8import pathlib
9import warnings
10
11from dataclasses import dataclass
12
13from searx.utils import load_module
14from searx.result_types.answer import BaseAnswer
15
16
17_default = pathlib.Path(__file__).parent
18log: logging.Logger = logging.getLogger("searx.answerers")
19
20
21@dataclass
23 """Object that holds information about an answerer, these infos are shown
24 to the user in the Preferences menu.
25
26 To be able to translate the information into other languages, the text must
27 be written in English and translated with :py:obj:`flask_babel.gettext`.
28 """
29
30 name: str
31 """Name of the *answerer*."""
32
33 description: str
34 """Short description of the *answerer*."""
35
36 examples: list[str]
37 """List of short examples of the usage / of query terms."""
38
39 keywords: list[str]
40 """See :py:obj:`Answerer.keywords`"""
41
42
43class Answerer(abc.ABC):
44 """Abstract base class of answerers."""
45
46 keywords: list[str]
47 """Keywords to which the answerer has *answers*."""
48
49 @abc.abstractmethod
50 def answer(self, query: str) -> list[BaseAnswer]:
51 """Function that returns a list of answers to the question/query."""
52
53 @abc.abstractmethod
54 def info(self) -> AnswererInfo:
55 """Information about the *answerer*, see :py:obj:`AnswererInfo`."""
56
57
58class ModuleAnswerer(Answerer):
59 """A wrapper class for legacy *answerers* where the names (keywords, answer,
60 info) are implemented on the module level (not in a class).
61
62 .. note::
63
64 For internal use only!
65 """
66
67 def __init__(self, mod):
68
69 for name in ["keywords", "self_info", "answer"]:
70 if not getattr(mod, name, None):
71 raise SystemExit(2)
72 if not isinstance(mod.keywords, tuple):
73 raise SystemExit(2)
74
75 self.module = mod
76 self.keywords = mod.keywords # type: ignore
77
78 def answer(self, query: str) -> list[BaseAnswer]:
79 return self.module.answer(query)
80
81 def info(self) -> AnswererInfo:
82 kwargs = self.module.self_info()
83 kwargs["keywords"] = self.keywords
84 return AnswererInfo(**kwargs)
85
86
87class AnswerStorage(dict): # type: ignore
88 """A storage for managing the *answerers* of SearXNG. With the
89 :py:obj:`AnswerStorage.ask`” method, a caller can ask questions to all
90 *answerers* and receives a list of the results."""
91
92 answerer_list: set[Answerer]
93 """The list of :py:obj:`Answerer` in this storage."""
94
95 def __init__(self):
96 super().__init__()
97 self.answerer_list = set()
98
99 def load_builtins(self):
100 """Loads ``answerer.py`` modules from the python packages in
101 :origin:`searx/answerers`. The python modules are wrapped by
102 :py:obj:`ModuleAnswerer`."""
103
104 for f in _default.iterdir():
105 if f.name.startswith("_"):
106 continue
107
108 if f.is_file() and f.suffix == ".py":
109 self.register_by_fqn(f"searx.answerers.{f.stem}.SXNGAnswerer")
110 continue
111
112 # for backward compatibility (if a fork has additional answerers)
113
114 if f.is_dir() and (f / "answerer.py").exists():
115 warnings.warn(
116 f"answerer module {f} is deprecated / migrate to searx.answerers.Answerer", DeprecationWarning
117 )
118 mod = load_module("answerer.py", str(f))
119 self.register(ModuleAnswerer(mod))
120
121 def register_by_fqn(self, fqn: str):
122 """Register a :py:obj:`Answerer` via its fully qualified class namen(FQN)."""
123
124 mod_name, _, obj_name = fqn.rpartition('.')
125 mod = importlib.import_module(mod_name)
126 code_obj = getattr(mod, obj_name, None)
127
128 if code_obj is None:
129 msg = f"answerer {fqn} is not implemented"
130 log.critical(msg)
131 raise ValueError(msg)
132
133 self.register(code_obj())
134
135 def register(self, answerer: Answerer):
136 """Register a :py:obj:`Answerer`."""
137
138 self.answerer_list.add(answerer)
139 for _kw in answerer.keywords:
140 self[_kw] = self.get(_kw, [])
141 self[_kw].append(answerer)
142
143 def ask(self, query: str) -> list[BaseAnswer]:
144 """An answerer is identified via keywords, if there is a keyword at the
145 first position in the ``query`` for which there is one or more
146 answerers, then these are called, whereby the entire ``query`` is passed
147 as argument to the answerer function."""
148
149 results = []
150 keyword = None
151 for keyword in query.split():
152 if keyword:
153 break
154
155 if not keyword or keyword not in self:
156 return results
157
158 for answerer in self[keyword]:
159 for answer in answerer.answer(query):
160 # In case of *answers* prefix ``answerer:`` is set, see searx.result_types.Result
161 answer.engine = f"answerer: {keyword}"
162 results.append(answer)
163
164 return results
165
166 @property
167 def info(self) -> list[AnswererInfo]:
168 return [a.info() for a in self.answerer_list]
register(self, Answerer answerer)
Definition _core.py:135
list[AnswererInfo] info(self)
Definition _core.py:167
list[BaseAnswer] ask(self, str query)
Definition _core.py:143
AnswererInfo info(self)
Definition _core.py:54
list[BaseAnswer] answer(self, str query)
Definition _core.py:50
list[BaseAnswer] answer(self, str query)
Definition _core.py:78