[fix] engine: re-implement mullvad leta integration

searx/engines/mullvad_leta.py

@@ -1,34 +1,14 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 """This is the implementation of the Mullvad-Leta meta-search engine.
-
-This engine **REQUIRES** that searxng operate within a Mullvad VPN
-
-If using docker, consider using gluetun for easily connecting to the Mullvad
-
-- https://github.com/qdm12/gluetun
-
-Otherwise, follow instructions provided by Mullvad for enabling the VPN on Linux
-
-- https://mullvad.net/en/help/install-mullvad-app-linux
-
-.. hint::
-
-   The :py:obj:`EngineTraits` is empty by default.  Maintainers have to run
-   ``make data.traits`` (in the Mullvad VPN / :py:obj:`fetch_traits`) and rebase
-   the modified JSON file ``searx/data/engine_traits.json`` on every single
-   update of SearXNG!
 """
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, TypedDict
+from urllib.parse import urlencode
 from httpx import Response
-from lxml import html
 from searx.enginelib.traits import EngineTraits
-from searx.locales import region_tag, get_official_locales
-from searx.utils import eval_xpath, extract_text, eval_xpath_list
-from searx.exceptions import SearxEngineResponseException
 
 if TYPE_CHECKING:
     import logging
@@ -37,15 +17,13 @@ if TYPE_CHECKING:
 
 traits: EngineTraits
 
-use_cache: bool = True  # non-cache use only has 100 searches per day!
-
 leta_engine: str = 'google'
 
