mullvad_leta.py

Go to the documentation of this file.

# SPDX-License-Identifier: AGPL-3.0-or-later
"""Mullvad Leta is a search engine proxy. Currently Leta only offers text
search results not image, news or any other types of search result.  Leta acts
as a proxy to Google and Brave search results. You can select which backend
search engine you wish to use, see  (:py:obj:`leta_engine`).
 
.. hint::
 
   Leta caches each search for up to 30 days.  For example, if you use search
   terms like ``news``, contrary to your intention you'll get very old results!
 
 
Configuration
=============
 
The engine has the following additional settings:
 
- :py:obj:`leta_engine` (:py:obj:`LetaEnginesType`)
 
You can configure one Leta engine for Google and one for Brave:
 
.. code:: yaml
 
  - name: mullvadleta
    engine: mullvad_leta
    leta_engine: google
    shortcut: ml
 
  - name: mullvadleta brave
    engine: mullvad_leta
    network: mullvadleta  # use network from engine "mullvadleta" configured above
    leta_engine: brave
    shortcut: mlb
 
Implementations
===============
 
"""
 
from __future__ import annotations
 
import typing
from urllib.parse import urlencode
import babel
from httpx import Response
from lxml import html
from searx.enginelib.traits import EngineTraits
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
about = {
    "website": search_url,
    "wikidata_id": 'Q47008412',  # the Mullvad id - not leta, but related
    "official_api_documentation": 'https://leta.mullvad.net/faq',
    "use_official_api": False,
    "require_api_key": False,
    "results": 'HTML',
}
 
# engine dependent config
categories = ["general", "web"]
paging = True
max_page = 10
time_range_support = True
time_range_dict = {
    "day": "d",
    "week": "w",
    "month": "m",
    "year": "y",
}
 
LetaEnginesType = typing.Literal["google", "brave"]
"""Engine types supported by mullvadleta."""
 
leta_engine: LetaEnginesType = "google"
"""Select Leta's engine type from :py:obj:`LetaEnginesType`."""
 
 
def init(_):
    l = typing.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):
    """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
 
 
class DataNodeResultIndices(typing.TypedDict):
    """Indices into query resultsdata."""
 
    link: int
    snippet: int
    title: int
    favicon: int
 
 
def request(query: str, params: dict):
    params["method"] = "GET"
    args = {
        "q": query,
        "engine": leta_engine,
        "x-sveltekit-invalidated": "001",  # hardcoded from all requests seen
    }
 
    country = traits.get_region(params.get("searxng_locale"), traits.all_locale)  # type: ignore
    if country:
        args["country"] = country
 
    language = traits.get_language(params.get("searxng_locale"), traits.all_locale)  # type: ignore
    if language:
        args["language"] = language
 
    if params["time_range"] in time_range_dict:
        args["lastUpdated"] = time_range_dict[params["time_range"]]
 
    if params["pageno"] > 1:
        args["page"] = params["pageno"]
 
    params["url"] = f"{search_url}/search/__data.json?{urlencode(args)}"
 
    return params
 
 
def response(resp: Response) -> EngineResults:
    json_response = resp.json()
 
    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
 
    query_meta_data: DataNodeQueryMetaDataIndices = data_nodes[0]
 
    query_items_indices = query_meta_data["items"]
 
    results = EngineResults()
    for idx in data_nodes[query_items_indices]:
        query_item_indices: DataNodeResultIndices = data_nodes[idx]
        results.add(
            MainResult(
                url=data_nodes[query_item_indices["link"]],
                title=data_nodes[query_item_indices["title"]],
                content=data_nodes[query_item_indices["snippet"]],
            )
        )
 
    return results
 
 
def fetch_traits(engine_traits: EngineTraits) -> None:
    """Fetch languages and regions from Mullvad-Leta"""
 
    def extract_table_data(table):
        for row in table.xpath(".//tr")[2:]:
            cells = row.xpath(".//td | .//th")  # includes headers and data
            if len(cells) > 1:  # ensure the column exists
                cell0 = cells[0].text_content().strip()
                cell1 = cells[1].text_content().strip()
                yield [cell0, cell1]
 
    # pylint: disable=import-outside-toplevel
    # see https://github.com/searxng/searxng/issues/762
    from searx.network import get as http_get
 
    # pylint: enable=import-outside-toplevel
 
    resp = http_get(f"{search_url}/documentation")
    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)
 
    # There are 4 HTML tables on the documentation page for extracting information:
    # 0. Keyboard Shortcuts
    # 1. Query Parameters (shoutout to Mullvad for accessible docs for integration)
    # 2. Country Codes [Country, Code]
    # 3. Language Codes [Language, Code]
    tables = eval_xpath_list(dom.body, "//table")
    if tables is None or len(tables) <= 0:
        print("ERROR: could not find any tables. Was the page updated?")
 
    language_table = tables[3]
    lang_map = {
        "zh-hant": "zh_Hans",
        "zh-hans": "zh_Hant",
        "jp": "ja",
    }
 
    for language, code in extract_table_data(language_table):
 
        locale_tag = lang_map.get(code, code).replace("-", "_")  # type: ignore
        try:
            locale = babel.Locale.parse(locale_tag)
        except babel.UnknownLocaleError:
            print(f"ERROR: Mullvad-Leta language {language} ({code}) is unknown by babel")
            continue
 
        sxng_tag = language_tag(locale)
        engine_traits.languages[sxng_tag] = code
 
    country_table = tables[2]
    country_map = {
        "cn": "zh-CN",
        "hk": "zh-HK",
        "jp": "ja-JP",
        "my": "ms-MY",
        "tw": "zh-TW",
        "uk": "en-GB",
        "us": "en-US",
    }
 
    for country, code in extract_table_data(country_table):
 
        sxng_tag = country_map.get(code)
        if sxng_tag:
            engine_traits.regions[sxng_tag] = code
            continue
 
        try:
            locale = babel.Locale.parse(f"{code.lower()}_{code.upper()}")
        except babel.UnknownLocaleError:
            locale = None
 
        if locale:
            engine_traits.regions[region_tag(locale)] = code
            continue
 
        official_locales = get_official_locales(code, engine_traits.languages.keys(), regional=True)
        if not official_locales:
            print(f"ERROR: Mullvad-Leta country '{code}' ({country}) could not be mapped as expected.")
            continue
 
        for locale in official_locales:
            engine_traits.regions[region_tag(locale)] = code