3 Commits

Author SHA1 Message Date
Markus Heiser
f91c998aa0 [docs] fix some typos in the description of result class Code (#5174)
Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>
2025-09-03 14:53:01 +02:00
Markus Heiser
f24d85bc4b [mod] drop: from __future__ import annotations
Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>
2025-09-03 13:37:36 +02:00
Markus Heiser
57b9673efb [mod] addition of various type hints / tbc
- pyright configuration [1]_
- stub files: types-lxml [2]_
- addition of various type hints
- enable use of new type system features on older Python versions [3]_
- ``.tool-versions`` - set python to lowest version we support (3.10.18) [4]_:
  Older versions typically lack some typing features found in newer Python
  versions.  Therefore, for local type checking (before commit), it is necessary
  to use the older Python interpreter.

.. [1] https://docs.basedpyright.com/v1.20.0/configuration/config-files/
.. [2] https://pypi.org/project/types-lxml/
.. [3] https://typing-extensions.readthedocs.io/en/latest/#
.. [4] https://mise.jdx.dev/configuration.html#tool-versions

Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>
Format: reST
2025-09-03 13:37:36 +02:00
149 changed files with 1229 additions and 1331 deletions

View File

@@ -10,7 +10,7 @@ trim_trailing_whitespace = true
end_of_line = lf
charset = utf-8
[*.py]
[{*.py,*.pyi}]
# code formatter accepts length of 120, but editor should prefer 80
max_line_length = 80

View File

@@ -311,7 +311,7 @@ dummy-variables-rgx=_+$|(_[a-zA-Z0-9_]*[a-zA-Z0-9]+?$)|dummy|^ignored_|^unused_
ignored-argument-names=_.*|^ignored_|^unused_
# Tells whether we should check for unused import in __init__ files.
init-import=no
init-import=yes
# List of qualified module names which can have objects that can redefine
# builtins.

View File

@@ -1,4 +1,4 @@
nodejs 24.3.0
python 3.13.1
nodejs 24.3.0
python 3.10.18
shellcheck 0.10.0
sqlite 3.47.2
sqlite 3.47.2

View File

@@ -151,6 +151,7 @@ intersphinx_mapping = {
"sphinx" : ("https://www.sphinx-doc.org/en/master/", None),
"valkey": ('https://valkey-py.readthedocs.io/en/stable/', None),
"pygments": ("https://pygments.org/", None),
"lxml": ('https://lxml.de/apidoc', None),
}
issues_github_path = "searxng/searxng"

View File

@@ -4,10 +4,10 @@
Search
======
.. autoclass:: searx.search.EngineRef
.. autoclass:: searx.search.models.EngineRef
:members:
.. autoclass:: searx.search.SearchQuery
.. autoclass:: searx.search.models.SearchQuery
:members:
.. autoclass:: searx.search.Search

View File

@@ -6,14 +6,21 @@
"searxng_extra",
"tests"
],
"reportAny" : "information",
"enableTypeIgnoreComments": true,
"reportIgnoreCommentWithoutRule": true,
"reportConstantRedefinition": false,
"reportIgnoreCommentWithoutRule": "information",
"reportImplicitOverride": false,
"reportImplicitStringConcatenation": false,
"reportImportCycles": "warning",
"reportMissingTypeStubs": "information",
"reportUninitializedInstanceVariable": false,
"reportUnnecessaryIsInstance": false,
"reportUnnecessaryTypeIgnoreComment": "error",
"reportUnreachable": "information",
"reportUnusedCallResult": false,
"enableTypeIgnoreComments": true,
"executionEnvironments": [
{
"root": "searx",

View File

@@ -23,3 +23,4 @@ docutils>=0.21.2
parameterized==0.9.0
granian[reload]==2.5.1
basedpyright==1.31.3
types-lxml==2025.3.30

View File

@@ -20,3 +20,4 @@ msgspec==0.19.0
typer-slim==0.16.1
isodate==0.7.2
whitenoise==6.9.0
typing-extensions==4.14.1

View File

@@ -1,28 +1,29 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring, cyclic-import
from __future__ import annotations
import typing as t
import sys
import os
from os.path import dirname, abspath
import logging
import searx.unixthreadname
import searx.settings_loader
from searx.settings_defaults import SCHEMA, apply_schema
import searx.unixthreadname # pylint: disable=unused-import
# Debug
LOG_FORMAT_DEBUG = '%(levelname)-7s %(name)-30.30s: %(message)s'
LOG_FORMAT_DEBUG: str = '%(levelname)-7s %(name)-30.30s: %(message)s'
# Production
LOG_FORMAT_PROD = '%(asctime)-15s %(levelname)s:%(name)s: %(message)s'
LOG_FORMAT_PROD: str = '%(asctime)-15s %(levelname)s:%(name)s: %(message)s'
LOG_LEVEL_PROD = logging.WARNING
searx_dir = abspath(dirname(__file__))
searx_parent_dir = abspath(dirname(dirname(__file__)))
searx_dir: str = abspath(dirname(__file__))
searx_parent_dir: str = abspath(dirname(dirname(__file__)))
settings = {}
sxng_debug = False
settings: dict[str, t.Any] = {}
sxng_debug: bool = False
logger = logging.getLogger('searx')
_unset = object()
@@ -33,9 +34,13 @@ def init_settings():
``logger`` from ``SEARXNG_SETTINGS_PATH``.
"""
# pylint: disable=import-outside-toplevel
from searx import settings_loader
from searx.settings_defaults import SCHEMA, apply_schema
global settings, sxng_debug # pylint: disable=global-variable-not-assigned
cfg, msg = searx.settings_loader.load_settings(load_user_settings=True)
cfg, msg = settings_loader.load_settings(load_user_settings=True)
cfg = cfg or {}
apply_schema(cfg, SCHEMA, [])
@@ -52,7 +57,7 @@ def init_settings():
logger.info(msg)
# log max_request_timeout
max_request_timeout = settings['outgoing']['max_request_timeout']
max_request_timeout: int | None = settings['outgoing']['max_request_timeout']
if max_request_timeout is None:
logger.info('max_request_timeout=%s', repr(max_request_timeout))
else:
@@ -66,22 +71,22 @@ def init_settings():
)
def get_setting(name, default=_unset):
def get_setting(name: str, default: t.Any = _unset) -> t.Any:
"""Returns the value to which ``name`` point. If there is no such name in the
settings and the ``default`` is unset, a :py:obj:`KeyError` is raised.
"""
value = settings
value: dict[str, t.Any] = settings
for a in name.split('.'):
if isinstance(value, dict):
value = value.get(a, _unset)
else:
value = _unset
value = _unset # type: ignore
if value is _unset:
if default is _unset:
raise KeyError(name)
value = default
value = default # type: ignore
break
return value
@@ -119,9 +124,14 @@ def _logging_config_debug():
'programname': {'color': 'cyan'},
'username': {'color': 'yellow'},
}
coloredlogs.install(level=log_level, level_styles=level_styles, field_styles=field_styles, fmt=LOG_FORMAT_DEBUG)
coloredlogs.install( # type: ignore
level=log_level,
level_styles=level_styles,
field_styles=field_styles,
fmt=LOG_FORMAT_DEBUG,
)
else:
logging.basicConfig(level=logging.getLevelName(log_level), format=LOG_FORMAT_DEBUG)
logging.basicConfig(level=getattr(logging, log_level, "ERROR"), format=LOG_FORMAT_DEBUG)
init_settings()

View File

@@ -38,7 +38,6 @@ area:
"""
from __future__ import annotations
__all__ = ["AnswererInfo", "Answerer", "AnswerStorage"]

View File

@@ -1,7 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=too-few-public-methods, missing-module-docstring
from __future__ import annotations
import abc
import importlib
@@ -85,7 +84,7 @@ class ModuleAnswerer(Answerer):
return AnswererInfo(**kwargs)
class AnswerStorage(dict):
class AnswerStorage(dict): # type: ignore
"""A storage for managing the *answerers* of SearXNG. With the
:py:obj:`AnswerStorage.ask`” method, a caller can ask questions to all
*answerers* and receives a list of the results."""

View File

@@ -1,7 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring
from __future__ import annotations
import hashlib
import random

View File

@@ -1,6 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring
from __future__ import annotations
from functools import reduce
from operator import mul

View File

@@ -6,109 +6,105 @@
import json
import html
import typing as t
from urllib.parse import urlencode, quote_plus
import lxml.etree
import lxml.html
from httpx import HTTPError
from searx.extended_types import SXNG_Response
from searx import settings
from searx.engines import (
engines,
google,
)
from searx.network import get as http_get, post as http_post
from searx.network import get as http_get, post as http_post # pyright: ignore[reportUnknownVariableType]
from searx.exceptions import SearxEngineResponseException
from searx.utils import extr, gen_useragent
if t.TYPE_CHECKING:
from searx.extended_types import SXNG_Response
def update_kwargs(**kwargs):
def update_kwargs(**kwargs) -> None: # type: ignore
if 'timeout' not in kwargs:
kwargs['timeout'] = settings['outgoing']['request_timeout']
kwargs['raise_for_httperror'] = True
def get(*args, **kwargs) -> SXNG_Response:
update_kwargs(**kwargs)
return http_get(*args, **kwargs)
def get(*args, **kwargs) -> "SXNG_Response": # type: ignore
update_kwargs(**kwargs) # pyright: ignore[reportUnknownArgumentType]
return http_get(*args, **kwargs) # pyright: ignore[reportUnknownArgumentType]
def post(*args, **kwargs) -> SXNG_Response:
update_kwargs(**kwargs)
return http_post(*args, **kwargs)
def post(*args, **kwargs) -> "SXNG_Response": # type: ignore
update_kwargs(**kwargs) # pyright: ignore[reportUnknownArgumentType]
return http_post(*args, **kwargs) # pyright: ignore[reportUnknownArgumentType]
def baidu(query, _lang):
def baidu(query: str, _sxng_locale: str) -> list[str]:
# baidu search autocompleter
base_url = "https://www.baidu.com/sugrec?"
response = get(base_url + urlencode({'ie': 'utf-8', 'json': 1, 'prod': 'pc', 'wd': query}))
results = []
results: list[str] = []
if response.ok:
data = response.json()
data: dict[str, t.Any] = response.json()
if 'g' in data:
for item in data['g']:
results.append(item['q'])
return results
def brave(query, _lang):
def brave(query: str, _sxng_locale: str) -> list[str]:
# brave search autocompleter
url = 'https://search.brave.com/api/suggest?'
url += urlencode({'q': query})
country = 'all'
# if lang in _brave:
# country = lang
kwargs = {'cookies': {'country': country}}
resp = get(url, **kwargs)
results = []
results: list[str] = []
if resp.ok:
data = resp.json()
data: list[list[str]] = resp.json()
for item in data[1]:
results.append(item)
return results
def dbpedia(query, _lang):
# dbpedia autocompleter, no HTTPS
def dbpedia(query: str, _sxng_locale: str) -> list[str]:
autocomplete_url = 'https://lookup.dbpedia.org/api/search.asmx/KeywordSearch?'
resp = get(autocomplete_url + urlencode(dict(QueryString=query)))
results: list[str] = []
response = get(autocomplete_url + urlencode(dict(QueryString=query)))
results = []
if response.ok:
dom = lxml.etree.fromstring(response.content)
results = dom.xpath('//Result/Label//text()')
if resp.ok:
dom = lxml.etree.fromstring(resp.content)
results = [str(x) for x in dom.xpath('//Result/Label//text()')]
return results
def duckduckgo(query, sxng_locale):
def duckduckgo(query: str, sxng_locale: str) -> list[str]:
"""Autocomplete from DuckDuckGo. Supports DuckDuckGo's languages"""
traits = engines['duckduckgo'].traits
args = {
args: dict[str, str] = {
'q': query,
'kl': traits.get_region(sxng_locale, traits.all_locale),
}
url = 'https://duckduckgo.com/ac/?type=list&' + urlencode(args)
resp = get(url)
results: list[str] = []
ret_val = []
if resp.ok:
j = resp.json()
if len(j) > 1:
ret_val = j[1]
return ret_val
results = j[1]
return results
def google_complete(query, sxng_locale):
def google_complete(query: str, sxng_locale: str) -> list[str]:
"""Autocomplete from Google. Supports Google's languages and subdomains
(:py:obj:`searx.engines.google.get_google_info`) by using the async REST
API::
@@ -117,8 +113,7 @@ def google_complete(query, sxng_locale):
"""
google_info = google.get_google_info({'searxng_locale': sxng_locale}, engines['google'].traits)
google_info: dict[str, t.Any] = google.get_google_info({'searxng_locale': sxng_locale}, engines['google'].traits)
url = 'https://{subdomain}/complete/search?{args}'
args = urlencode(
{
@@ -127,7 +122,8 @@ def google_complete(query, sxng_locale):
'hl': google_info['params']['hl'],
}
)
results = []
results: list[str] = []
resp = get(url.format(subdomain=google_info['subdomain'], args=args))
if resp and resp.ok:
json_txt = resp.text[resp.text.find('[') : resp.text.find(']', -3) + 1]
@@ -137,54 +133,51 @@ def google_complete(query, sxng_locale):
return results
def mwmbl(query, _lang):
def mwmbl(query: str, _sxng_locale: str) -> list[str]:
"""Autocomplete from Mwmbl_."""
# mwmbl autocompleter
url = 'https://api.mwmbl.org/search/complete?{query}'
results = get(url.format(query=urlencode({'q': query}))).json()[1]
results: list[str] = get(url.format(query=urlencode({'q': query}))).json()[1]
# results starting with `go:` are direct urls and not useful for auto completion
return [result for result in results if not result.startswith("go: ") and not result.startswith("search: ")]
def naver(query, _lang):
def naver(query: str, _sxng_locale: str) -> list[str]:
# Naver search autocompleter
url = f"https://ac.search.naver.com/nx/ac?{urlencode({'q': query, 'r_format': 'json', 'st': 0})}"
response = get(url)
results = []
results: list[str] = []
if response.ok:
data = response.json()
data: dict[str, t.Any] = response.json()
if data.get('items'):
for item in data['items'][0]:
results.append(item[0])
return results
def qihu360search(query, _lang):
def qihu360search(query: str, _sxng_locale: str) -> list[str]:
# 360Search search autocompleter
url = f"https://sug.so.360.cn/suggest?{urlencode({'format': 'json', 'word': query})}"
response = get(url)
results = []
results: list[str] = []
if response.ok:
data = response.json()
data: dict[str, t.Any] = response.json()
if 'result' in data:
for item in data['result']:
results.append(item['word'])
return results
def quark(query, _lang):
def quark(query: str, _sxng_locale: str) -> list[str]:
# Quark search autocompleter
url = f"https://sugs.m.sm.cn/web?{urlencode({'q': query})}"
response = get(url)
results = []
results: list[str] = []
if response.ok:
data = response.json()
@@ -193,10 +186,9 @@ def quark(query, _lang):
return results
def seznam(query, _lang):
def seznam(query: str, _sxng_locale: str) -> list[str]:
# seznam search autocompleter
url = 'https://suggest.seznam.cz/fulltext/cs?{query}'
resp = get(
url.format(
query=urlencode(
@@ -204,36 +196,35 @@ def seznam(query, _lang):
)
)
)
results: list[str] = []
if not resp.ok:
return []
data = resp.json()
return [
''.join([part.get('text', '') for part in item.get('text', [])])
for item in data.get('result', [])
if item.get('itemType', None) == 'ItemType.TEXT'
]
if resp.ok:
data = resp.json()
results = [
''.join([part.get('text', '') for part in item.get('text', [])])
for item in data.get('result', [])
if item.get('itemType', None) == 'ItemType.TEXT'
]
return results
def sogou(query, _lang):
def sogou(query: str, _sxng_locale: str) -> list[str]:
# Sogou search autocompleter
base_url = "https://sor.html5.qq.com/api/getsug?"
response = get(base_url + urlencode({'m': 'searxng', 'key': query}))
if response.ok:
raw_json = extr(response.text, "[", "]", default="")
resp = get(base_url + urlencode({'m': 'searxng', 'key': query}))
results: list[str] = []
if resp.ok:
raw_json = extr(resp.text, "[", "]", default="")
try:
data = json.loads(f"[{raw_json}]]")
return data[1]
results = data[1]
except json.JSONDecodeError:
return []
return []
pass
return results
def startpage(query, sxng_locale):
def startpage(query: str, sxng_locale: str) -> list[str]:
"""Autocomplete from Startpage's Firefox extension.
Supports the languages specified in lang_map.
"""
@@ -266,46 +257,44 @@ def startpage(query, sxng_locale):
h = {'User-Agent': gen_useragent()}
resp = get(url, headers=h)
results: list[str] = []
if resp.ok:
try:
data = resp.json()
if len(data) >= 2 and isinstance(data[1], list):
return data[1]
results = data[1]
except json.JSONDecodeError:
pass
return []
return results
def stract(query, _lang):
def stract(query: str, _sxng_locale: str) -> list[str]:
# stract autocompleter (beta)
url = f"https://stract.com/beta/api/autosuggest?q={quote_plus(query)}"
resp = post(url)
results: list[str] = []
if not resp.ok:
return []
if resp.ok:
results = [html.unescape(suggestion['raw']) for suggestion in resp.json()]
return [html.unescape(suggestion['raw']) for suggestion in resp.json()]
return results
def swisscows(query, _lang):
def swisscows(query: str, _sxng_locale: str) -> list[str]:
# swisscows autocompleter
url = 'https://swisscows.ch/api/suggest?{query}&itemsCount=5'
resp = json.loads(get(url.format(query=urlencode({'query': query}))).text)
return resp
results: list[str] = json.loads(get(url.format(query=urlencode({'query': query}))).text)
return results
def qwant(query, sxng_locale):
def qwant(query: str, sxng_locale: str) -> list[str]:
"""Autocomplete from Qwant. Supports Qwant's regions."""
results = []
locale = engines['qwant'].traits.get_region(sxng_locale, 'en_US')
url = 'https://api.qwant.com/v3/suggest?{query}'
resp = get(url.format(query=urlencode({'q': query, 'locale': locale, 'version': '2'})))
results: list[str] = []
if resp.ok:
data = resp.json()
@@ -316,14 +305,12 @@ def qwant(query, sxng_locale):
return results
def wikipedia(query, sxng_locale):
def wikipedia(query: str, sxng_locale: str) -> list[str]:
"""Autocomplete from Wikipedia. Supports Wikipedia's languages (aka netloc)."""
results = []
eng_traits = engines['wikipedia'].traits
wiki_lang = eng_traits.get_language(sxng_locale, 'en')
wiki_netloc = eng_traits.custom['wiki_netloc'].get(wiki_lang, 'en.wikipedia.org') # type: ignore
wiki_netloc: str = eng_traits.custom['wiki_netloc'].get(wiki_lang, 'en.wikipedia.org') # type: ignore
url = 'https://{wiki_netloc}/w/api.php?{args}'
args = urlencode(
{
'action': 'opensearch',
@@ -334,7 +321,9 @@ def wikipedia(query, sxng_locale):
'limit': '10',
}
)
resp = get(url.format(args=args, wiki_netloc=wiki_netloc))
resp = get(f'https://{wiki_netloc}/w/api.php?{args}')
results: list[str] = []
if resp.ok:
data = resp.json()
if len(data) > 1:
@@ -343,17 +332,18 @@ def wikipedia(query, sxng_locale):
return results
def yandex(query, _lang):
def yandex(query: str, _sxng_locale: str) -> list[str]:
# yandex autocompleter
url = "https://suggest.yandex.com/suggest-ff.cgi?{0}"
resp = json.loads(get(url.format(urlencode(dict(part=query)))).text)
results: list[str] = []
if len(resp) > 1:
return resp[1]
return []
results = resp[1]
return results
backends = {
backends: dict[str, t.Callable[[str, str], list[str]]] = {
'360search': qihu360search,
'baidu': baidu,
'brave': brave,
@@ -374,7 +364,7 @@ backends = {
}
def search_autocomplete(backend_name, query, sxng_locale):
def search_autocomplete(backend_name: str, query: str, sxng_locale: str) -> list[str]:
backend = backends.get(backend_name)
if backend is None:
return []

View File

@@ -4,7 +4,7 @@
Implementations used for bot detection.
"""
from __future__ import annotations
__all__ = ["init", "dump_request", "get_network", "too_many_requests", "ProxyFix"]

View File

@@ -1,6 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring, invalid-name
from __future__ import annotations
import typing as t
__all__ = ["log_error_only_once", "dump_request", "get_network", "logger", "too_many_requests"]
@@ -53,7 +53,7 @@ def too_many_requests(network: IPv4Network | IPv6Network, log_msg: str) -> werkz
return flask.make_response(('Too Many Requests', 429))
def get_network(real_ip: IPv4Address | IPv6Address, cfg: config.Config) -> IPv4Network | IPv6Network:
def get_network(real_ip: IPv4Address | IPv6Address, cfg: "config.Config") -> IPv4Network | IPv6Network:
"""Returns the (client) network of whether the ``real_ip`` is part of.
The ``ipv4_prefix`` and ``ipv6_prefix`` define the number of leading bits in
@@ -71,7 +71,7 @@ def get_network(real_ip: IPv4Address | IPv6Address, cfg: config.Config) -> IPv4N
prefix: int = cfg["botdetection.ipv4_prefix"]
if real_ip.version == 6:
prefix: int = cfg["botdetection.ipv6_prefix"]
prefix = cfg["botdetection.ipv6_prefix"]
network = ip_network(f"{real_ip}/{prefix}", strict=False)
# logger.debug("get_network(): %s", network.compressed)
return network

View File

@@ -6,7 +6,7 @@ The :py:class:`Config` class implements a configuration that is based on
structured dictionaries. The configuration schema is defined in a dictionary
structure and the configuration data is given in a dictionary structure.
"""
from __future__ import annotations
import typing
import copy
@@ -19,26 +19,27 @@ __all__ = ['Config', 'UNSET', 'SchemaIssue', 'set_global_cfg', 'get_global_cfg']
log = logging.getLogger(__name__)
CFG: Config | None = None
CFG: "Config | None" = None
"""Global config of the botdetection."""
def set_global_cfg(cfg: Config):
def set_global_cfg(cfg: "Config"):
global CFG # pylint: disable=global-statement
CFG = cfg
def get_global_cfg() -> Config:
def get_global_cfg() -> "Config":
if CFG is None:
raise ValueError("Botdetection's config is not yet initialized.")
return CFG
@typing.final
class FALSE:
"""Class of ``False`` singleton"""
# pylint: disable=multiple-statements
def __init__(self, msg):
def __init__(self, msg: str):
self.msg = msg
def __bool__(self):
@@ -53,6 +54,7 @@ class FALSE:
UNSET = FALSE('<UNSET>')
@typing.final
class SchemaIssue(ValueError):
"""Exception to store and/or raise a message from a schema issue."""
@@ -67,10 +69,10 @@ class SchemaIssue(ValueError):
class Config:
"""Base class used for configuration"""
UNSET = UNSET
UNSET: object = UNSET
@classmethod
def from_toml(cls, schema_file: pathlib.Path, cfg_file: pathlib.Path, deprecated: dict[str, str]) -> Config:
def from_toml(cls, schema_file: pathlib.Path, cfg_file: pathlib.Path, deprecated: dict[str, str]) -> "Config":
# init schema
@@ -102,9 +104,9 @@ class Config:
These values are needed for validation, see :py:obj:`validate`.
"""
self.cfg_schema = cfg_schema
self.deprecated = deprecated
self.cfg = copy.deepcopy(cfg_schema)
self.cfg_schema: dict[str, typing.Any] = cfg_schema
self.deprecated: dict[str, str] = deprecated
self.cfg: dict[str, typing.Any] = copy.deepcopy(cfg_schema)
def __getitem__(self, key: str) -> typing.Any:
return self.get(key)
@@ -115,7 +117,7 @@ class Config:
return validate(self.cfg_schema, cfg, self.deprecated)
def update(self, upd_cfg: dict):
def update(self, upd_cfg: dict[str, typing.Any]):
"""Update this configuration by ``upd_cfg``."""
dict_deepupdate(self.cfg, upd_cfg)
@@ -142,7 +144,7 @@ class Config:
val = val % self
return val
def set(self, name: str, val):
def set(self, name: str, val: typing.Any):
"""Set the value to which ``name`` points in the configuration.
If there is no such ``name`` in the config, a :py:obj:`KeyError` is
@@ -151,17 +153,17 @@ class Config:
parent = self._get_parent_dict(name)
parent[name.split('.')[-1]] = val
def _get_parent_dict(self, name):
def _get_parent_dict(self, name: str) -> dict[str, typing.Any]:
parent_name = '.'.join(name.split('.')[:-1])
if parent_name:
parent = value(parent_name, self.cfg)
parent: dict[str, typing.Any] = value(parent_name, self.cfg)
else:
parent = self.cfg
if (parent is UNSET) or (not isinstance(parent, dict)):
raise KeyError(parent_name)
return parent
def path(self, name: str, default=UNSET):
def path(self, name: str, default: typing.Any = UNSET):
"""Get a :py:class:`pathlib.Path` object from a config string."""
val = self.get(name, default)
@@ -171,7 +173,7 @@ class Config:
return default
return pathlib.Path(str(val))
def pyobj(self, name, default=UNSET):
def pyobj(self, name: str, default: typing.Any = UNSET):
"""Get python object referred by full qualiffied name (FQN) in the config
string."""
@@ -185,7 +187,7 @@ class Config:
return getattr(m, name)
def toml_load(file_name):
def toml_load(file_name: str | pathlib.Path):
try:
with open(file_name, "rb") as f:
return tomllib.load(f)
@@ -198,7 +200,7 @@ def toml_load(file_name):
# working with dictionaries
def value(name: str, data_dict: dict):
def value(name: str, data_dict: dict[str, typing.Any]):
"""Returns the value to which ``name`` points in the ``dat_dict``.
.. code: python
@@ -228,7 +230,7 @@ def value(name: str, data_dict: dict):
def validate(
schema_dict: dict[str, typing.Any], data_dict: dict[str, typing.Any], deprecated: dict[str, str]
) -> tuple[bool, list[str]]:
) -> tuple[bool, list[SchemaIssue]]:
"""Deep validation of dictionary in ``data_dict`` against dictionary in
``schema_dict``. Argument deprecated is a dictionary that maps deprecated
configuration names to a messages::
@@ -254,9 +256,9 @@ def validate(
:py:obj:`SchemaIssue` is raised.
"""
names = []
is_valid = True
issue_list = []
names: list[str] = []
is_valid: bool = True
issue_list: list[SchemaIssue] = []
if not isinstance(schema_dict, dict):
raise SchemaIssue('invalid', "schema_dict is not a dict type")
@@ -268,15 +270,16 @@ def validate(
def _validate(
names: typing.List,
issue_list: typing.List,
schema_dict: typing.Dict,
data_dict: typing.Dict,
deprecated: typing.Dict[str, str],
) -> typing.Tuple[bool, typing.List]:
names: list[str],
issue_list: list[SchemaIssue],
schema_dict: dict[str, typing.Any],
data_dict: dict[str, typing.Any],
deprecated: dict[str, str],
) -> tuple[bool, list[SchemaIssue]]:
is_valid = True
data_value: dict[str, typing.Any]
for key, data_value in data_dict.items():
names.append(key)
@@ -311,7 +314,7 @@ def _validate(
return is_valid, issue_list
def dict_deepupdate(base_dict: dict, upd_dict: dict, names=None):
def dict_deepupdate(base_dict: dict[str, typing.Any], upd_dict: dict[str, typing.Any], names: list[str] | None = None):
"""Deep-update of dictionary in ``base_dict`` by dictionary in ``upd_dict``.
For each ``upd_key`` & ``upd_val`` pair in ``upd_dict``:
@@ -350,7 +353,7 @@ def dict_deepupdate(base_dict: dict, upd_dict: dict, names=None):
raise TypeError(f"type mismatch {'.'.join(names)}: is not a dict type in base_dict")
dict_deepupdate(
base_dict[upd_key],
upd_val,
upd_val, # pyright: ignore[reportUnknownArgumentType]
names
+ [
upd_key,
@@ -359,7 +362,7 @@ def dict_deepupdate(base_dict: dict, upd_dict: dict, names=None):
else:
# if base_dict[upd_key] not exist, set base_dict[upd_key] from deepcopy of upd_val
base_dict[upd_key] = copy.deepcopy(upd_val)
base_dict[upd_key] = copy.deepcopy(upd_val) # pyright: ignore[reportUnknownArgumentType]
elif isinstance(upd_val, list):
@@ -373,7 +376,7 @@ def dict_deepupdate(base_dict: dict, upd_dict: dict, names=None):
else:
# if base_dict[upd_key] doesn't exists, set base_dict[key] from a deepcopy of the
# list in upd_val.
base_dict[upd_key] = copy.deepcopy(upd_val)
base_dict[upd_key] = copy.deepcopy(upd_val) # pyright: ignore[reportUnknownArgumentType]
elif isinstance(upd_val, set):

View File

@@ -13,7 +13,7 @@ Accept_ header ..
"""
from __future__ import annotations
from ipaddress import (
IPv4Network,
IPv6Network,

View File

@@ -14,7 +14,7 @@ bot if the Accept-Encoding_ header ..
"""
from __future__ import annotations
from ipaddress import (
IPv4Network,
IPv6Network,

View File

@@ -11,7 +11,7 @@ if the Accept-Language_ header is unset.
"""
from __future__ import annotations
from ipaddress import (
IPv4Network,
IPv6Network,

View File

@@ -11,7 +11,7 @@ the Connection_ header is set to ``close``.
"""
from __future__ import annotations
from ipaddress import (
IPv4Network,
IPv6Network,

View File

@@ -22,7 +22,7 @@ Metadata`_. A request is filtered out in case of:
"""
# pylint: disable=unused-argument
from __future__ import annotations
from ipaddress import (
IPv4Network,
IPv6Network,

View File

@@ -12,7 +12,7 @@ the User-Agent_ header is unset or matches the regular expression
"""
from __future__ import annotations
import re
from ipaddress import (
IPv4Network,

View File

@@ -36,7 +36,7 @@ dropped.
https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-For
"""
from __future__ import annotations
from ipaddress import (
IPv4Network,
IPv6Network,

View File

@@ -25,7 +25,7 @@ The ``ip_lists`` method implements :py:obj:`block-list <block_ip>` and
"""
# pylint: disable=unused-argument
from __future__ import annotations
from typing import Tuple
from ipaddress import (
ip_network,

View File

@@ -34,7 +34,7 @@ And in the HTML template from flask a stylesheet link is needed (the value of
https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-For
"""
from __future__ import annotations
from ipaddress import (
IPv4Network,
IPv6Network,

View File

@@ -3,7 +3,7 @@
(:py:obj:`flask.request.remote_addr`) behind a proxy chain."""
# pylint: disable=too-many-branches
from __future__ import annotations
import typing as t
from collections import abc
@@ -19,6 +19,7 @@ if t.TYPE_CHECKING:
from _typeshed.wsgi import WSGIEnvironment
@t.final
class ProxyFix:
"""A middleware like the ProxyFix_ class, where the ``x_for`` argument is
replaced by a method that determines the number of trusted proxies via the
@@ -54,7 +55,7 @@ class ProxyFix:
"""
def __init__(self, wsgi_app: WSGIApplication) -> None:
def __init__(self, wsgi_app: "WSGIApplication") -> None:
self.wsgi_app = wsgi_app
def trusted_proxies(self) -> list[IPv4Network | IPv6Network]:
@@ -84,7 +85,7 @@ class ProxyFix:
# fallback to first address
return x_forwarded_for[0].compressed
def __call__(self, environ: WSGIEnvironment, start_response: StartResponse) -> abc.Iterable[bytes]:
def __call__(self, environ: "WSGIEnvironment", start_response: "StartResponse") -> abc.Iterable[bytes]:
# pylint: disable=too-many-statements
trusted_proxies = self.trusted_proxies()

View File

@@ -1,7 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Providing a Valkey database for the botdetection methods."""
from __future__ import annotations
import valkey

View File

@@ -5,8 +5,6 @@
----
"""
from __future__ import annotations
__all__ = ["ExpireCacheCfg", "ExpireCacheStats", "ExpireCache", "ExpireCacheSQLite"]
import abc
@@ -64,7 +62,7 @@ class ExpireCacheCfg(msgspec.Struct): # pylint: disable=too-few-public-methods
if required.
"""
password: bytes = get_setting("server.secret_key").encode() # type: ignore
password: bytes = get_setting("server.secret_key").encode()
"""Password used by :py:obj:`ExpireCache.secret_hash`.
The default password is taken from :ref:`secret_key <server.secret_key>`.
@@ -101,7 +99,7 @@ class ExpireCacheStats:
def report(self):
c_ctx = 0
c_kv = 0
lines = []
lines: list[str] = []
for ctx_name, kv_list in self.cached_items.items():
c_ctx += 1
@@ -125,7 +123,7 @@ class ExpireCache(abc.ABC):
cfg: ExpireCacheCfg
hash_token = "hash_token"
hash_token: str = "hash_token"
@abc.abstractmethod
def set(self, key: str, value: typing.Any, expire: int | None, ctx: str | None = None) -> bool:
@@ -148,7 +146,7 @@ class ExpireCache(abc.ABC):
"""
@abc.abstractmethod
def get(self, key: str, default=None, ctx: str | None = None) -> typing.Any:
def get(self, key: str, default: typing.Any = None, ctx: str | None = None) -> typing.Any:
"""Return *value* of *key*. If key is unset, ``None`` is returned."""
@abc.abstractmethod
@@ -170,7 +168,7 @@ class ExpireCache(abc.ABC):
about the status of the cache."""
@staticmethod
def build_cache(cfg: ExpireCacheCfg) -> ExpireCache:
def build_cache(cfg: ExpireCacheCfg) -> "ExpireCacheSQLite":
"""Factory to build a caching instance.
.. note::
@@ -222,18 +220,18 @@ class ExpireCacheSQLite(sqlitedb.SQLiteAppl, ExpireCache):
- :py:obj:`ExpireCacheCfg.MAINTENANCE_MODE`
"""
DB_SCHEMA = 1
DB_SCHEMA: int = 1
# The key/value tables will be created on demand by self.create_table
DDL_CREATE_TABLES = {}
DDL_CREATE_TABLES: dict[str, str] = {}
CACHE_TABLE_PREFIX = "CACHE-TABLE"
CACHE_TABLE_PREFIX: str = "CACHE-TABLE"
def __init__(self, cfg: ExpireCacheCfg):
"""An instance of the SQLite expire cache is build up from a
:py:obj:`config <ExpireCacheCfg>`."""
self.cfg = cfg
self.cfg: ExpireCacheCfg = cfg
if cfg.db_url == ":memory:":
log.critical("don't use SQLite DB in :memory: in production!!")
super().__init__(cfg.db_url)
@@ -374,7 +372,7 @@ class ExpireCacheSQLite(sqlitedb.SQLiteAppl, ExpireCache):
return True
def get(self, key: str, default=None, ctx: str | None = None) -> typing.Any:
def get(self, key: str, default: typing.Any = None, ctx: str | None = None) -> typing.Any:
"""Get value of ``key`` from table given by argument ``ctx``. If
``ctx`` argument is ``None`` (the default), a table name is generated
from the :py:obj:`ExpireCacheCfg.name`. If ``key`` not exists (in
@@ -412,7 +410,7 @@ class ExpireCacheSQLite(sqlitedb.SQLiteAppl, ExpireCache):
yield row[0], self.deserialize(row[1])
def state(self) -> ExpireCacheStats:
cached_items = {}
cached_items: dict[str, list[tuple[str, typing.Any, int]]] = {}
for table in self.table_names:
cached_items[table] = []
for row in self.DB.execute(f"SELECT key, value, expire FROM {table}"):

View File

@@ -4,27 +4,53 @@
make data.all
"""
from __future__ import annotations
# pylint: disable=invalid-name
__all__ = ["ahmia_blacklist_loader"]
__all__ = ["ahmia_blacklist_loader", "data_dir", "get_cache"]
import json
import typing
import typing as t
from .core import log, data_dir
from .core import log, data_dir, get_cache
from .currencies import CurrenciesDB
from .tracker_patterns import TrackerPatternsDB
CURRENCIES: CurrenciesDB
USER_AGENTS: dict[str, typing.Any]
EXTERNAL_URLS: dict[str, typing.Any]
WIKIDATA_UNITS: dict[str, typing.Any]
EXTERNAL_BANGS: dict[str, typing.Any]
OSM_KEYS_TAGS: dict[str, typing.Any]
ENGINE_DESCRIPTIONS: dict[str, typing.Any]
ENGINE_TRAITS: dict[str, typing.Any]
LOCALES: dict[str, typing.Any]
class UserAgentType(t.TypedDict):
"""Data structure of ``useragents.json``"""
os: list[str]
ua: str
versions: list[str]
class WikiDataUnitType(t.TypedDict):
"""Data structure of an item in ``wikidata_units.json``"""
si_name: str
symbol: str
to_si_factor: float
class LocalesType(t.TypedDict):
"""Data structure of an item in ``locales.json``"""
LOCALE_NAMES: dict[str, str]
RTL_LOCALES: list[str]
USER_AGENTS: UserAgentType
WIKIDATA_UNITS: dict[str, WikiDataUnitType]
TRACKER_PATTERNS: TrackerPatternsDB
LOCALES: LocalesType
CURRENCIES: CurrenciesDB
EXTERNAL_URLS: dict[str, dict[str, dict[str, str | dict[str, str]]]]
EXTERNAL_BANGS: dict[str, dict[str, t.Any]]
OSM_KEYS_TAGS: dict[str, dict[str, t.Any]]
ENGINE_DESCRIPTIONS: dict[str, dict[str, t.Any]]
ENGINE_TRAITS: dict[str, dict[str, t.Any]]
lazy_globals = {
"CURRENCIES": CurrenciesDB(),
@@ -51,7 +77,7 @@ data_json_files = {
}
def __getattr__(name):
def __getattr__(name: str) -> t.Any:
# lazy init of the global objects
if name not in lazy_globals:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -68,7 +94,7 @@ def __getattr__(name):
return lazy_globals[name]
def ahmia_blacklist_loader():
def ahmia_blacklist_loader() -> list[str]:
"""Load data from `ahmia_blacklist.txt` and return a list of MD5 values of onion
names. The MD5 values are fetched by::

View File

@@ -1,6 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring
from __future__ import annotations
import pathlib
@@ -9,9 +9,9 @@ from searx.cache import ExpireCacheCfg, ExpireCacheSQLite
log = logger.getChild("data")
data_dir = pathlib.Path(__file__).parent
data_dir: pathlib.Path = pathlib.Path(__file__).parent
_DATA_CACHE: ExpireCacheSQLite = None # type: ignore
_DATA_CACHE: ExpireCacheSQLite | None = None
def get_cache():

View File

@@ -1,7 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Simple implementation to store currencies data in a SQL database."""
from __future__ import annotations
__all__ = ["CurrenciesDB"]

View File

@@ -1,7 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Simple implementation to store TrackerPatterns data in a SQL database."""
from __future__ import annotations
import typing
__all__ = ["TrackerPatternsDB"]

View File

@@ -22,21 +22,25 @@ an example in which the command line is called in the development environment::
-----
"""
from __future__ import annotations
__all__ = ["EngineCache", "Engine", "ENGINES_CACHE"]
from typing import List, Callable, TYPE_CHECKING, Any
import typing as t
import abc
from collections.abc import Callable
import logging
import string
import typer
from ..cache import ExpireCache, ExpireCacheCfg
from ..cache import ExpireCacheSQLite, ExpireCacheCfg
if TYPE_CHECKING:
if t.TYPE_CHECKING:
from searx.enginelib import traits
from searx.enginelib.traits import EngineTraits
from searx.extended_types import SXNG_Response
from searx.result_types import EngineResults
ENGINES_CACHE = ExpireCache.build_cache(
ENGINES_CACHE: ExpireCacheSQLite = ExpireCacheSQLite.build_cache(
ExpireCacheCfg(
name="ENGINES_CACHE",
MAXHOLD_TIME=60 * 60 * 24 * 7, # 7 days
@@ -62,7 +66,7 @@ def state():
title = f"properties of {ENGINES_CACHE.cfg.name}"
print(title)
print("=" * len(title))
print(str(ENGINES_CACHE.properties)) # type: ignore
print(str(ENGINES_CACHE.properties))
@app.command()
@@ -152,11 +156,11 @@ class EngineCache:
"""
def __init__(self, engine_name: str, expire: int | None = None):
self.expire = expire or ENGINES_CACHE.cfg.MAXHOLD_TIME
self.expire: int = expire or ENGINES_CACHE.cfg.MAXHOLD_TIME
_valid = "-_." + string.ascii_letters + string.digits
self.table_name = "".join([c if c in _valid else "_" for c in engine_name])
self.table_name: str = "".join([c if c in _valid else "_" for c in engine_name])
def set(self, key: str, value: Any, expire: int | None = None) -> bool:
def set(self, key: str, value: t.Any, expire: int | None = None) -> bool:
return ENGINES_CACHE.set(
key=key,
value=value,
@@ -164,14 +168,14 @@ class EngineCache:
ctx=self.table_name,
)
def get(self, key: str, default=None) -> Any:
def get(self, key: str, default: t.Any = None) -> t.Any:
return ENGINES_CACHE.get(key, default=default, ctx=self.table_name)
def secret_hash(self, name: str | bytes) -> str:
return ENGINES_CACHE.secret_hash(name=name)
class Engine: # pylint: disable=too-few-public-methods
class Engine(abc.ABC): # pylint: disable=too-few-public-methods
"""Class of engine instances build from YAML settings.
Further documentation see :ref:`general engine configuration`.
@@ -181,6 +185,8 @@ class Engine: # pylint: disable=too-few-public-methods
This class is currently never initialized and only used for type hinting.
"""
logger: logging.Logger
# Common options in the engine module
engine_type: str
@@ -220,15 +226,15 @@ class Engine: # pylint: disable=too-few-public-methods
region: fr-BE
"""
fetch_traits: Callable
fetch_traits: "Callable[[EngineTraits, bool], None]"
"""Function to to fetch engine's traits from origin."""
traits: traits.EngineTraits
traits: "traits.EngineTraits"
"""Traits of the engine."""
# settings.yml
categories: List[str]
categories: list[str]
"""Specifies to which :ref:`engine categories` the engine should be added."""
name: str
@@ -269,7 +275,7 @@ class Engine: # pylint: disable=too-few-public-methods
inactive: bool
"""Remove the engine from the settings (*disabled & removed*)."""
about: dict
about: dict[str, dict[str, str]]
"""Additional fields describing the engine.
.. code:: yaml
@@ -291,9 +297,21 @@ class Engine: # pylint: disable=too-few-public-methods
the user is used to build and send a ``Accept-Language`` header in the
request to the origin search engine."""
tokens: List[str]
tokens: list[str]
"""A list of secret tokens to make this engine *private*, more details see
:ref:`private engines`."""
weight: int
"""Weighting of the results of this engine (:ref:`weight <settings engines>`)."""
def init(self, engine_settings: dict[str, t.Any]) -> None: # pyright: ignore[reportUnusedParameter]
"""Initialization of the engine. If no initialization is needed, drop
this init function."""
@abc.abstractmethod
def request(self, query: str, params: dict[str, t.Any]) -> None:
"""Build up the params for the online request."""
@abc.abstractmethod
def response(self, resp: "SXNG_Response") -> "EngineResults":
"""Parse out the result items from the response."""

View File

@@ -9,18 +9,17 @@ To load traits from the persistence :py:obj:`EngineTraitsMap.from_data` can be
used.
"""
from __future__ import annotations
import os
import json
import dataclasses
import types
from typing import Dict, Literal, Iterable, Union, Callable, Optional, TYPE_CHECKING
import typing as t
import pathlib
from searx import locales
from searx.data import data_dir, ENGINE_TRAITS
if TYPE_CHECKING:
if t.TYPE_CHECKING:
from . import Engine
@@ -28,7 +27,7 @@ class EngineTraitsEncoder(json.JSONEncoder):
"""Encodes :class:`EngineTraits` to a serializable object, see
:class:`json.JSONEncoder`."""
def default(self, o):
def default(self, o: t.Any) -> t.Any:
"""Return dictionary of a :class:`EngineTraits` object."""
if isinstance(o, EngineTraits):
return o.__dict__
@@ -39,7 +38,7 @@ class EngineTraitsEncoder(json.JSONEncoder):
class EngineTraits:
"""The class is intended to be instantiated for each engine."""
regions: Dict[str, str] = dataclasses.field(default_factory=dict)
regions: dict[str, str] = dataclasses.field(default_factory=dict)
"""Maps SearXNG's internal representation of a region to the one of the engine.
SearXNG's internal representation can be parsed by babel and the value is
@@ -56,7 +55,7 @@ class EngineTraits:
...
"""
languages: Dict[str, str] = dataclasses.field(default_factory=dict)
languages: dict[str, str] = dataclasses.field(default_factory=dict)
"""Maps SearXNG's internal representation of a language to the one of the engine.
SearXNG's internal representation can be parsed by babel and the value is
@@ -73,20 +72,20 @@ class EngineTraits:
...
"""
all_locale: Optional[str] = None
all_locale: str | None = None
"""To which locale value SearXNG's ``all`` language is mapped (shown a "Default
language").
"""
data_type: Literal['traits_v1'] = 'traits_v1'
data_type: t.Literal['traits_v1'] = 'traits_v1'
"""Data type, default is 'traits_v1'.
"""
custom: Dict[str, Union[Dict[str, Dict], Iterable[str]]] = dataclasses.field(default_factory=dict)
custom: dict[str, t.Any] = dataclasses.field(default_factory=dict)
"""A place to store engine's custom traits, not related to the SearXNG core.
"""
def get_language(self, searxng_locale: str, default=None):
def get_language(self, searxng_locale: str, default: t.Any = None):
"""Return engine's language string that *best fits* to SearXNG's locale.
:param searxng_locale: SearXNG's internal representation of locale
@@ -102,7 +101,7 @@ class EngineTraits:
return self.all_locale
return locales.get_engine_locale(searxng_locale, self.languages, default=default)
def get_region(self, searxng_locale: str, default=None):
def get_region(self, searxng_locale: str, default: t.Any = None) -> t.Any:
"""Return engine's region string that best fits to SearXNG's locale.
:param searxng_locale: SearXNG's internal representation of locale
@@ -133,10 +132,10 @@ class EngineTraits:
def copy(self):
"""Create a copy of the dataclass object."""
return EngineTraits(**dataclasses.asdict(self))
return EngineTraits(**dataclasses.asdict(self)) # type: ignore
@classmethod
def fetch_traits(cls, engine: Engine) -> Union['EngineTraits', None]:
def fetch_traits(cls, engine: "Engine | types.ModuleType") -> "EngineTraits | None":
"""Call a function ``fetch_traits(engine_traits)`` from engines namespace to fetch
and set properties from the origin engine in the object ``engine_traits``. If
function does not exists, ``None`` is returned.
@@ -150,7 +149,7 @@ class EngineTraits:
fetch_traits(engine_traits)
return engine_traits
def set_traits(self, engine: Engine):
def set_traits(self, engine: "Engine | types.ModuleType"):
"""Set traits from self object in a :py:obj:`.Engine` namespace.
:param engine: engine instance build by :py:func:`searx.engines.load_engine`
@@ -161,14 +160,14 @@ class EngineTraits:
else:
raise TypeError('engine traits of type %s is unknown' % self.data_type)
def _set_traits_v1(self, engine: Engine):
def _set_traits_v1(self, engine: "Engine | types.ModuleType"):
# For an engine, when there is `language: ...` in the YAML settings the engine
# does support only this one language (region)::
#
# - name: google italian
# engine: google
# language: it
# region: it-IT # type: ignore
# region: it-IT
traits = self.copy()
@@ -186,16 +185,16 @@ class EngineTraits:
raise ValueError(_msg % (engine.name, 'region', engine.region))
traits.regions = {engine.region: regions[engine.region]}
engine.language_support = bool(traits.languages or traits.regions)
engine.language_support = bool(traits.languages or traits.regions) # type: ignore
# set the copied & modified traits in engine's namespace
engine.traits = traits
engine.traits = traits # pyright: ignore[reportAttributeAccessIssue]
class EngineTraitsMap(Dict[str, EngineTraits]):
class EngineTraitsMap(dict[str, EngineTraits]):
"""A python dictionary to map :class:`EngineTraits` by engine name."""
ENGINE_TRAITS_FILE = (data_dir / 'engine_traits.json').resolve()
ENGINE_TRAITS_FILE: pathlib.Path = (data_dir / 'engine_traits.json').resolve()
"""File with persistence of the :py:obj:`EngineTraitsMap`."""
def save_data(self):
@@ -212,7 +211,7 @@ class EngineTraitsMap(Dict[str, EngineTraits]):
return obj
@classmethod
def fetch_traits(cls, log: Callable) -> 'EngineTraitsMap':
def fetch_traits(cls, log: t.Callable[[str], None]) -> 'EngineTraitsMap':
from searx import engines # pylint: disable=cyclic-import, import-outside-toplevel
names = list(engines.engines)
@@ -220,7 +219,7 @@ class EngineTraitsMap(Dict[str, EngineTraits]):
obj = cls()
for engine_name in names:
engine = engines.engines[engine_name]
engine: Engine | types.ModuleType = engines.engines[engine_name]
traits = None
# pylint: disable=broad-exception-caught
@@ -242,7 +241,7 @@ class EngineTraitsMap(Dict[str, EngineTraits]):
return obj
def set_traits(self, engine: Engine | types.ModuleType):
def set_traits(self, engine: "Engine | types.ModuleType"):
"""Set traits in a :py:obj:`Engine` namespace.
:param engine: engine instance build by :py:func:`searx.engines.load_engine`

View File

@@ -10,13 +10,15 @@ intended monkey patching of the engine modules.
engine modules aren't converted to an engine class, these builtin types will
still be needed.
"""
from __future__ import annotations
import logging
from searx.enginelib import traits as _traits
logger: logging.Logger
supported_languages: str
language_aliases: str
language_support: bool
traits: _traits.EngineTraits
# from searx.engines.ENGINE_DEFAULT_ARGS
about: dict[str, dict[str, str | None | bool]]

View File

@@ -8,7 +8,6 @@ usage::
"""
from __future__ import annotations
import typing as t
import sys
@@ -51,8 +50,8 @@ ENGINE_DEFAULT_ARGS: dict[str, int | str | list[t.Any] | dict[str, t.Any] | bool
# set automatically when an engine does not have any tab category
DEFAULT_CATEGORY = 'other'
categories: dict[str, list[str]] = {'general': []}
engines: dict[str, Engine | types.ModuleType] = {}
categories: "dict[str, list[Engine|types.ModuleType]]" = {'general': []}
engines: "dict[str, Engine | types.ModuleType]" = {}
engine_shortcuts = {}
"""Simple map of registered *shortcuts* to name of the engine (or ``None``).
@@ -76,7 +75,7 @@ def check_engine_module(module: types.ModuleType):
raise TypeError(msg)
def load_engine(engine_data: dict[str, t.Any]) -> Engine | types.ModuleType | None:
def load_engine(engine_data: dict[str, t.Any]) -> "Engine | types.ModuleType | None":
"""Load engine from ``engine_data``.
:param dict engine_data: Attributes from YAML ``settings:engines/<engine>``
@@ -151,7 +150,7 @@ def load_engine(engine_data: dict[str, t.Any]) -> Engine | types.ModuleType | No
return engine
def set_loggers(engine, engine_name):
def set_loggers(engine: "Engine|types.ModuleType", engine_name: str):
# set the logger for engine
engine.logger = logger.getChild(engine_name)
# the engine may have load some other engines
@@ -170,7 +169,7 @@ def set_loggers(engine, engine_name):
module.logger = logger.getChild(module_engine_name) # type: ignore
def update_engine_attributes(engine: Engine | types.ModuleType, engine_data):
def update_engine_attributes(engine: "Engine | types.ModuleType", engine_data: dict[str, t.Any]):
# set engine attributes from engine_data
for param_name, param_value in engine_data.items():
if param_name == 'categories':
@@ -188,13 +187,13 @@ def update_engine_attributes(engine: Engine | types.ModuleType, engine_data):
setattr(engine, arg_name, copy.deepcopy(arg_value))
def update_attributes_for_tor(engine: Engine | types.ModuleType):
def update_attributes_for_tor(engine: "Engine | types.ModuleType"):
if using_tor_proxy(engine) and hasattr(engine, 'onion_url'):
engine.search_url = engine.onion_url + getattr(engine, 'search_path', '') # type: ignore
engine.timeout += settings['outgoing'].get('extra_proxy_timeout', 0) # type: ignore
def is_missing_required_attributes(engine):
def is_missing_required_attributes(engine: "Engine | types.ModuleType"):
"""An attribute is required when its name doesn't start with ``_`` (underline).
Required attributes must not be ``None``.
@@ -207,12 +206,12 @@ def is_missing_required_attributes(engine):
return missing
def using_tor_proxy(engine: Engine | types.ModuleType):
def using_tor_proxy(engine: "Engine | types.ModuleType"):
"""Return True if the engine configuration declares to use Tor."""
return settings['outgoing'].get('using_tor_proxy') or getattr(engine, 'using_tor_proxy', False)
def is_engine_active(engine: Engine | types.ModuleType):
def is_engine_active(engine: "Engine | types.ModuleType"):
# check if engine is inactive
if engine.inactive is True:
return False
@@ -224,7 +223,7 @@ def is_engine_active(engine: Engine | types.ModuleType):
return True
def register_engine(engine: Engine | types.ModuleType):
def register_engine(engine: "Engine | types.ModuleType"):
if engine.name in engines:
logger.error('Engine config error: ambiguous name: {0}'.format(engine.name))
sys.exit(1)
@@ -239,7 +238,7 @@ def register_engine(engine: Engine | types.ModuleType):
categories.setdefault(category_name, []).append(engine)
def load_engines(engine_list):
def load_engines(engine_list: list[dict[str, t.Any]]):
"""usage: ``engine_list = settings['engines']``"""
engines.clear()
engine_shortcuts.clear()

View File

@@ -35,19 +35,12 @@ Implementation
==============
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from datetime import datetime, timedelta
from urllib.parse import urlencode
import isodate
if TYPE_CHECKING:
import logging
logger: logging.Logger
about = {
"website": "https://stock.adobe.com/",
"wikidata_id": "Q5977430",

View File

@@ -32,18 +32,24 @@ Implementations
===============
"""
import typing as t
from typing import List, Dict, Any, Optional
from urllib.parse import urlencode
from lxml import html
from lxml.etree import ElementBase
from searx.utils import extract_text, eval_xpath, eval_xpath_getindex, eval_xpath_list
from searx.enginelib.traits import EngineTraits
from searx.data import ENGINE_TRAITS
from searx.exceptions import SearxEngineXPathException
from searx.result_types import EngineResults
if t.TYPE_CHECKING:
from searx.extended_types import SXNG_Response
# about
about: Dict[str, Any] = {
about: dict[str, t.Any] = {
"website": "https://annas-archive.org/",
"wikidata_id": "Q115288326",
"official_api_documentation": None,
@@ -53,7 +59,7 @@ about: Dict[str, Any] = {
}
# engine dependent config
categories: List[str] = ["files"]
categories: list[str] = ["files"]
paging: bool = True
# search-url
@@ -85,7 +91,7 @@ aa_ext: str = ''
"""
def init(engine_settings=None): # pylint: disable=unused-argument
def init(engine_settings: dict[str, t.Any]) -> None: # pylint: disable=unused-argument
"""Check of engine's settings."""
traits = EngineTraits(**ENGINE_TRAITS['annas archive'])
@@ -99,8 +105,8 @@ def init(engine_settings=None): # pylint: disable=unused-argument
raise ValueError(f'invalid setting ext: {aa_ext}')
def request(query, params: Dict[str, Any]) -> Dict[str, Any]:
lang = traits.get_language(params["language"], traits.all_locale) # type: ignore
def request(query: str, params: dict[str, t.Any]) -> None:
lang = traits.get_language(params["language"], traits.all_locale)
args = {
'lang': lang,
'content': aa_content,
@@ -112,11 +118,10 @@ def request(query, params: Dict[str, Any]) -> Dict[str, Any]:
# filter out None and empty values
filtered_args = dict((k, v) for k, v in args.items() if v)
params["url"] = f"{base_url}/search?{urlencode(filtered_args)}"
return params
def response(resp) -> List[Dict[str, Optional[str]]]:
results: List[Dict[str, Optional[str]]] = []
def response(resp: "SXNG_Response") -> EngineResults:
res = EngineResults()
dom = html.fromstring(resp.text)
# The rendering of the WEB page is strange; positions of Anna's result page
@@ -126,16 +131,17 @@ def response(resp) -> List[Dict[str, Optional[str]]]:
for item in eval_xpath_list(dom, '//main//div[contains(@class, "js-aarecord-list-outer")]/div'):
try:
results.append(_get_result(item))
kwargs: dict[str, t.Any] = _get_result(item)
except SearxEngineXPathException:
pass
return results
continue
res.add(res.types.LegacyResult(**kwargs))
return res
def _get_result(item):
def _get_result(item: ElementBase) -> dict[str, t.Any]:
return {
'template': 'paper.html',
'url': base_url + extract_text(eval_xpath_getindex(item, './a/@href', 0)),
'url': base_url + eval_xpath_getindex(item, './a/@href', 0),
'title': extract_text(eval_xpath(item, './div//a[starts-with(@href, "/md5")]')),
'authors': [extract_text(eval_xpath_getindex(item, './/a[starts-with(@href, "/search")]', 0))],
'publisher': extract_text(
@@ -160,9 +166,9 @@ def fetch_traits(engine_traits: EngineTraits):
engine_traits.custom['sort'] = []
resp = get(base_url + '/search')
if not resp.ok: # type: ignore
if not resp.ok:
raise RuntimeError("Response from Anna's search page is not OK.")
dom = html.fromstring(resp.text) # type: ignore
dom = html.fromstring(resp.text)
# supported language codes

View File

@@ -8,7 +8,6 @@ Arch Wiki blocks access to it.
"""
from typing import TYPE_CHECKING
from urllib.parse import urlencode, urljoin, urlparse
import lxml
import babel
@@ -17,13 +16,6 @@ from searx.utils import extract_text, eval_xpath_list, eval_xpath_getindex
from searx.enginelib.traits import EngineTraits
from searx.locales import language_tag
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
about = {
"website": 'https://wiki.archlinux.org/',

View File

@@ -26,7 +26,6 @@ category for the Chinese market.
"""
# pylint: disable=too-many-branches, invalid-name
from typing import TYPE_CHECKING
import base64
import re
import time
@@ -40,13 +39,6 @@ from searx.locales import language_tag, region_tag
from searx.enginelib.traits import EngineTraits
from searx.exceptions import SearxEngineAPIException
if TYPE_CHECKING:
import logging
logger = logging.getLogger()
traits: EngineTraits
about = {
"website": 'https://www.bing.com',
"wikidata_id": 'Q182496',

View File

@@ -2,26 +2,14 @@
"""Bing-Images: description see :py:obj:`searx.engines.bing`.
"""
# pylint: disable=invalid-name
from typing import TYPE_CHECKING
import json
from urllib.parse import urlencode
from lxml import html
from searx.enginelib.traits import EngineTraits
from searx.engines.bing import set_bing_cookies
from searx.engines.bing import fetch_traits # pylint: disable=unused-import
if TYPE_CHECKING:
import logging
logger = logging.getLogger()
traits: EngineTraits
# about
about = {
"website": 'https://www.bing.com/images',

View File

@@ -9,7 +9,6 @@
# pylint: disable=invalid-name
from typing import TYPE_CHECKING
from urllib.parse import urlencode
from lxml import html
@@ -18,14 +17,6 @@ from searx.utils import eval_xpath, extract_text, eval_xpath_list, eval_xpath_ge
from searx.enginelib.traits import EngineTraits
from searx.engines.bing import set_bing_cookies
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {
"website": 'https://www.bing.com/news',

View File

@@ -3,24 +3,15 @@
"""Bing-Videos: description see :py:obj:`searx.engines.bing`.
"""
from typing import TYPE_CHECKING
import json
from urllib.parse import urlencode
from lxml import html
from searx.enginelib.traits import EngineTraits
from searx.engines.bing import set_bing_cookies
from searx.engines.bing import fetch_traits # pylint: disable=unused-import
from searx.engines.bing_images import time_map
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
about = {
"website": 'https://www.bing.com/videos',

View File

@@ -117,7 +117,7 @@ Implementations
"""
from typing import Any, TYPE_CHECKING
import typing as t
from urllib.parse import (
urlencode,
@@ -139,13 +139,7 @@ from searx.utils import (
)
from searx.enginelib.traits import EngineTraits
from searx.result_types import EngineResults
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
from searx.extended_types import SXNG_Response
about = {
"website": 'https://search.brave.com/',
@@ -158,17 +152,19 @@ about = {
base_url = "https://search.brave.com/"
categories = []
brave_category = 'search'
Goggles = Any
brave_category: t.Literal["search", "videos", "images", "news", "goggles"] = 'search'
"""Brave supports common web-search, videos, images, news, and goggles search.
- ``search``: Common WEB search
- ``videos``: search for videos
- ``images``: search for images
- ``news``: search for news
- ``goggles``: Common WEB search with custom rules
- ``goggles``: Common WEB search with custom rules, requires a :py:obj:`Goggles` URL.
"""
Goggles: str = ""
"""This should be a URL ending in ``.goggle``"""
brave_spellcheck = False
"""Brave supports some kind of spell checking. When activated, Brave tries to
fix typos, e.g. it searches for ``food`` when the user queries for ``fooh``. In
@@ -192,7 +188,7 @@ time_range_support = False
"""Brave only supports time-range in :py:obj:`brave_category` ``search`` (UI
category All) and in the goggles category."""
time_range_map = {
time_range_map: dict[str, str] = {
'day': 'pd',
'week': 'pw',
'month': 'pm',
@@ -200,12 +196,12 @@ time_range_map = {
}
def request(query, params):
def request(query: str, params: dict[str, t.Any]) -> None:
# Don't accept br encoding / see https://github.com/searxng/searxng/pull/1787
params['headers']['Accept-Encoding'] = 'gzip, deflate'
args = {
args: dict[str, t.Any] = {
'q': query,
'source': 'web',
}
@@ -254,7 +250,7 @@ def _extract_published_date(published_date_raw):
return None
def response(resp) -> EngineResults:
def response(resp: SXNG_Response) -> EngineResults:
if brave_category in ('search', 'goggles'):
return _parse_search(resp)

View File

@@ -54,8 +54,8 @@ Implementations
"""
import typing as t
import base64
import typing
import secrets
from urllib.parse import urlencode
@@ -78,7 +78,7 @@ time_range_support = True
results_per_page = 10
categories = []
ChinasoCategoryType = typing.Literal['news', 'videos', 'images']
ChinasoCategoryType = t.Literal['news', 'videos', 'images']
"""ChinaSo supports news, videos, images search.
- ``news``: search for news
@@ -91,7 +91,7 @@ In the category ``news`` you can additionally filter by option
chinaso_category = 'news'
"""Configure ChinaSo category (:py:obj:`ChinasoCategoryType`)."""
ChinasoNewsSourceType = typing.Literal['CENTRAL', 'LOCAL', 'BUSINESS', 'EPAPER', 'all']
ChinasoNewsSourceType = t.Literal['CENTRAL', 'LOCAL', 'BUSINESS', 'EPAPER', 'all']
"""Filtering ChinaSo-News results by source:
- ``CENTRAL``: central publication
@@ -111,7 +111,7 @@ base_url = "https://www.chinaso.com"
def init(_):
if chinaso_category not in ('news', 'videos', 'images'):
raise ValueError(f"Unsupported category: {chinaso_category}")
if chinaso_category == 'news' and chinaso_news_source not in typing.get_args(ChinasoNewsSourceType):
if chinaso_category == 'news' and chinaso_news_source not in t.get_args(ChinasoNewsSourceType):
raise ValueError(f"Unsupported news source: {chinaso_news_source}")

View File

@@ -10,8 +10,6 @@ Dailymotion (Videos)
"""
from typing import TYPE_CHECKING
from datetime import datetime, timedelta
from urllib.parse import urlencode
import time
@@ -23,13 +21,6 @@ from searx.exceptions import SearxEngineAPIException
from searx.locales import region_tag, language_tag
from searx.enginelib.traits import EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {
"website": 'https://www.dailymotion.com',

View File

@@ -12,13 +12,14 @@ close to the implementation, its just a simple example. To get in use of this
"""
import typing as t
import json
from searx.result_types import EngineResults
from searx.enginelib import EngineCache
engine_type = 'offline'
categories = ['general']
engine_type = "offline"
categories = ["general"]
disabled = True
timeout = 2.0
@@ -38,13 +39,13 @@ CACHE: EngineCache
seconds."""
def init(engine_settings):
def init(engine_settings: dict[str, t.Any]) -> None:
"""Initialization of the (offline) engine. The origin of this demo engine is a
simple json string which is loaded in this example while the engine is
initialized."""
global _my_offline_engine, CACHE # pylint: disable=global-statement
CACHE = EngineCache(engine_settings["name"]) # type:ignore
CACHE = EngineCache(engine_settings["name"])
_my_offline_engine = (
'[ {"value": "%s"}'
@@ -55,20 +56,22 @@ def init(engine_settings):
)
def search(query, request_params) -> EngineResults:
def search(query: str, params: dict[str, t.Any]) -> EngineResults:
"""Query (offline) engine and return results. Assemble the list of results
from your local engine. In this demo engine we ignore the 'query' term,
usual you would pass the 'query' term to your local engine to filter out the
results.
"""
res = EngineResults()
count = CACHE.get("count", 0)
for row in json.loads(_my_offline_engine):
count: int = CACHE.get("count", 0)
data_rows: list[dict[str, str]] = json.loads(_my_offline_engine)
for row in data_rows:
count += 1
kvmap = {
'query': query,
'language': request_params['searxng_locale'],
'language': params['searxng_locale'],
'value': row.get("value"),
}
res.add(

View File

@@ -15,29 +15,35 @@ list in ``settings.yml``:
"""
import typing as t
from json import loads
from urllib.parse import urlencode
from searx.result_types import EngineResults
engine_type = 'online'
if t.TYPE_CHECKING:
from searx.extended_types import SXNG_Response
engine_type = "online"
send_accept_language_header = True
categories = ['general']
categories = ["general"]
disabled = True
timeout = 2.0
categories = ['images']
categories = ["images"]
paging = True
page_size = 20
search_api = 'https://api.artic.edu/api/v1/artworks/search?'
image_api = 'https://www.artic.edu/iiif/2/'
search_api = "https://api.artic.edu/api/v1/artworks/search?"
image_api = "https://www.artic.edu/iiif/2/"
about = {
"website": 'https://www.artic.edu',
"wikidata_id": 'Q239303',
"official_api_documentation": 'http://api.artic.edu/docs/',
"website": "https://www.artic.edu",
"wikidata_id": "Q239303",
"official_api_documentation": "http://api.artic.edu/docs/",
"use_official_api": True,
"require_api_key": False,
"results": 'JSON',
"results": "JSON",
}
@@ -45,33 +51,30 @@ about = {
_my_online_engine = None
def init(engine_settings):
def init(engine_settings: dict[str, t.Any]) -> None:
"""Initialization of the (online) engine. If no initialization is needed, drop
this init function.
"""
this init function."""
global _my_online_engine # pylint: disable=global-statement
_my_online_engine = engine_settings.get('name')
_my_online_engine = engine_settings.get("name")
def request(query, params):
def request(query: str, params: dict[str, t.Any]) -> None:
"""Build up the ``params`` for the online request. In this example we build a
URL to fetch images from `artic.edu <https://artic.edu>`__
"""
args = urlencode(
{
'q': query,
'page': params['pageno'],
'fields': 'id,title,artist_display,medium_display,image_id,date_display,dimensions,artist_titles',
'limit': page_size,
"q": query,
"page": params["pageno"],
"fields": "id,title,artist_display,medium_display,image_id,date_display,dimensions,artist_titles",
"limit": page_size,
}
)
params['url'] = search_api + args
return params
params["url"] = search_api + args
def response(resp) -> EngineResults:
def response(resp: "SXNG_Response") -> EngineResults:
"""Parse out the result items from the response. In this example we parse the
response from `api.artic.edu <https://artic.edu>`__ and filter out all
images.
@@ -87,20 +90,20 @@ def response(resp) -> EngineResults:
)
)
for result in json_data['data']:
for result in json_data["data"]:
if not result['image_id']:
if not result["image_id"]:
continue
res.append(
{
'url': 'https://artic.edu/artworks/%(id)s' % result,
'title': result['title'] + " (%(date_display)s) // %(artist_display)s" % result,
'content': "%(medium_display)s // %(dimensions)s" % result,
'author': ', '.join(result['artist_titles']),
'img_src': image_api + '/%(image_id)s/full/843,/0/default.jpg' % result,
'template': 'images.html',
}
)
kwargs: dict[str, t.Any] = {
"url": "https://artic.edu/artworks/%(id)s" % result,
"title": result["title"] + " (%(date_display)s) // %(artist_display)s" % result,
"content": "%(medium_display)s // %(dimensions)s" % result,
"author": ", ".join(result["artist_titles"]),
"img_src": image_api + "/%(image_id)s/full/843,/0/default.jpg" % result,
"template": "images.html",
}
res.add(res.types.LegacyResult(**kwargs))
return res

View File

@@ -4,11 +4,8 @@ DuckDuckGo WEB
~~~~~~~~~~~~~~
"""
from __future__ import annotations
import json
import re
import typing
from urllib.parse import quote_plus
@@ -31,13 +28,6 @@ from searx.enginelib import EngineCache
from searx.exceptions import SearxEngineCaptchaException
from searx.result_types import EngineResults
if typing.TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
about = {
"website": 'https://lite.duckduckgo.com/lite/',
"wikidata_id": 'Q12805',

View File

@@ -13,8 +13,6 @@ most of the features are based on English terms.
"""
from typing import TYPE_CHECKING
from urllib.parse import urlencode, urlparse, urljoin
from lxml import html
@@ -23,11 +21,6 @@ from searx.utils import extract_text, html_to_text, get_string_replaces_function
from searx.external_urls import get_external_url, get_earth_coordinates_url, area_to_osm_zoom
from searx.result_types import EngineResults
if TYPE_CHECKING:
import logging
logger: logging.Logger
# about
about = {
"website": 'https://duckduckgo.com/',

View File

@@ -4,23 +4,12 @@ DuckDuckGo Extra (images, videos, news)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
"""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from urllib.parse import urlencode
from searx.utils import get_embeded_stream_url, html_to_text
from searx.engines.duckduckgo import fetch_traits # pylint: disable=unused-import
from searx.engines.duckduckgo import get_ddg_lang, get_vqd
from searx.enginelib.traits import EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {

View File

@@ -3,7 +3,6 @@
DuckDuckGo Weather
~~~~~~~~~~~~~~~~~~
"""
from __future__ import annotations
import typing as t
from json import loads
@@ -13,19 +12,11 @@ from dateutil import parser as date_parser
from searx.engines.duckduckgo import fetch_traits # pylint: disable=unused-import
from searx.engines.duckduckgo import get_ddg_lang
from searx.enginelib.traits import EngineTraits
from searx.result_types import EngineResults
from searx.extended_types import SXNG_Response
from searx import weather
if t.TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
about = {
"website": 'https://duckduckgo.com/',

View File

@@ -3,19 +3,12 @@
"""
from typing import TYPE_CHECKING
import json
from time import time
import re
from urllib.parse import urlencode
from searx.utils import ecma_unescape, html_to_text
if TYPE_CHECKING:
import logging
logger: logging.Logger
# about
about = {
"website": 'https://www.flickr.com',

View File

@@ -65,7 +65,6 @@ code lines are just relabeled (starting from 1) and appended (a disjoint set of
code blocks in a single file might be returned from the API).
"""
from __future__ import annotations
import typing as t
from urllib.parse import urlencode

View File

@@ -10,9 +10,6 @@ engines:
- :ref:`google autocomplete`
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import re
import random
@@ -31,13 +28,6 @@ from searx.exceptions import SearxEngineCaptchaException
from searx.enginelib.traits import EngineTraits
from searx.result_types import EngineResults
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {

View File

@@ -13,8 +13,6 @@ This internal API offer results in
.. _Protobuf: https://en.wikipedia.org/wiki/Protocol_Buffers
"""
from typing import TYPE_CHECKING
from urllib.parse import urlencode
from json import loads
@@ -25,14 +23,6 @@ from searx.engines.google import (
detect_google_sorry,
)
if TYPE_CHECKING:
import logging
from searx.enginelib.traits import EngineTraits
logger: logging.Logger
traits: EngineTraits
# about
about = {
"website": 'https://images.google.com',

View File

@@ -24,8 +24,6 @@ The google news API ignores some parameters from the common :ref:`google API`:
.. _save: https://developers.google.com/custom-search/docs/xml_results#safesp
"""
from typing import TYPE_CHECKING
from urllib.parse import urlencode
import base64
from lxml import html
@@ -46,13 +44,6 @@ from searx.engines.google import (
)
from searx.enginelib.traits import EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {
"website": 'https://news.google.com',

View File

@@ -7,9 +7,6 @@ can make use of the :ref:`google API` to assemble the arguments of the GET
request.
"""
from typing import TYPE_CHECKING
from typing import Optional
from urllib.parse import urlencode
from datetime import datetime
from lxml import html
@@ -28,14 +25,6 @@ from searx.engines.google import (
get_google_info,
time_range_dict,
)
from searx.enginelib.traits import EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {
@@ -115,7 +104,7 @@ def request(query, params):
return params
def parse_gs_a(text: Optional[str]):
def parse_gs_a(text: str | None):
"""Parse the text written in green.
Possible formats:

View File

@@ -11,7 +11,6 @@
.. _data URLs:
https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/Data_URIs
"""
from __future__ import annotations
from urllib.parse import urlencode, urlparse, parse_qs
from lxml import html
@@ -32,11 +31,8 @@ from searx.engines.google import (
ui_async,
parse_data_images,
)
from searx.enginelib.traits import EngineTraits
from searx.utils import get_embeded_stream_url
traits: EngineTraits
# about
about = {
"website": 'https://www.google.com',

View File

@@ -7,7 +7,6 @@ No public instance offer a public API now
- https://github.com/searxng/searxng/issues/2722#issuecomment-2884993248
"""
from __future__ import annotations
import time
import random

View File

@@ -26,7 +26,6 @@ Implementations
===============
"""
from __future__ import annotations
import typing as t
from urllib.parse import urlencode, quote_plus

View File

@@ -26,8 +26,6 @@ Implementations
"""
from typing import TYPE_CHECKING
try:
import mariadb # pyright: ignore [reportMissingImports]
except ImportError:
@@ -37,12 +35,6 @@ except ImportError:
from searx.result_types import EngineResults
if TYPE_CHECKING:
import logging
logger = logging.getLogger()
engine_type = 'offline'
host = "127.0.0.1"

View File

@@ -32,21 +32,11 @@ Implementations
===============
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from datetime import datetime
from urllib.parse import urlencode, quote
from searx.utils import html_to_text
from searx.enginelib.traits import EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {

View File

@@ -1,8 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Mojeek (general, images, news)"""
from typing import TYPE_CHECKING
from datetime import datetime
from urllib.parse import urlencode
from lxml import html
@@ -50,13 +48,6 @@ region_param = 'arc'
_delta_kwargs = {'day': 'days', 'week': 'weeks', 'month': 'months', 'year': 'years'}
if TYPE_CHECKING:
import logging
logger = logging.getLogger()
traits: EngineTraits
def init(_):
if search_type not in ('', 'images', 'news'):

View File

@@ -37,7 +37,6 @@ Implementations
===============
"""
from __future__ import annotations
import re

View File

@@ -36,10 +36,8 @@ Implementations
===============
"""
import typing as t
from __future__ import annotations
import typing
from urllib.parse import urlencode
import babel
from httpx import Response
@@ -49,13 +47,6 @@ from searx.locales import get_official_locales, language_tag, region_tag
from searx.utils import eval_xpath_list
from searx.result_types import EngineResults, MainResult
if typing.TYPE_CHECKING:
import logging
logger = logging.getLogger()
traits: EngineTraits
search_url = "https://leta.mullvad.net"
# about
@@ -80,7 +71,7 @@ time_range_dict = {
"year": "y",
}
LetaEnginesType = typing.Literal["google", "brave"]
LetaEnginesType = t.Literal["google", "brave"]
"""Engine types supported by mullvadleta."""
leta_engine: LetaEnginesType = "google"
@@ -88,12 +79,12 @@ leta_engine: LetaEnginesType = "google"
def init(_):
l = typing.get_args(LetaEnginesType)
l = t.get_args(LetaEnginesType)
if leta_engine not in l:
raise ValueError(f"leta_engine '{leta_engine}' is invalid, use one of {', '.join(l)}")
class DataNodeQueryMetaDataIndices(typing.TypedDict):
class DataNodeQueryMetaDataIndices(t.TypedDict):
"""Indices into query metadata."""
success: int
@@ -112,7 +103,7 @@ class DataNodeQueryMetaDataIndices(typing.TypedDict):
previous: int
class DataNodeResultIndices(typing.TypedDict):
class DataNodeResultIndices(t.TypedDict):
"""Indices into query resultsdata."""
link: int

View File

@@ -14,8 +14,6 @@ from searx.network import get
from searx.locales import language_tag
from searx.enginelib.traits import EngineTraits
traits: EngineTraits
# Engine metadata
about = {
"website": "https://odysee.com/",

View File

@@ -3,7 +3,6 @@
#
# Engine is documented in: docs/dev/engines/online/openalex.rst
from __future__ import annotations
import typing as t
from datetime import datetime

View File

@@ -17,8 +17,6 @@ from searx.locales import language_tag
from searx.utils import html_to_text, humanize_number
from searx.enginelib.traits import EngineTraits
traits: EngineTraits
about = {
# pylint: disable=line-too-long
"website": 'https://joinpeertube.org',

View File

@@ -45,7 +45,6 @@ Implementations
===============
"""
from __future__ import annotations
import time
import random

View File

@@ -64,8 +64,6 @@ from searx.utils import (
get_embeded_stream_url,
)
traits: EngineTraits
# about
about = {
"website": 'https://www.qwant.com/',

View File

@@ -5,9 +5,6 @@
https://de1.api.radio-browser.info/#Advanced_station_search
"""
from __future__ import annotations
import typing
import random
import socket
from urllib.parse import urlencode
@@ -19,12 +16,6 @@ from searx.enginelib import EngineCache
from searx.enginelib.traits import EngineTraits
from searx.locales import language_tag
if typing.TYPE_CHECKING:
import logging
logger = logging.getLogger()
traits: EngineTraits
about = {
"website": 'https://www.radio-browser.info/',

View File

@@ -1,10 +1,10 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""SensCritique (movies)
"""
from __future__ import annotations
import typing as t
from json import dumps, loads
from typing import Any, Optional
from searx.result_types import EngineResults, MainResult
about = {
@@ -61,7 +61,7 @@ graphql_query = """query SearchProductExplorer($query: String, $offset: Int, $li
}"""
def request(query: str, params: dict[str, Any]) -> dict[str, Any]:
def request(query: str, params: dict[str, t.Any]) -> dict[str, t.Any]:
offset = (params['pageno'] - 1) * page_size
data = {
@@ -95,7 +95,7 @@ def response(resp) -> EngineResults:
return res
def parse_item(item: dict[str, Any]) -> MainResult | None:
def parse_item(item: dict[str, t.Any]) -> MainResult | None:
"""Parse a single item from the SensCritique API response"""
title = item.get('title', '')
if not title:
@@ -118,7 +118,7 @@ def parse_item(item: dict[str, Any]) -> MainResult | None:
)
def build_content_parts(item: dict[str, Any], title: str, original_title: Optional[str]) -> list[str]:
def build_content_parts(item: dict[str, t.Any], title: str, original_title: str | None) -> list[str]:
"""Build the content parts for an item"""
content_parts = []

View File

@@ -5,8 +5,6 @@ peertube engines.
"""
from typing import TYPE_CHECKING
from urllib.parse import urlencode
from datetime import datetime
@@ -17,14 +15,6 @@ from searx.engines.peertube import (
safesearch_table,
time_range_table,
)
from searx.enginelib.traits import EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
about = {
# pylint: disable=line-too-long

View File

@@ -1,9 +1,7 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""SoundCloud is a German audio streaming service."""
from __future__ import annotations
import re
import typing
import datetime
from urllib.parse import quote_plus, urlencode
@@ -14,11 +12,6 @@ from lxml import html
from searx.network import get as http_get
from searx.enginelib import EngineCache
if typing.TYPE_CHECKING:
import logging
logger: logging.Logger
about = {
"website": "https://soundcloud.com",
"wikidata_id": "Q568769",

View File

@@ -44,7 +44,7 @@ Implementations
===============
"""
import typing
import typing as t
import sqlite3
import contextlib
@@ -59,7 +59,7 @@ database = ""
query_str = ""
"""SQL query that returns the result items."""
result_type: typing.Literal["MainResult", "KeyValue"] = "KeyValue"
result_type: t.Literal["MainResult", "KeyValue"] = "KeyValue"
"""The result type can be :py:obj:`MainResult` or :py:obj:`KeyValue`."""
limit = 10

View File

@@ -78,9 +78,9 @@ Startpage's category (for Web-search, News, Videos, ..) is set by
"""
# pylint: disable=too-many-statements
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import typing as t
from collections import OrderedDict
import re
from unicodedata import normalize, combining
@@ -98,13 +98,6 @@ from searx.locales import region_tag
from searx.enginelib.traits import EngineTraits
from searx.enginelib import EngineCache
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {
"website": 'https://startpage.com',
@@ -377,7 +370,7 @@ def _get_news_result(result):
}
def _get_image_result(result) -> dict[str, Any] | None:
def _get_image_result(result) -> dict[str, t.Any] | None:
url = result.get('altClickUrl')
if not url:
return None

View File

@@ -22,8 +22,6 @@ paging = True
base_url = "https://stract.com/beta/api"
search_url = base_url + "/search"
traits: EngineTraits
def request(query, params):
params['url'] = search_url

View File

@@ -15,17 +15,11 @@ This SearXNG engine uses the `/api2u/search`_ API.
.. _OpenAPI: https://swagger.io/specification/
"""
from typing import TYPE_CHECKING
from datetime import datetime
from urllib.parse import urlencode
import re
if TYPE_CHECKING:
import logging
logger: logging.Logger
about = {
'website': "https://tagesschau.de",
'wikidata_id': "Q703907",

View File

@@ -14,18 +14,12 @@ billion images `[tineye.com] <https://tineye.com/how>`_.
"""
from typing import TYPE_CHECKING
from urllib.parse import urlencode
from datetime import datetime
from flask_babel import gettext
from searx.result_types import EngineResults
if TYPE_CHECKING:
import logging
logger = logging.getLogger()
about = {
"website": 'https://tineye.com',
"wikidata_id": 'Q2382535',

View File

@@ -47,10 +47,8 @@ Implementations
===============
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import List, Dict, Any
import typing as t
from datetime import datetime
from urllib.parse import quote
from lxml import etree # type: ignore
@@ -58,14 +56,12 @@ from lxml import etree # type: ignore
from searx.exceptions import SearxEngineAPIException
from searx.utils import humanize_bytes
if TYPE_CHECKING:
import httpx
import logging
if t.TYPE_CHECKING:
from searx.extended_types import SXNG_Response
logger: logging.Logger
# engine settings
about: Dict[str, Any] = {
about: dict[str, t.Any] = {
"website": None,
"wikidata_id": None,
"official_api_documentation": "https://torznab.github.io/spec-1.3-draft",
@@ -73,7 +69,7 @@ about: Dict[str, Any] = {
"require_api_key": False,
"results": 'XML',
}
categories: List[str] = ['files']
categories: list[str] = ['files']
paging: bool = False
time_range_support: bool = False
@@ -82,7 +78,7 @@ time_range_support: bool = False
base_url: str = ''
api_key: str = ''
# https://newznab.readthedocs.io/en/latest/misc/api/#predefined-categories
torznab_categories: List[str] = []
torznab_categories: list[str] = []
show_torrent_files: bool = False
show_magnet_links: bool = True
@@ -93,7 +89,7 @@ def init(engine_settings=None): # pylint: disable=unused-argument
raise ValueError('missing torznab base_url')
def request(query: str, params: Dict[str, Any]) -> Dict[str, Any]:
def request(query: str, params: dict[str, t.Any]) -> dict[str, t.Any]:
"""Build the request params."""
search_url: str = base_url + '?t=search&q={search_query}'
@@ -109,7 +105,7 @@ def request(query: str, params: Dict[str, Any]) -> Dict[str, Any]:
return params
def response(resp: httpx.Response) -> List[Dict[str, Any]]:
def response(resp: "SXNG_Response") -> list[dict[str, t.Any]]:
"""Parse the XML response and return a list of results."""
results = []
search_results = etree.XML(resp.content)
@@ -122,13 +118,13 @@ def response(resp: httpx.Response) -> List[Dict[str, Any]]:
item: etree.Element
for item in channel.iterfind('item'):
result: Dict[str, Any] = build_result(item)
result: dict[str, t.Any] = build_result(item)
results.append(result)
return results
def build_result(item: etree.Element) -> Dict[str, Any]:
def build_result(item: etree.Element) -> dict[str, t.Any]:
"""Build a result from a XML item."""
# extract attributes from XML
@@ -150,7 +146,7 @@ def build_result(item: etree.Element) -> Dict[str, Any]:
peers = get_torznab_attribute(item, 'peers')
# map attributes to SearXNG result
result: Dict[str, Any] = {
result: dict[str, t.Any] = {
'template': 'torrent.html',
'title': get_attribute(item, 'title'),
'filesize': humanize_bytes(int(filesize)) if filesize else None,

View File

@@ -74,7 +74,6 @@ Implementations
===============
"""
from __future__ import annotations
from urllib.parse import urlencode
from dateutil.parser import parse

View File

@@ -5,7 +5,6 @@ from :ref:`wikipedia engine`.
"""
# pylint: disable=missing-class-docstring
from typing import TYPE_CHECKING
from hashlib import md5
from urllib.parse import urlencode, unquote
from json import loads
@@ -23,13 +22,6 @@ from searx.engines.wikipedia import (
)
from searx.enginelib.traits import EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
traits: EngineTraits
# about
about = {
"website": 'https://wikidata.org/',

View File

@@ -64,8 +64,6 @@ from searx import network as _network
from searx import locales
from searx.enginelib.traits import EngineTraits
traits: EngineTraits
# about
about = {
"website": 'https://www.wikipedia.org/',

View File

@@ -3,7 +3,6 @@
Wolfram|Alpha (Science)
"""
from __future__ import annotations
from json import loads
from urllib.parse import urlencode

View File

@@ -51,7 +51,6 @@ Implementations
"""
# pylint: disable=fixme
from __future__ import annotations
import random
from json import loads

View File

@@ -6,7 +6,6 @@ found in :py:obj:`lang2domain` URL ``<lang>.search.yahoo.com`` is used.
"""
from typing import TYPE_CHECKING
from urllib.parse import (
unquote,
urlencode,
@@ -19,14 +18,6 @@ from searx.utils import (
extract_text,
html_to_text,
)
from searx.enginelib.traits import EngineTraits
traits: EngineTraits
if TYPE_CHECKING:
import logging
logger: logging.Logger
# about
about = {

View File

@@ -32,27 +32,23 @@ Implementations
===============
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import List, Dict, Any, Optional
import typing as t
from datetime import datetime
from urllib.parse import quote
from lxml import html
from flask_babel import gettext
from flask_babel import gettext # pyright: ignore[reportUnknownVariableType]
from searx.utils import extract_text, eval_xpath, eval_xpath_list
from searx.enginelib.traits import EngineTraits
from searx.data import ENGINE_TRAITS
from searx.exceptions import SearxException
if TYPE_CHECKING:
import httpx
import logging
logger: logging.Logger
if t.TYPE_CHECKING:
from searx.extended_types import SXNG_Response
# about
about: Dict[str, Any] = {
about: dict[str, t.Any] = {
"website": "https://zlibrary-global.se",
"wikidata_id": "Q104863992",
"official_api_documentation": None,
@@ -61,7 +57,7 @@ about: Dict[str, Any] = {
"results": "HTML",
}
categories: List[str] = ["files"]
categories: list[str] = ["files"]
paging: bool = True
base_url: str = "https://zlibrary-global.se"
@@ -79,7 +75,7 @@ zlib_ext: str = ""
"""
def init(engine_settings=None) -> None: # pylint: disable=unused-argument
def init(engine_settings: dict[str, t.Any] | None = None) -> None: # pylint: disable=unused-argument
"""Check of engine's settings."""
traits: EngineTraits = EngineTraits(**ENGINE_TRAITS["z-library"])
@@ -91,7 +87,7 @@ def init(engine_settings=None) -> None: # pylint: disable=unused-argument
raise ValueError(f"invalid setting year_to: {zlib_year_to}")
def request(query: str, params: Dict[str, Any]) -> Dict[str, Any]:
def request(query: str, params: dict[str, t.Any]) -> dict[str, t.Any]:
lang: str = traits.get_language(params["language"], traits.all_locale) # type: ignore
search_url: str = (
base_url
@@ -117,8 +113,8 @@ def domain_is_seized(dom):
return bool(dom.xpath('//title') and "seized" in dom.xpath('//title')[0].text.lower())
def response(resp: httpx.Response) -> List[Dict[str, Any]]:
results: List[Dict[str, Any]] = []
def response(resp: "SXNG_Response") -> list[dict[str, t.Any]]:
results: list[dict[str, t.Any]] = []
dom = html.fromstring(resp.text)
if domain_is_seized(dom):
@@ -139,7 +135,7 @@ i18n_book_rating = gettext("Book rating")
i18n_file_quality = gettext("File quality")
def _parse_result(item) -> Dict[str, Any]:
def _parse_result(item) -> dict[str, t.Any]:
author_elements = eval_xpath_list(item, './/div[@class="authors"]//a[@itemprop="author"]')
@@ -152,7 +148,7 @@ def _parse_result(item) -> Dict[str, Any]:
"type": _text(item, './/div[contains(@class, "property__file")]//div[contains(@class, "property_value")]'),
}
thumbnail = _text(item, './/img[contains(@class, "cover")]/@data-src')
thumbnail: str = _text(item, './/img[contains(@class, "cover")]/@data-src')
if not thumbnail.startswith('/'):
result["thumbnail"] = thumbnail
@@ -199,7 +195,7 @@ def fetch_traits(engine_traits: EngineTraits) -> None:
_use_old_values()
return
if not resp.ok: # type: ignore
if not resp.ok:
raise RuntimeError("Response from zlibrary's search page is not OK.")
dom = html.fromstring(resp.text) # type: ignore
@@ -220,20 +216,20 @@ def fetch_traits(engine_traits: EngineTraits) -> None:
engine_traits.custom["year_to"].append(year.get("value"))
for ext in eval_xpath_list(dom, "//div[@id='advSearch-noJS']//select[@id='sf_extensions']/option"):
value: Optional[str] = ext.get("value")
value: str | None = ext.get("value")
if value is None:
value = ""
engine_traits.custom["ext"].append(value)
# Handle languages
# Z-library uses English names for languages, so we need to map them to their respective locales
language_name_locale_map: Dict[str, babel.Locale] = {}
language_name_locale_map: dict[str, babel.Locale] = {}
for locale in babel.core.localedata.locale_identifiers(): # type: ignore
# Create a Locale object for the current locale
loc = babel.Locale.parse(locale)
if loc.english_name is None:
continue
language_name_locale_map[loc.english_name.lower()] = loc # type: ignore
language_name_locale_map[loc.english_name.lower()] = loc
for x in eval_xpath_list(dom, "//div[@id='advSearch-noJS']//select[@id='sf_languages']/option"):
eng_lang = x.get("value")

View File

@@ -1,9 +1,9 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Exception types raised by SearXNG modules.
"""
from __future__ import annotations
from typing import Optional, Union
import typing as t
from lxml.etree import XPath
class SearxException(Exception):
@@ -13,21 +13,22 @@ class SearxException(Exception):
class SearxParameterException(SearxException):
"""Raised when query miss a required parameter"""
def __init__(self, name, value):
def __init__(self, name: str, value: t.Any):
if value == '' or value is None:
message = 'Empty ' + name + ' parameter'
message = f"Empty {name} parameter"
else:
message = 'Invalid value "' + value + '" for parameter ' + name
message = f"Invalid value {value} for parameter {name}"
super().__init__(message)
self.message = message
self.parameter_name = name
self.parameter_value = value
self.message: str = message
self.parameter_name: str = name
self.parameter_value: t.Any = value
@t.final
class SearxSettingsException(SearxException):
"""Error while loading the settings"""
def __init__(self, message: Union[str, Exception], filename: Optional[str]):
def __init__(self, message: str | Exception, filename: str | None):
super().__init__(message)
self.message = message
self.filename = filename
@@ -40,11 +41,11 @@ class SearxEngineException(SearxException):
class SearxXPathSyntaxException(SearxEngineException):
"""Syntax error in a XPATH"""
def __init__(self, xpath_spec, message):
def __init__(self, xpath_spec: str | XPath, message: str):
super().__init__(str(xpath_spec) + " " + message)
self.message = message
self.message: str = message
# str(xpath_spec) to deal with str and XPath instance
self.xpath_str = str(xpath_spec)
self.xpath_str: str = str(xpath_spec)
class SearxEngineResponseException(SearxEngineException):
@@ -58,7 +59,7 @@ class SearxEngineAPIException(SearxEngineResponseException):
class SearxEngineAccessDeniedException(SearxEngineResponseException):
"""The website is blocking the access"""
SUSPEND_TIME_SETTING = "search.suspended_times.SearxEngineAccessDenied"
SUSPEND_TIME_SETTING: str = "search.suspended_times.SearxEngineAccessDenied"
"""This settings contains the default suspended time (default 86400 sec / 1
day)."""
@@ -74,8 +75,8 @@ class SearxEngineAccessDeniedException(SearxEngineResponseException):
if suspended_time is None:
suspended_time = self._get_default_suspended_time()
super().__init__(message + ', suspended_time=' + str(suspended_time))
self.suspended_time = suspended_time
self.message = message
self.suspended_time: int = suspended_time
self.message: str = message
def _get_default_suspended_time(self) -> int:
from searx import get_setting # pylint: disable=C0415
@@ -86,11 +87,11 @@ class SearxEngineAccessDeniedException(SearxEngineResponseException):
class SearxEngineCaptchaException(SearxEngineAccessDeniedException):
"""The website has returned a CAPTCHA."""
SUSPEND_TIME_SETTING = "search.suspended_times.SearxEngineCaptcha"
SUSPEND_TIME_SETTING: str = "search.suspended_times.SearxEngineCaptcha"
"""This settings contains the default suspended time (default 86400 sec / 1
day)."""
def __init__(self, suspended_time: int | None = None, message='CAPTCHA'):
def __init__(self, suspended_time: int | None = None, message: str = 'CAPTCHA'):
super().__init__(message=message, suspended_time=suspended_time)
@@ -100,19 +101,19 @@ class SearxEngineTooManyRequestsException(SearxEngineAccessDeniedException):
By default, SearXNG stops sending requests to this engine for 1 hour.
"""
SUSPEND_TIME_SETTING = "search.suspended_times.SearxEngineTooManyRequests"
SUSPEND_TIME_SETTING: str = "search.suspended_times.SearxEngineTooManyRequests"
"""This settings contains the default suspended time (default 3660 sec / 1
hour)."""
def __init__(self, suspended_time: int | None = None, message='Too many request'):
def __init__(self, suspended_time: int | None = None, message: str = 'Too many request'):
super().__init__(message=message, suspended_time=suspended_time)
class SearxEngineXPathException(SearxEngineResponseException):
"""Error while getting the result of an XPath expression"""
def __init__(self, xpath_spec, message):
def __init__(self, xpath_spec: str | XPath, message: str):
super().__init__(str(xpath_spec) + " " + message)
self.message = message
self.message: str = message
# str(xpath_spec) to deal with str and XPath instance
self.xpath_str = str(xpath_spec)
self.xpath_str: str = str(xpath_spec)

View File

@@ -20,7 +20,6 @@
"""
# pylint: disable=invalid-name
from __future__ import annotations
__all__ = ["SXNG_Request", "sxng_request", "SXNG_Response"]
@@ -62,6 +61,8 @@ class SXNG_Request(flask.Request):
"""A list of :py:obj:`searx.results.Timing` of the engines, calculatid in
and hold by :py:obj:`searx.results.ResultContainer.timings`."""
remote_addr: str
#: A replacement for :py:obj:`flask.request` with type cast :py:`SXNG_Request`.
sxng_request = typing.cast(SXNG_Request, flask.request)

View File

@@ -1,13 +1,20 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring
__all__ = ["get_bang_url"]
import typing as t
from urllib.parse import quote_plus, urlparse
from searx.data import EXTERNAL_BANGS
LEAF_KEY = chr(16)
if t.TYPE_CHECKING:
from searx.search.models import SearchQuery
def get_node(external_bangs_db, bang):
def get_node(external_bangs_db: dict[str, t.Any], bang: str):
node = external_bangs_db['trie']
after = ''
before = ''
@@ -20,7 +27,7 @@ def get_node(external_bangs_db, bang):
return node, before, after
def get_bang_definition_and_ac(external_bangs_db, bang):
def get_bang_definition_and_ac(external_bangs_db: dict[str, t.Any], bang: str):
node, before, after = get_node(external_bangs_db, bang)
bang_definition = None
@@ -39,7 +46,7 @@ def get_bang_definition_and_ac(external_bangs_db, bang):
return bang_definition, bang_ac_list
def resolve_bang_definition(bang_definition, query):
def resolve_bang_definition(bang_definition: str, query: str) -> tuple[str, int]:
url, rank = bang_definition.split(chr(1))
if url.startswith('//'):
url = 'https:' + url
@@ -54,7 +61,9 @@ def resolve_bang_definition(bang_definition, query):
return (url, rank)
def get_bang_definition_and_autocomplete(bang, external_bangs_db=None): # pylint: disable=invalid-name
def get_bang_definition_and_autocomplete(
bang: str, external_bangs_db: dict[str, t.Any] | None = None
): # pylint: disable=invalid-name
if external_bangs_db is None:
external_bangs_db = EXTERNAL_BANGS
@@ -81,7 +90,7 @@ def get_bang_definition_and_autocomplete(bang, external_bangs_db=None): # pylin
return bang_definition, new_autocomplete
def get_bang_url(search_query, external_bangs_db=None):
def get_bang_url(search_query: "SearchQuery", external_bangs_db: dict[str, t.Any] | None = None) -> str | None:
"""
Redirects if the user supplied a correct bang search.
:param search_query: This is a search_query object which contains preferences and the submitted queries.

View File

@@ -8,7 +8,6 @@ an example in which the command line is called in the development environment::
(py3) python -m searx.favicons --help
"""
from __future__ import annotations
__all__ = ["init", "favicon_url", "favicon_proxy"]

View File

@@ -17,8 +17,7 @@
"""
from __future__ import annotations
from typing import Literal
import typing as t
import os
import abc
@@ -90,10 +89,11 @@ def init(cfg: "FaviconCacheConfig"):
raise NotImplementedError(f"favicons db_type '{cfg.db_type}' is unknown")
@t.final
class FaviconCacheConfig(msgspec.Struct): # pylint: disable=too-few-public-methods
"""Configuration of the favicon cache."""
db_type: Literal["sqlite", "mem"] = "sqlite"
db_type: t.Literal["sqlite", "mem"] = "sqlite"
"""Type of the database:
``sqlite``:
@@ -125,7 +125,7 @@ class FaviconCacheConfig(msgspec.Struct): # pylint: disable=too-few-public-meth
"""Maintenance period in seconds / when :py:obj:`MAINTENANCE_MODE` is set to
``auto``."""
MAINTENANCE_MODE: Literal["auto", "off"] = "auto"
MAINTENANCE_MODE: t.Literal["auto", "off"] = "auto"
"""Type of maintenance mode
``auto``:
@@ -147,14 +147,14 @@ class FaviconCacheStats:
domains: int | None = None
resolvers: int | None = None
field_descr = (
field_descr: tuple[tuple[str, str, t.Callable[[int, int], str] | type], ...] = (
("favicons", "number of favicons in cache", humanize_number),
("bytes", "total size (approx. bytes) of cache", humanize_bytes),
("domains", "total number of domains in cache", humanize_number),
("resolvers", "number of resolvers", str),
)
def __sub__(self, other) -> FaviconCacheStats:
def __sub__(self, other: "FaviconCacheStats") -> "FaviconCacheStats":
if not isinstance(other, self.__class__):
raise TypeError(f"unsupported operand type(s) for +: '{self.__class__}' and '{type(other)}'")
kwargs = {}
@@ -166,17 +166,17 @@ class FaviconCacheStats:
kwargs[field] = self_val - other_val
else:
kwargs[field] = self_val
return self.__class__(**kwargs)
return self.__class__(**kwargs) # type: ignore
def report(self, fmt: str = "{descr}: {val}\n"):
s = []
s: list[str] = []
for field, descr, cast in self.field_descr:
val = getattr(self, field)
val: str | None = getattr(self, field)
if val is None:
val = "--"
else:
val = cast(val)
s.append(fmt.format(descr=descr, val=val))
val = cast(val) # type: ignore
s.append(fmt.format(descr=descr, val=val)) # pyright: ignore[reportUnknownArgumentType]
return "".join(s)
@@ -204,10 +204,11 @@ class FaviconCache(abc.ABC):
on the state of the cache."""
@abc.abstractmethod
def maintenance(self, force=False):
def maintenance(self, force: bool = False):
"""Performs maintenance on the cache"""
@t.final
class FaviconCacheNull(FaviconCache):
"""A dummy favicon cache that caches nothing / a fallback solution. The
NullCache is used when more efficient caches such as the
@@ -227,11 +228,12 @@ class FaviconCacheNull(FaviconCache):
def state(self):
return FaviconCacheStats(favicons=0)
def maintenance(self, force=False):
def maintenance(self, force: bool = False):
pass
class FaviconCacheSQLite(sqlitedb.SQLiteAppl, FaviconCache):
@t.final
class FaviconCacheSQLite(sqlitedb.SQLiteAppl, FaviconCache): # pyright: ignore[reportUnsafeMultipleInheritance]
"""Favicon cache that manages the favicon BLOBs in a SQLite DB. The DB
model in the SQLite DB is implemented using the abstract class
:py:obj:`sqlitedb.SQLiteAppl`.
@@ -376,7 +378,7 @@ CREATE TABLE IF NOT EXISTS blob_map (
return self.cfg.MAINTENANCE_PERIOD + self.properties.m_time("LAST_MAINTENANCE")
def maintenance(self, force=False):
def maintenance(self, force: bool = False):
# Prevent parallel DB maintenance cycles from other DB connections
# (e.g. in multi thread or process environments).
@@ -406,7 +408,7 @@ CREATE TABLE IF NOT EXISTS blob_map (
x = total_bytes - self.cfg.LIMIT_TOTAL_BYTES
c = 0
sha_list = []
sha_list: list[str] = []
for row in conn.execute(self.SQL_ITER_BLOBS_SHA256_BYTES_C):
sha256, bytes_c = row
sha_list.append(sha256)
@@ -424,7 +426,7 @@ CREATE TABLE IF NOT EXISTS blob_map (
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
conn.close()
def _query_val(self, sql, default=None):
def _query_val(self, sql: str, default: t.Any = None):
val = self.DB.execute(sql).fetchone()
if val is not None:
val = val[0]
@@ -441,6 +443,7 @@ CREATE TABLE IF NOT EXISTS blob_map (
)
@t.final
class FaviconCacheMEM(FaviconCache):
"""Favicon cache in process' memory. Its just a POC that stores the
favicons in the memory of the process.
@@ -451,11 +454,11 @@ class FaviconCacheMEM(FaviconCache):
"""
def __init__(self, cfg):
def __init__(self, cfg: FaviconCacheConfig):
self.cfg = cfg
self._data = {}
self._sha_mime = {}
self._data: dict[str, t.Any] = {}
self._sha_mime: dict[str, tuple[str, str | None]] = {}
def __call__(self, resolver: str, authority: str) -> None | tuple[bytes | None, str | None]:
@@ -489,5 +492,5 @@ class FaviconCacheMEM(FaviconCache):
def state(self):
return FaviconCacheStats(favicons=len(self._data.keys()))
def maintenance(self, force=False):
def maintenance(self, force: bool = False):
pass

View File

@@ -1,7 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# pylint: disable=missing-module-docstring
from __future__ import annotations
import pathlib
import msgspec

View File

@@ -1,7 +1,6 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Implementations for a favicon proxy"""
from __future__ import annotations
from typing import Callable
@@ -24,10 +23,10 @@ from .resolvers import DEFAULT_RESOLVER_MAP
from . import cache
DEFAULT_FAVICON_URL = {}
CFG: FaviconProxyConfig = None # type: ignore
CFG: "FaviconProxyConfig" = None # type: ignore
def init(cfg: FaviconProxyConfig):
def init(cfg: "FaviconProxyConfig"):
global CFG # pylint: disable=global-statement
CFG = cfg

View File

@@ -6,7 +6,6 @@ timeout``) and returns a tuple ``(data, mime)``.
"""
from __future__ import annotations
__all__ = ["DEFAULT_RESOLVER_MAP", "allesedv", "duckduckgo", "google", "yandex"]

View File

@@ -18,14 +18,13 @@ Usage in a Flask app route:
"""
from __future__ import annotations
__all__ = ['InfoPage', 'InfoPageSet']
import typing as t
import os
import os.path
import logging
import typing
import urllib.parse
from functools import cached_property
@@ -43,7 +42,7 @@ _INFO_FOLDER = os.path.abspath(os.path.dirname(__file__))
INFO_PAGES: 'InfoPageSet'
def __getattr__(name):
def __getattr__(name: str):
if name == 'INFO_PAGES':
global INFO_PAGES # pylint: disable=global-statement
INFO_PAGES = InfoPageSet()
@@ -55,8 +54,8 @@ def __getattr__(name):
class InfoPage:
"""A page of the :py:obj:`online documentation <InfoPageSet>`."""
def __init__(self, fname):
self.fname = fname
def __init__(self, fname: str):
self.fname: str = fname
@cached_property
def raw_content(self):
@@ -74,14 +73,14 @@ class InfoPage:
@cached_property
def title(self):
"""Title of the content (without any markup)"""
t = ""
_t = ""
for l in self.raw_content.split('\n'):
if l.startswith('# '):
t = l.strip('# ')
return t
_t = l.strip('# ')
return _t
@cached_property
def html(self):
def html(self) -> str:
"""Render Markdown (CommonMark_) to HTML by using markdown-it-py_.
.. _CommonMark: https://commonmark.org/
@@ -92,18 +91,18 @@ class InfoPage:
MarkdownIt("commonmark", {"typographer": True}).enable(["replacements", "smartquotes"]).render(self.content)
)
def get_ctx(self):
def get_ctx(self) -> dict[str, str]:
"""Jinja context to render :py:obj:`InfoPage.content`"""
def _md_link(name, url):
def _md_link(name: str, url: str):
url = url_for(url, _external=True)
return "[%s](%s)" % (name, url)
def _md_search(query):
def _md_search(query: str):
url = '%s?q=%s' % (url_for('search', _external=True), urllib.parse.quote(query))
return '[%s](%s)' % (query, url)
ctx = {}
ctx: dict[str, t.Any] = {}
ctx['GIT_URL'] = GIT_URL
ctx['get_setting'] = get_setting
ctx['link'] = _md_link
@@ -125,31 +124,29 @@ class InfoPageSet: # pylint: disable=too-few-public-methods
:type info_folder: str
"""
def __init__(
self, page_class: typing.Optional[typing.Type[InfoPage]] = None, info_folder: typing.Optional[str] = None
):
self.page_class = page_class or InfoPage
def __init__(self, page_class: type[InfoPage] | None = None, info_folder: str | None = None):
self.page_class: type[InfoPage] = page_class or InfoPage
self.folder: str = info_folder or _INFO_FOLDER
"""location of the Markdown files"""
self.CACHE: typing.Dict[tuple, typing.Optional[InfoPage]] = {}
self.CACHE: dict[tuple[str, str], InfoPage | None] = {}
self.locale_default: str = 'en'
"""default language"""
self.locales: typing.List[str] = [
self.locales: list[str] = [
locale.replace('_', '-') for locale in os.listdir(_INFO_FOLDER) if locale.replace('_', '-') in LOCALE_NAMES
]
"""list of supported languages (aka locales)"""
self.toc: typing.List[str] = [
self.toc: list[str] = [
'search-syntax',
'about',
'donate',
]
"""list of articles in the online documentation"""
def get_page(self, pagename: str, locale: typing.Optional[str] = None):
def get_page(self, pagename: str, locale: str | None = None):
"""Return ``pagename`` instance of :py:obj:`InfoPage`
:param pagename: name of the page, a value from :py:obj:`InfoPageSet.toc`
@@ -184,7 +181,7 @@ class InfoPageSet: # pylint: disable=too-few-public-methods
self.CACHE[cache_key] = page
return page
def iter_pages(self, locale: typing.Optional[str] = None, fallback_to_default=False):
def iter_pages(self, locale: str | None = None, fallback_to_default: bool = False):
"""Iterate over all pages of the TOC"""
locale = locale or self.locale_default
for page_name in self.toc:

View File

@@ -92,7 +92,6 @@ Implementation
"""
from __future__ import annotations
from ipaddress import ip_address
import sys
@@ -124,7 +123,7 @@ from searx.botdetection import (
# coherency, the logger is "limiter"
logger = logger.getChild('limiter')
CFG: config.Config | None = None # type: ignore
CFG: config.Config | None = None
_INSTALLED = False
LIMITER_CFG_SCHEMA = Path(__file__).parent / "limiter.toml"

View File

@@ -26,15 +26,15 @@ SearXNGs locale implementations
================================
"""
from __future__ import annotations
import typing as t
from pathlib import Path
import babel
from babel.support import Translations
import babel.languages
import babel.core
import flask_babel
import flask_babel # pyright: ignore[reportMissingTypeStubs]
from flask.ctx import has_request_context
from searx import (
@@ -50,7 +50,7 @@ logger = logger.getChild('locales')
# safe before monkey patching flask_babel.get_translations
_flask_babel_get_translations = flask_babel.get_translations
LOCALE_NAMES = {}
LOCALE_NAMES: dict[str, str] = {}
"""Mapping of locales and their description. Locales e.g. 'fr' or 'pt-BR' (see
:py:obj:`locales_initialize`).
@@ -84,9 +84,9 @@ Kong."""
def localeselector():
locale = 'en'
locale: str = 'en'
if has_request_context():
value = sxng_request.preferences.get_value('locale')
value: str = sxng_request.preferences.get_value('locale')
if value:
locale = value
@@ -128,7 +128,7 @@ def get_translation_locales() -> list[str]:
if _TR_LOCALES:
return _TR_LOCALES
tr_locales = []
tr_locales: list[str] = []
for folder in (Path(searx_dir) / 'translations').iterdir():
if not folder.is_dir():
continue
@@ -179,7 +179,7 @@ def get_locale(locale_tag: str) -> babel.Locale | None:
def get_official_locales(
territory: str, languages=None, regional: bool = False, de_facto: bool = True
territory: str, languages: list[str] | None = None, regional: bool = False, de_facto: bool = True
) -> set[babel.Locale]:
"""Returns a list of :py:obj:`babel.Locale` with languages from
:py:obj:`babel.languages.get_official_languages`.
@@ -198,7 +198,7 @@ def get_official_locales(
which are “de facto” official are not returned.
"""
ret_val = set()
ret_val: set[babel.Locale] = set()
o_languages = babel.languages.get_official_languages(territory, regional=regional, de_facto=de_facto)
if languages:
@@ -215,7 +215,7 @@ def get_official_locales(
return ret_val
def get_engine_locale(searxng_locale, engine_locales, default=None):
def get_engine_locale(searxng_locale: str, engine_locales: dict[str, str], default: str | None = None) -> str | None:
"""Return engine's language (aka locale) string that best fits to argument
``searxng_locale``.
@@ -312,11 +312,14 @@ def get_engine_locale(searxng_locale, engine_locales, default=None):
if locale.language:
terr_lang_dict = {}
terr_lang_dict: dict[str, dict[str, t.Any]] = {}
territory: str
langs: dict[str, dict[str, t.Any]]
for territory, langs in babel.core.get_global("territory_languages").items():
if not langs.get(searxng_lang, {}).get('official_status'):
_lang = langs.get(searxng_lang)
if _lang is None or _lang.get('official_status') is None:
continue
terr_lang_dict[territory] = langs.get(searxng_lang)
terr_lang_dict[territory] = _lang
# first: check fr-FR, de-DE .. is supported by the engine
# exception: 'en' --> 'en-US'
@@ -347,7 +350,7 @@ def get_engine_locale(searxng_locale, engine_locales, default=None):
# - 'fr-MF', 'population_percent': 100.0, 'official_status': 'official'
# - 'fr-BE', 'population_percent': 38.0, 'official_status': 'official'
terr_lang_list = []
terr_lang_list: list[tuple[str, dict[str, t.Any]]] = []
for k, v in terr_lang_dict.items():
terr_lang_list.append((k, v))
@@ -404,7 +407,7 @@ def match_locale(searxng_locale: str, locale_tag_list: list[str], fallback: str
# clean up locale_tag_list
tag_list = []
tag_list: list[str] = []
for tag in locale_tag_list:
if tag in ('all', 'auto') or tag in ADDITIONAL_TRANSLATIONS:
continue
@@ -415,7 +418,7 @@ def match_locale(searxng_locale: str, locale_tag_list: list[str], fallback: str
return get_engine_locale(searxng_locale, engine_locales, default=fallback)
def build_engine_locales(tag_list: list[str]):
def build_engine_locales(tag_list: list[str]) -> dict[str, str]:
"""From a list of locale tags a dictionary is build that can be passed by
argument ``engine_locales`` to :py:obj:`get_engine_locale`. This function
is mainly used by :py:obj:`match_locale` and is similar to what the
@@ -445,7 +448,7 @@ def build_engine_locales(tag_list: list[str]):
be assigned to the **regions** that SearXNG supports.
"""
engine_locales = {}
engine_locales: dict[str, str] = {}
for tag in tag_list:
locale = get_locale(tag)

Some files were not shown because too many files have changed in this diff Show More