summaryrefslogtreecommitdiff
path: root/searx/autocomplete.py
diff options
context:
space:
mode:
Diffstat (limited to 'searx/autocomplete.py')
-rw-r--r--searx/autocomplete.py184
1 files changed, 87 insertions, 97 deletions
diff --git a/searx/autocomplete.py b/searx/autocomplete.py
index 31121c7e7..f887e5c56 100644
--- a/searx/autocomplete.py
+++ b/searx/autocomplete.py
@@ -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 []
generated by cgit v1.2.3 (git 2.25.1) at 2026-03-03 18:43:03 +0000