From 5a061d443a7225cb806f892b3c4920bdf0679b2a Mon Sep 17 00:00:00 2001 From: Grant Lanham Date: Sat, 12 Apr 2025 22:17:09 -0400 Subject: [PATCH] [fix] engine: re-implement mullvad leta integration --- searx/engines/mullvad_leta.py | 201 ++++++++++++---------------------- searx/settings.yml | 2 - 2 files changed, 69 insertions(+), 134 deletions(-) diff --git a/searx/engines/mullvad_leta.py b/searx/engines/mullvad_leta.py index 7e2b590e8..8939cbeed 100644 --- a/searx/engines/mullvad_leta.py +++ b/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", - "week": "w1", - "month": "m1", - "year": "y1", + "day": "d", + "week": "w", + "month": "m", + "year": "y", } available_leta_engines = [ @@ -71,19 +49,32 @@ available_leta_engines = [ ] -def is_vpn_connected(dom: html.HtmlElement) -> bool: - """Returns true if the VPN is connected, False otherwise""" - connected_text = extract_text(eval_xpath(dom, '//main/div/p[1]')) - return connected_text != 'You are not connected to Mullvad VPN.' +class DataNodeQueryMetaDataIndices(TypedDict): + """Indices into query metadata""" + + 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: - """Assigns the headers to make a request to Mullvad Leta""" - 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" - headers['Host'] = "leta.mullvad.net" - headers['Origin'] = "https://leta.mullvad.net" - return headers +class DataNodeResultIndices(TypedDict): + """Indices into query resultsdata""" + + link: int + snippet: int + title: int + 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'] = 'POST' - params['data'] = { - "q": query, - "gl": country if country is str else '', + params['method'] = 'GET' + + args = { + 'q': query, 'engine': result_engine, + 'x-sveltekit-invalidated': "001", # hardcoded from all requests seen } - # pylint: disable=undefined-variable - if use_cache: - params['data']['oc'] = "on" - # pylint: enable=undefined-variable - + if isinstance(country, str): + args['country'] = country if params['time_range'] in time_range_dict: - params['dateRestrict'] = time_range_dict[params['time_range']] - else: - params['dateRestrict'] = '' - + args['lastUpdated'] = time_range_dict[params['time_range']] if params['pageno'] > 1: - # Page 1 is n/a, Page 2 is 11, page 3 is 21, ... - params['data']['start'] = ''.join([str(params['pageno'] - 1), "1"]) + args['page'] = params['pageno'] - if params['headers'] is None: - params['headers'] = {} + params['url'] = f"{base_url}/search/__data.json?{urlencode(args)}" - 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 - the DOM resp: requests response object""" + json_response = resp.json() - dom = html.fromstring(resp.text) - if not is_vpn_connected(dom): - raise SearxEngineResponseException('Not connected to Mullvad VPN') - search_results = eval_xpath(dom.body, '//main/div[2]/div') - return list(extract_results(search_results)) + nodes = json_response["nodes"] + # 0: is None + # 1: has "connected=True", not useful + # 2: query results within "data" + 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): - """Fetch languages and regions from Mullvad-Leta + query_meta_data: DataNodeQueryMetaDataIndices = data_nodes[0] - .. warning:: + query_items_indices = query_meta_data['items'] - Fetching the engine traits also requires a Mullvad VPN connection. If - not connected, then an error message will print and no traits will be - updated. - """ - # pylint: disable=import-outside-toplevel - # see https://github.com/searxng/searxng/issues/762 - from searx.network import post as http_post + results = [] + for idx in data_nodes[query_items_indices]: + query_item_indices: DataNodeResultIndices = data_nodes[idx] + results.append( + { + 'url': data_nodes[query_item_indices['link']], + '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 - 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 + return results diff --git a/searx/settings.yml b/searx/settings.yml index ec72f8736..37bec7849 100644 --- a/searx/settings.yml +++ b/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