|
@@ -20,6 +20,8 @@ Otherwise, follow instructions provided by Mullvad for enabling the VPN on Linux
|
|
update of SearXNG!
|
|
update of SearXNG!
|
|
"""
|
|
"""
|
|
|
|
|
|
|
|
+from __future__ import annotations
|
|
|
|
+
|
|
from typing import TYPE_CHECKING
|
|
from typing import TYPE_CHECKING
|
|
from httpx import Response
|
|
from httpx import Response
|
|
from lxml import html
|
|
from lxml import html
|
|
@@ -37,6 +39,8 @@ traits: EngineTraits
|
|
|
|
|
|
use_cache: bool = True # non-cache use only has 100 searches per day!
|
|
use_cache: bool = True # non-cache use only has 100 searches per day!
|
|
|
|
|
|
|
|
+leta_engine: str = 'google'
|
|
|
|
+
|
|
search_url = "https://leta.mullvad.net"
|
|
search_url = "https://leta.mullvad.net"
|
|
|
|
|
|
# about
|
|
# about
|
|
@@ -61,6 +65,11 @@ time_range_dict = {
|
|
"year": "y1",
|
|
"year": "y1",
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+available_leta_engines = [
|
|
|
|
+ 'google', # first will be default if provided engine is invalid
|
|
|
|
+ 'brave',
|
|
|
|
+]
|
|
|
|
+
|
|
|
|
|
|
def is_vpn_connected(dom: html.HtmlElement) -> bool:
|
|
def is_vpn_connected(dom: html.HtmlElement) -> bool:
|
|
"""Returns true if the VPN is connected, False otherwise"""
|
|
"""Returns true if the VPN is connected, False otherwise"""
|
|
@@ -80,11 +89,22 @@ def assign_headers(headers: dict) -> dict:
|
|
def request(query: str, params: dict):
|
|
def request(query: str, params: dict):
|
|
country = traits.get_region(params.get('searxng_locale', 'all'), traits.all_locale) # type: ignore
|
|
country = traits.get_region(params.get('searxng_locale', 'all'), traits.all_locale) # type: ignore
|
|
|
|
|
|
|
|
+ result_engine = leta_engine
|
|
|
|
+ if leta_engine not in available_leta_engines:
|
|
|
|
+ result_engine = available_leta_engines[0]
|
|
|
|
+ logger.warning(
|
|
|
|
+ 'Configured engine "%s" not one of the available engines %s, defaulting to "%s"',
|
|
|
|
+ leta_engine,
|
|
|
|
+ available_leta_engines,
|
|
|
|
+ result_engine,
|
|
|
|
+ )
|
|
|
|
+
|
|
params['url'] = search_url
|
|
params['url'] = search_url
|
|
params['method'] = 'POST'
|
|
params['method'] = 'POST'
|
|
params['data'] = {
|
|
params['data'] = {
|
|
"q": query,
|
|
"q": query,
|
|
"gl": country if country is str else '',
|
|
"gl": country if country is str else '',
|
|
|
|
+ 'engine': result_engine,
|
|
}
|
|
}
|
|
# pylint: disable=undefined-variable
|
|
# pylint: disable=undefined-variable
|
|
if use_cache:
|
|
if use_cache:
|
|
@@ -107,8 +127,8 @@ def request(query: str, params: dict):
|
|
return params
|
|
return params
|
|
|
|
|
|
|
|
|
|
-def extract_result(dom_result: html.HtmlElement):
|
|
|
|
- [a_elem, h3_elem, p_elem] = eval_xpath_list(dom_result, 'div/div/*')
|
|
|
|
|
|
+def extract_result(dom_result: list[html.HtmlElement]):
|
|
|
|
+ [a_elem, h3_elem, p_elem] = dom_result
|
|
return {
|
|
return {
|
|
'url': extract_text(a_elem.text),
|
|
'url': extract_text(a_elem.text),
|
|
'title': extract_text(h3_elem),
|
|
'title': extract_text(h3_elem),
|
|
@@ -116,6 +136,14 @@ def extract_result(dom_result: html.HtmlElement):
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
+def extract_results(search_results: html.HtmlElement):
|
|
|
|
+ for search_result in search_results:
|
|
|
|
+ dom_result = eval_xpath_list(search_result, 'div/div/*')
|
|
|
|
+ # sometimes an info box pops up, will need to filter that out
|
|
|
|
+ if len(dom_result) == 3:
|
|
|
|
+ yield extract_result(dom_result)
|
|
|
|
+
|
|
|
|
+
|
|
def response(resp: Response):
|
|
def response(resp: Response):
|
|
"""Checks if connected to Mullvad VPN, then extracts the search results from
|
|
"""Checks if connected to Mullvad VPN, then extracts the search results from
|
|
the DOM resp: requests response object"""
|
|
the DOM resp: requests response object"""
|
|
@@ -124,7 +152,7 @@ def response(resp: Response):
|
|
if not is_vpn_connected(dom):
|
|
if not is_vpn_connected(dom):
|
|
raise SearxEngineResponseException('Not connected to Mullvad VPN')
|
|
raise SearxEngineResponseException('Not connected to Mullvad VPN')
|
|
search_results = eval_xpath(dom.body, '//main/div[2]/div')
|
|
search_results = eval_xpath(dom.body, '//main/div[2]/div')
|
|
- return [extract_result(sr) for sr in search_results]
|
|
|
|
|
|
+ return list(extract_results(search_results))
|
|
|
|
|
|
|
|
|
|
def fetch_traits(engine_traits: EngineTraits):
|
|
def fetch_traits(engine_traits: EngineTraits):
|