-search_url = "https://leta.mullvad.net"
+base_url = "https://leta.mullvad.net"
 
 # about
 about = {
-    "website": search_url,
+    "website": base_url,
     "wikidata_id": 'Q47008412',  # the Mullvad id - not leta, but related
     "official_api_documentation": 'https://leta.mullvad.net/faq',
     "use_official_api": False,
@@ -56,13 +34,13 @@ about = {
 # engine dependent config
 categories = ['general', 'web']
 paging = True
-max_page = 50
+max_page = 10
 time_range_support = True
 time_range_dict = {
-    "day": "d1",
+    "day": "d",
-    "week": "w1",
+    "week": "w",
-    "month": "m1",
+    "month": "m",
-    "year": "y1",
+    "year": "y",
 }
 
 available_leta_engines = [
@@ -71,19 +49,32 @@ available_leta_engines = [
 ]
 
 
-def is_vpn_connected(dom: html.HtmlElement) -> bool:
+class DataNodeQueryMetaDataIndices(TypedDict):
-    """Returns true if the VPN is connected, False otherwise"""
+    """Indices into query metadata"""
-    connected_text = extract_text(eval_xpath(dom, '//main/div/p[1]'))
+
-    return connected_text != 'You are not connected to Mullvad VPN.'
+    success: int
+    q: int  # pylint: disable=invalid-name
+    country: int
+    language: int
+    lastUpdated: int
+    engine: int
+    items: int
+    infobox: int
+    news: int
+    timestamp: int
+    altered: int
+    page: int
+    next: int  # if -1, there no more results are available
+    previous: int
 
 
-def assign_headers(headers: dict) -> dict:
+class DataNodeResultIndices(TypedDict):
-    """Assigns the headers to make a request to Mullvad Leta"""
+    """Indices into query resultsdata"""
-    headers['Accept'] = "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8"
+
-    headers['Content-Type'] = "application/x-www-form-urlencoded"
+    link: int
-    headers['Host'] = "leta.mullvad.net"
+    snippet: int
-    headers['Origin'] = "https://leta.mullvad.net"
+    title: int
-    return headers
+    favicon: int
 
 
 def request(query: str, params: dict):
@@ -99,109 +90,55 @@ def request(query: str, params: dict):
             result_engine,
         )
 
-    params['url'] = search_url
+    params['method'] = 'GET'
-    params['method'] = 'POST'
+
-    params['data'] = {
+    args = {
-        "q": query,
+        'q': query,
-        "gl": country if country is str else '',
         'engine': result_engine,
+        'x-sveltekit-invalidated': "001",  # hardcoded from all requests seen
     }
-    # pylint: disable=undefined-variable
+    if isinstance(country, str):
-    if use_cache:
+        args['country'] = country
-        params['data']['oc'] = "on"
-    # pylint: enable=undefined-variable
-
     if params['time_range'] in time_range_dict:
-        params['dateRestrict'] = time_range_dict[params['time_range']]
+        args['lastUpdated'] = time_range_dict[params['time_range']]
-    else:
-        params['dateRestrict'] = ''
-
     if params['pageno'] > 1:
-        #  Page 1 is n/a, Page 2 is 11, page 3 is 21, ...
+        args['page'] = params['pageno']
-        params['data']['start'] = ''.join([str(params['pageno'] - 1), "1"])
 
-    if params['headers'] is None:
+    params['url'] = f"{base_url}/search/__data.json?{urlencode(args)}"
-        params['headers'] = {}
 
-    assign_headers(params['headers'])
     return params
 
 
-def extract_result(dom_result: list[html.HtmlElement]):
-    # Infoboxes sometimes appear in the beginning and will have a length of 0
-    if len(dom_result) == 3:
-        [a_elem, h3_elem, p_elem] = dom_result
-    elif len(dom_result) == 4:
-        [_, a_elem, h3_elem, p_elem] = dom_result
-    else:
-        return None
-
-    return {
-        'url': extract_text(a_elem.text),
-        'title': extract_text(h3_elem),
-        'content': extract_text(p_elem),
-    }
-
-
-def extract_results(search_results: html.HtmlElement):
-    for search_result in search_results:
-        dom_result = eval_xpath_list(search_result, 'div/div/*')
-        result = extract_result(dom_result)
-        if result is not None:
-            yield result
-
-
 def response(resp: Response):
-    """Checks if connected to Mullvad VPN, then extracts the search results from
+    json_response = resp.json()
-    the DOM resp: requests response object"""
 
-    dom = html.fromstring(resp.text)
+    nodes = json_response["nodes"]
-    if not is_vpn_connected(dom):
+    # 0: is None
-        raise SearxEngineResponseException('Not connected to Mullvad VPN')
+    # 1: has "connected=True", not useful
-    search_results = eval_xpath(dom.body, '//main/div[2]/div')
+    # 2: query results within "data"
-    return list(extract_results(search_results))
 
+    data_nodes = nodes[2]["data"]
+    # Instead of nested object structure, all objects are flattened into a
+    # list. Rather, the first object in data_node provides indices into the
+    # "data_nodes" to access each searchresult (which is an object of more
+    # indices)
+    #
+    # Read the relative TypedDict definitions for details
 
-def fetch_traits(engine_traits: EngineTraits):
+    query_meta_data: DataNodeQueryMetaDataIndices = data_nodes[0]
-    """Fetch languages and regions from Mullvad-Leta
 
-    .. warning::
+    query_items_indices = query_meta_data['items']
 
-        Fetching the engine traits also requires a Mullvad VPN connection. If
+    results = []
-        not connected, then an error message will print and no traits will be
+    for idx in data_nodes[query_items_indices]:
-        updated.
+        query_item_indices: DataNodeResultIndices = data_nodes[idx]
-    """
+        results.append(
-    # pylint: disable=import-outside-toplevel
+            {
-    # see https://github.com/searxng/searxng/issues/762
+                'url': data_nodes[query_item_indices['link']],
-    from searx.network import post as http_post
+                'title': data_nodes[query_item_indices['title']],
+                'content': data_nodes[query_item_indices['snippet']],
+                'favicon_url': base_url + "/" + data_nodes[query_item_indices['favicon']],
+            }
+        )
 
-    # pylint: enable=import-outside-toplevel
+    return results
-    resp = http_post(search_url, headers=assign_headers({}))
-    if not isinstance(resp, Response):
-        print("ERROR: failed to get response from mullvad-leta. Are you connected to the VPN?")
-        return
-    if not resp.ok:
-        print("ERROR: response from mullvad-leta is not OK. Are you connected to the VPN?")
-        return
-    dom = html.fromstring(resp.text)
-    if not is_vpn_connected(dom):
-        print('ERROR: Not connected to Mullvad VPN')
-        return
-    # supported region codes
-    options = eval_xpath_list(dom.body, '//main/div/form/div[2]/div/select[1]/option')
-    if options is None or len(options) <= 0:
-        print('ERROR: could not find any results. Are you connected to the VPN?')
-    for x in options:
-        eng_country = x.get("value")
-
-        sxng_locales = get_official_locales(eng_country, engine_traits.languages.keys(), regional=True)
-
-        if not sxng_locales:
-            print(
-                "ERROR: can't map from Mullvad-Leta country %s (%s) to a babel region."
-                % (x.get('data-name'), eng_country)
-            )
-            continue
-
-        for sxng_locale in sxng_locales:
-            engine_traits.regions[region_tag(sxng_locale)] = eng_country

searx/settings.yml

@@ -1398,11 +1398,9 @@ engines:
       require_api_key: false
       results: JSON
 
-  # read https://docs.searxng.org/dev/engines/online/mullvad_leta.html
   # - name: mullvadleta
   #   engine: mullvad_leta
   #   leta_engine: google # choose one of the following: google, brave
-  #   use_cache: true  # Only 100 non-cache searches per day, suggested only for private instances
   #   search_url: https://leta.mullvad.net
   #   categories: [general, web]
   #   shortcut: ml