path: root/searx/engines/wikidata.py

# SPDX-License-Identifier: AGPL-3.0-or-later
"""This module implements the Wikidata engine.

Some implementations are shared from :ref:`wikipedia engine`.
"""
# pylint: disable=missing-class-docstring

import typing as t

from hashlib import md5
from urllib.parse import urlencode, unquote
from json import loads


from searx.network import post, get
from searx.utils import get_string_replaces_function
from searx.external_urls import area_to_osm_zoom
from searx.engines.wikipedia import (
    fetch_wikimedia_traits,
    get_wiki_params,
)
from searx.enginelib.traits import EngineTraits
from searx.wikidata_properties import (
    QUERY_TEMPLATE,
    WDArticle,
    WDAttrList,
    WDGeoAttribute,
    WDImageAttribute,
    WDURLAttribute,
    get_attributes,
)
from searx.wikidata import SPARQL_ENDPOINT_URL, SPARQL_EXPLAIN_URL, get_wikidata_headers

if t.TYPE_CHECKING:
    from searx.extended_types import SXNG_Response
    from searx.search.processors import OnlineParams


# about
about = {
    "website": 'https://wikidata.org/',
    "wikidata_id": 'Q2013',
    "official_api_documentation": 'https://query.wikidata.org/',
    "use_official_api": True,
    "require_api_key": False,
    "results": 'JSON',
}
language_support = True

display_type = ["infobox"]
"""A list of display types composed from ``infobox`` and ``list``.  The latter
one will add a hit to the result list.  The first one will show a hit in the
info box.  Both values can be set, or one of the two can be set."""

# see the property "dummy value" of https://www.wikidata.org/wiki/Q2013 (Wikidata)
# hard coded here to avoid to an additional SPARQL request when the server starts
DUMMY_ENTITY_URLS = set(
    "http://www.wikidata.org/entity/" + wid for wid in ("Q4115189", "Q13406268", "Q15397819", "Q17339402")
)


# https://www.w3.org/TR/sparql11-query/#rSTRING_LITERAL1
# https://lists.w3.org/Archives/Public/public-rdf-dawg/2011OctDec/0175.html
sparql_string_escape = get_string_replaces_function(
    # fmt: off
    {"\t": "\\\t", "\n": "\\\n", "\r": "\\\r", "\b": "\\\b", "\f": "\\\f", "\"": "\\\"", "'": "\\'", "\\": "\\\\"}
    # fmt: on
)

replace_http_by_https = get_string_replaces_function({"http:": "https:"})


def request(query: str, params: "OnlineParams") -> None:

    attributes: WDAttrList
    eng_tag, _wiki_netloc = get_wiki_params(params["searxng_locale"], traits)
    query, attributes = get_query(query, eng_tag or "en")
    logger.debug("request --> language %s // len(attributes): %s", eng_tag, len(attributes))

    params["method"] = "POST"
    params["url"] = SPARQL_ENDPOINT_URL
    params["data"] = {"query": query}
    params["headers"] = get_wikidata_headers()

    # additional parameters (not a part of OnlineParams)
    params["language"] = eng_tag  # type: ignore
    params["attributes"] = attributes  # type: ignore


def response(resp: "SXNG_Response") -> list[dict[str, t.Any]]:

    results: list[dict[str, t.Any]] = []
    jsonresponse = loads(resp.content.decode())

    # additional parameters ..
    language: str = resp.search_params["language"]  # type: ignore
    attributes: WDAttrList = resp.search_params["attributes"]  # type: ignore

    logger.debug("request --> language %s // len(attributes): %s", language, len(attributes))

    seen_entities: set[str] = set()
    for result in jsonresponse.get("results", {}).get("bindings", []):
        attribute_result = {key: value["value"] for key, value in result.items()}
        entity_url: str = attribute_result["item"]
        if entity_url not in seen_entities and entity_url not in DUMMY_ENTITY_URLS:
            seen_entities.add(entity_url)
            results += get_results(attribute_result, attributes, language)
        else:
            logger.debug("The SPARQL request returns duplicate entities: %s", str(attribute_result))

    return results


_IMG_SRC_DEFAULT_URL_PREFIX = "https://commons.wikimedia.org/wiki/Special:FilePath/"
_IMG_SRC_NEW_URL_PREFIX = "https://upload.wikimedia.org/wikipedia/commons/thumb/"


def get_thumbnail(img_src: str | None) -> str | None:
    """Get Thumbnail image from wikimedia commons

    Images from commons.wikimedia.org are (HTTP) redirected to
    upload.wikimedia.org.  The redirected URL can be calculated by this
    function.

    - https://stackoverflow.com/a/33691240

    """
    logger.debug("get_thumbnail(): %s", img_src)
    if not img_src is None and _IMG_SRC_DEFAULT_URL_PREFIX in img_src.split()[0]:
        img_src_name = unquote(img_src.replace(_IMG_SRC_DEFAULT_URL_PREFIX, "").split("?", 1)[0].replace("%20", "_"))
        img_src_name_first = img_src_name
        img_src_name_second = img_src_name

        if ".svg" in img_src_name.split()[0]:
            img_src_name_second = img_src_name + ".png"

        img_src_size = img_src.replace(_IMG_SRC_DEFAULT_URL_PREFIX, "").split("?", 1)[1]
        img_src_size = img_src_size[img_src_size.index("=") + 1 : img_src_size.index("&")]
        img_src_name_md5 = md5(img_src_name.encode("utf-8")).hexdigest()
        img_src = (
            _IMG_SRC_NEW_URL_PREFIX
            + img_src_name_md5[0]
            + "/"
            + img_src_name_md5[0:2]
            + "/"
            + img_src_name_first
            + "/"
            + img_src_size
            + "px-"
            + img_src_name_second
        )
        logger.debug("get_thumbnail() redirected: %s", img_src)

    return img_src


