+# 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
+
+from typing import TYPE_CHECKING
+from hashlib import md5
+from urllib.parse import urlencode, unquote
+from json import loads
+
+from dateutil.parser import isoparse
+from babel.dates import format_datetime, format_date, format_time, get_datetime_format
+
+from searx.data import WIKIDATA_UNITS
+from searx.network import post, get
+from searx.utils import searx_useragent, get_string_replaces_function
+from searx.external_urls import get_external_url, get_earth_coordinates_url, area_to_osm_zoom
+from searx.engines.wikipedia import (
+ fetch_wikimedia_traits,
+ get_wiki_params,
+)
+from searx.enginelib.traits import EngineTraits
+
+if TYPE_CHECKING:
+ import logging
+
+ logger: logging.Logger
+
+traits: EngineTraits
+
+# 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',
+}
+
+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."""
+
+
+# SPARQL
+SPARQL_ENDPOINT_URL = 'https://query.wikidata.org/sparql'
+SPARQL_EXPLAIN_URL = 'https://query.wikidata.org/bigdata/namespace/wdq/sparql?explain'
+WIKIDATA_PROPERTIES = {
+ 'P434': 'MusicBrainz',
+ 'P435': 'MusicBrainz',
+ 'P436': 'MusicBrainz',
+ 'P966': 'MusicBrainz',
+ 'P345': 'IMDb',
+ 'P2397': 'YouTube',
+ 'P1651': 'YouTube',
+ 'P2002': 'Twitter',
+ 'P2013': 'Facebook',
+ 'P2003': 'Instagram',
+}
+
+# SERVICE wikibase:mwapi : https://www.mediawiki.org/wiki/Wikidata_Query_Service/User_Manual/MWAPI
+# SERVICE wikibase:label: https://en.wikibooks.org/wiki/SPARQL/SERVICE_-_Label#Manual_Label_SERVICE
+# https://en.wikibooks.org/wiki/SPARQL/WIKIDATA_Precision,_Units_and_Coordinates
+# https://www.mediawiki.org/wiki/Wikibase/Indexing/RDF_Dump_Format#Data_model
+# optimization:
+# * https://www.wikidata.org/wiki/Wikidata:SPARQL_query_service/query_optimization
+# * https://github.com/blazegraph/database/wiki/QueryHints
+QUERY_TEMPLATE = """
+SELECT ?item ?itemLabel ?itemDescription ?lat ?long %SELECT%
+WHERE
+{
+ SERVICE wikibase:mwapi {
+ bd:serviceParam wikibase:endpoint "www.wikidata.org";
+ wikibase:api "EntitySearch";
+ wikibase:limit 1;
+ mwapi:search "%QUERY%";
+ mwapi:language "%LANGUAGE%".
+ ?item wikibase:apiOutputItem mwapi:item.
+ }
+ hint:Prior hint:runFirst "true".
+
+ %WHERE%
+
+ SERVICE wikibase:label {
+ bd:serviceParam wikibase:language "%LANGUAGE%,en".
+ ?item rdfs:label ?itemLabel .
+ ?item schema:description ?itemDescription .
+ %WIKIBASE_LABELS%
+ }
+
+}
+GROUP BY ?item ?itemLabel ?itemDescription ?lat ?long %GROUP_BY%
+"""
+
+# Get the calendar names and the property names
+QUERY_PROPERTY_NAMES = """
+SELECT ?item ?name
+WHERE {
+ {
+ SELECT ?item
+ WHERE { ?item wdt:P279* wd:Q12132 }
+ } UNION {
+ VALUES ?item { %ATTRIBUTES% }
+ }
+ OPTIONAL { ?item rdfs:label ?name. }
+}
+"""
+
+# 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 get_headers():
+ # user agent: https://www.mediawiki.org/wiki/Wikidata_Query_Service/User_Manual#Query_limits
+ return {'Accept': 'application/sparql-results+json', 'User-Agent': searx_useragent()}
+
+
+def get_label_for_entity(entity_id, language):
+ name = WIKIDATA_PROPERTIES.get(entity_id)
+ if name is None:
+ name = WIKIDATA_PROPERTIES.get((entity_id, language))
+ if name is None:
+ name = WIKIDATA_PROPERTIES.get((entity_id, language.split('-')[0]))
+ if name is None:
+ name = WIKIDATA_PROPERTIES.get((entity_id, 'en'))
+ if name is None:
+ name = entity_id
+ return name
+
+
+def send_wikidata_query(query, method='GET'):
+ if method == 'GET':
+ # query will be cached by wikidata
+ http_response = get(SPARQL_ENDPOINT_URL + '?' + urlencode({'query': query}), headers=get_headers())
+ else:
+ # query won't be cached by wikidata
+ http_response = post(SPARQL_ENDPOINT_URL, data={'query': query}, headers=get_headers())
+ if http_response.status_code != 200:
+ logger.debug('SPARQL endpoint error %s', http_response.content.decode())
+ logger.debug('request time %s', str(http_response.elapsed))
+ http_response.raise_for_status()
+ return loads(http_response.content.decode())
+
+
+def request(query, params):
+
+ eng_tag, _wiki_netloc = get_wiki_params(params['searxng_locale'], traits)
+ query, attributes = get_query(query, eng_tag)
+ 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_headers()
+ params['language'] = eng_tag
+ params['attributes'] = attributes
+
+ return params
+
+
+def response(resp):
+
+ results = []
+ jsonresponse = loads(resp.content.decode())
+
+ language = resp.search_params['language']
+ attributes = resp.search_params['attributes']
+ logger.debug("request --> language %s // len(attributes): %s", language, len(attributes))
+
+ seen_entities = set()
+ for result in jsonresponse.get('results', {}).get('bindings', []):
+ attribute_result = {key: value['value'] for key, value in result.items()}
+ entity_url = 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/"
+
+
+
+
[docs]
+
def get_thumbnail(img_src):
+
"""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
+
[docs]
+
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'] = []
+