def get_results(
    attribute_result: dict[str, t.Any],
    attributes: WDAttrList,
    language: str,
):
    # pylint: disable=too-many-branches
    results: list[dict[str, t.Any]] = []
    infobox_title: str = attribute_result.get("itemLabel")  # pyright: ignore[reportAssignmentType]
    infobox_id = attribute_result["item"]
    infobox_id_lang: str | None = None
    infobox_urls: list[dict[str, str]] = []
    infobox_attributes: list[dict[str, str]] = []
    infobox_content = attribute_result.get("itemDescription", [])
    img_src: str | None = None
    img_src_priority = 0

    for attribute in attributes:
        value: str | None = attribute.get_str(attribute_result, language)
        if value is not None and value != "":
            if isinstance(attribute, (WDURLAttribute, WDArticle)):
                # get_select() method : there is group_concat(distinct ...;separator=", ")
                # split the value here
                for url in value.split(", "):
                    infobox_urls.append({"title": attribute.get_label(language), "url": url, **attribute.kwargs})
                    # "normal" results (not infobox) include official website and Wikipedia links.
                    if "list" in display_type and (
                        attribute.kwargs.get("official") or isinstance(attribute, WDArticle)
                    ):
                        results.append({"title": infobox_title, "url": url, "content": infobox_content})

                    # update the infobox_id with the wikipedia URL
                    # first the local wikipedia URL, and as fallback the english wikipedia URL
                    if isinstance(attribute, WDArticle) and (
                        (attribute.language == "en" and infobox_id_lang is None) or attribute.language != "en"
                    ):
                        infobox_id_lang = attribute.language
                        infobox_id = url
            elif isinstance(attribute, WDImageAttribute):
                # this attribute is an image.
                # replace the current image only the priority is lower
                # (the infobox contain only one image).
                if attribute.priority > img_src_priority:
                    img_src = get_thumbnail(value)
                    img_src_priority = attribute.priority
            elif isinstance(attribute, WDGeoAttribute):
                # geocoordinate link
                # use the area to get the OSM zoom
                # Note: ignore the unit (must be km² otherwise the calculation is wrong)
                # Should use normalized value p:P2046/psn:P2046/wikibase:quantityAmount
                area = attribute_result.get("P2046")
                osm_zoom: int = area_to_osm_zoom(area) if area else 19
                url = attribute.get_geo_url(attribute_result, osm_zoom=osm_zoom)
                if url:
                    infobox_urls.append({"title": attribute.get_label(language), "url": url, "entity": attribute.name})
            else:
                infobox_attributes.append(
                    {"label": attribute.get_label(language), "value": value, "entity": attribute.name}
                )

    if infobox_id:
        infobox_id = replace_http_by_https(infobox_id)

    # add the wikidata URL at the end
    infobox_urls.append({"title": "Wikidata", "url": attribute_result["item"]})

    if (
        "list" in display_type
        and img_src is None
        and len(infobox_attributes) == 0
        and len(infobox_urls) == 1
        and len(infobox_content) == 0
    ):
        results.append({"url": infobox_urls[0]["url"], "title": infobox_title, "content": infobox_content})
    elif "infobox" in display_type:
        results.append(
            {
                "infobox": infobox_title,
                "id": infobox_id,
                "content": infobox_content,
                "img_src": img_src,
                "urls": infobox_urls,
                "attributes": infobox_attributes,
            }
        )
    return results


def get_query(query: str, language: str) -> tuple[str, WDAttrList]:
    attributes = get_attributes(language)
    select = [a.get_select() for a in attributes]
    where = list(filter(lambda s: len(s) > 0, [a.get_where() for a in attributes]))
    wikibase_label = list(filter(lambda s: len(s) > 0, [a.get_wikibase_label() for a in attributes]))
    group_by = list(filter(lambda s: len(s) > 0, [a.get_group_by() for a in attributes]))
    query = (
        QUERY_TEMPLATE.replace("%QUERY%", sparql_string_escape(query))
        .replace("%SELECT%", " ".join(select))
        .replace("%WHERE%", "\n  ".join(where))
        .replace("%WIKIBASE_LABELS%", "\n      ".join(wikibase_label))
        .replace("%GROUP_BY%", " ".join(group_by))
        .replace("%LANGUAGE%", language)
    )
    return query, attributes


def debug_explain_wikidata_query(query: str, method: str = "GET"):
    if method == "GET":
        http_response = get(SPARQL_EXPLAIN_URL + "&" + urlencode({"query": query}), headers=get_wikidata_headers())
    else:
        http_response = post(SPARQL_EXPLAIN_URL, data={"query": query}, headers=get_wikidata_headers())
    http_response.raise_for_status()
    return http_response.content


def fetch_traits(engine_traits: EngineTraits):
    """Uses languages evaluated from :py:obj:`wikipedia.fetch_wikimedia_traits
    <searx.engines.wikipedia.fetch_wikimedia_traits>` and removes

    - ``traits.custom['wiki_netloc']``: wikidata does not have net-locations for
      the languages and the list of all

    - ``traits.custom['WIKIPEDIA_LANGUAGES']``: not used in the wikipedia engine

    """

    fetch_wikimedia_traits(engine_traits)
    engine_traits.custom["wiki_netloc"] = {}
    engine_traits.custom["WIKIPEDIA_LANGUAGES"] = []