Wikidata Elastic Search - Backend Implementation

pyproject.toml

@@ -28,6 +28,7 @@ loguru = "^0.6.0"
 orjson = "^3.8.2"
 drepr = "^2.10.0"
 rsoup = "^2.5.1"
+nh3 = "^0.2.13"
 
 lat_lon_parser = "^1.3.0"
 

sand/app.py

@@ -9,6 +9,7 @@
 from sand.controllers.assistant import assistant_bp
 from sand.controllers.project import project_bp
 from sand.controllers.table import table_bp, table_row_bp
+from sand.controllers.search import search_bp
 from sand.controllers.settings import setting_bp
 from sand.deserializer import deserialize_graph
 from sand.models import EntityAR, SemanticModel
@@ -27,6 +28,7 @@
         assistant_bp,
         table_row_bp,
         setting_bp,
+        search_bp,
         generate_api(
             SemanticModel,
             deserializers={"data": deserialize_graph},

sand/config.py

@@ -31,6 +31,7 @@
     "ont_classes": {
         "constructor": "sand.extensions.wikidata.get_ontclass_db",
         "uri2id": "sand.extensions.wikidata.uri2id",
+        "id2uri": "sm.namespaces.prelude.WikidataNamespace.get_entity_abs_uri",
         "args": {
             "dbfile": "/tmp/wdclasses.db",
             "proxy": True,
@@ -41,6 +42,7 @@
     "ont_props": {
         "constructor": "sand.extensions.wikidata.get_ontprop_db",
         "uri2id": "sand.extensions.wikidata.uri2id",
+        "id2uri": "sm.namespaces.prelude.WikidataNamespace.get_prop_abs_uri",
         "args": {
             "dbfile": "/tmp/wdprops.db",
             "proxy": True,
@@ -62,6 +64,11 @@
         "mtab": "sand.extensions.assistants.mtab.MTabAssistant",
         # "default": "mtab",
     },
+    "search": {
+        "entities": "sand.extensions.search.wikidata_search.WikidataSearch",
+        "classes": "sand.extensions.search.wikidata_search.WikidataSearch",
+        "props": "sand.extensions.search.wikidata_search.WikidataSearch"
+    },
     "exports": {
         "drepr": "sand.extensions.export.drepr.main.DreprExport",
         "default": "sand.extensions.export.drepr.main.DreprExport"

sand/controllers/search.py

@@ -0,0 +1,55 @@
+import threading
+from typing import Dict, List
+from flask.blueprints import Blueprint
+from sm.misc.funcs import import_func
+from sand.config import SETTINGS
+from flask import request, jsonify
+
+from sand.extension_interface.search import ISearch
+
+search_bp = Blueprint("search", "search")
+
+GetSearchCache = threading.local()
+
+
+def get_search(name) -> ISearch:
+    """
+    Returns an implementation of an ISearch Interface from the
+    configuration file.
+    """
+    global GetSearchCache
+
+    if not hasattr(GetSearchCache, "search"):
+        GetSearchCache.search = {}
+        search_config = SETTINGS["search"]
+        constructor = search_config[name]
+        GetSearchCache.search[name] = import_func(constructor)()
+
+    return GetSearchCache.search[name]
+
+
+@search_bp.route(f"/{search_bp.name}/classes", methods=["GET"])
+def search_classes():
+    """API Route to search for classes with their names"""
+    search_text = request.args.get('q')
+    wikidata_search = get_search('classes')
+    payload = wikidata_search.find_class_by_name(search_text)
+    return jsonify(payload)
+
+
+@search_bp.route(f"/{search_bp.name}/entities", methods=["GET"])
+def search_entities():
+    """API Route to search for entities with their names"""
+    search_text = request.args.get('q')
+    wikidata_search = get_search('entities')
+    payload = wikidata_search.find_entity_by_name(search_text)
+    return jsonify(payload)
+
+
+@search_bp.route(f"/{search_bp.name}/props", methods=["GET"])
+def search_props():
+    """API Route to search for properties with their names"""
+    search_text = request.args.get('q')
+    wikidata_search = get_search('props')
+    payload = wikidata_search.find_props_by_name(search_text)
+    return jsonify(payload)

sand/extension_interface/search.py

@@ -1,21 +1,23 @@
 from abc import ABC, abstractmethod
+from typing import Dict
 
 
 class ISearch(ABC):
     """ Search Interface to support searches from multiple
         KG datastores.
     """
+
     @abstractmethod
-    def find_class_by_name(self):
+    def find_class_by_name(self, search_text: str) -> Dict:
         """Search Class using name"""
         pass
 
     @abstractmethod
-    def find_entity_by_name(self):
+    def find_entity_by_name(self, search_text: str) -> Dict:
         """Search Entity using name"""
         pass
 
     @abstractmethod
-    def find_props_by_name(self):
+    def find_props_by_name(self, search_text: str) -> Dict:
         """Search properties using name"""
         pass

sand/extensions/search/wikidata_search.py

@@ -0,0 +1,105 @@
+import requests
+from flask import request, jsonify
+from typing import Dict
+import nh3
+from sand.controllers.search import ISearch
+from sand.models.entity import Entity
+from sand.models.ontology import OntClass, OntProperty
+
+
+class WikidataSearch(ISearch):
+
+    def __init__(self):
+        self.wikidata_url = "https://www.wikidata.org/w/api.php"
+        self.local_class_idsearch_uri = "http://0.0.0.0:5525/api/classes/"
+        self.PARAMS = {
+            "action": "query",
+            "format": "json",
+            "list": "search",
+            "srsearch": "",
+            "utf8": "",
+            "srnamespace": 0,
+            "srlimit": 10,
+            "srprop": "snippet|titlesnippet"
+        }
+        self.search_item_template = {
+            "label": "",
+            "id": "",
+            "description": "",
+            "uri": "",
+        }
+
+    def get_class_search_params(self, search_text: str) -> Dict:
+        """Updates class search parameters for wikidata API"""
+        class_params = self.PARAMS.copy()
+        class_params["srnamespace"] = 0
+        class_params['srsearch'] = f"haswbstatement:P279 {search_text}"
+        return class_params
+
+    def get_local_class_properties(self, id: str) -> Dict:
+        """Calls local class search API to fetch all class metadata using class ID"""
+        api_data = requests.get(self.local_class_idsearch_uri + str(id))
+        return api_data.json()
+
+    def get_entity_search_params(self, search_text: str) -> Dict:
+        """Updates entity search parameters for wikidata API"""
+        entity_params = self.PARAMS.copy()
+        entity_params["srnamespace"] = 0
+        entity_params['srsearch'] = search_text
+        return entity_params
+
+    def get_props_search_params(self, search_text: str) -> Dict:
+        """Updates property search parameters for wikidata API"""
+        props_params = self.PARAMS.copy()
+        props_params["srnamespace"] = 120
+        props_params['srsearch'] = search_text
+        return props_params
+
+    def find_class_by_name(self, search_text: str) -> Dict:
+        """
+        Uses Wikidata API to search for classes using their name/text.
+        Uses local ID based class search to fetch label and description data.
+        """
+        request_params = self.get_class_search_params(search_text)
+        api_data = requests.get(self.wikidata_url, request_params)
+        search_items = api_data.json()['query']['search']
+        payload = {"items": []}
+        for search_item in search_items:
+            item = self.search_item_template.copy()
+            item['id'] = search_item['title']
+            local_class_props = self.get_local_class_properties(item['id'])
+            item['label'] = local_class_props['label']
+            item['description'] = local_class_props['description']
+            item['uri'] = OntClass.id2uri(item['id'])
+            payload['items'].append(item)
+        return payload
+
+    def find_entity_by_name(self, search_text: str) -> Dict:
+        """Uses Wikidata API to search for entities using their name/text."""
+        request_params = self.get_entity_search_params(search_text)
+        api_data = requests.get(self.wikidata_url, request_params)
+        search_items = api_data.json()['query']['search']
+        payload = {"items": []}
+        for search_item in search_items:
+            item = self.search_item_template.copy()
+            item['label'] = nh3.clean(search_item['titlesnippet'], tags=set())
+            item['id'] = search_item['title']
+            item['description'] = nh3.clean(search_item['snippet'], tags=set())
+            item['uri'] = Entity.id2uri(item['id'])
+            payload['items'].append(item)
+        return payload
+
+    def find_props_by_name(self, search_text: str) -> Dict:
+        """Uses Wikidata API to search for properties using their name/text."""
+        request_params = self.get_props_search_params(search_text)
+        api_data = requests.get(self.wikidata_url, request_params)
+        search_items = api_data.json()['query']['search']
+        payload = {"items": []}
+        for search_item in search_items:
+            item = self.search_item_template.copy()
+            item['label'] = nh3.clean(search_item['titlesnippet'], tags=set())
+            item['id'] = search_item['title'].split(":")[1]
+            item['description'] = nh3.clean(search_item['snippet'], tags=set())
+            item['uri'] = OntProperty.id2uri(item['id'])
+            payload['items'].append(item)
+        return payload

sand/models/ontology.py

@@ -24,9 +24,15 @@ def readable_label(self):
     def uri2id(uri: str) -> str:
         """Convert class URI to entity ID."""
         raise NotImplementedError(
-            "The method is set when its store is initialized. Check the call order to ensure `OntClassAR` is called first"
+            "The method is set when its store is initialized. Check the call order to ensure `OntClassAR`is called first"
         )
 
+    @staticmethod
+    def id2uri(id: str) -> str:
+        """Convert class ID to class URI."""
+        raise NotImplementedError(
+            "The method is set when its store is initialized. Check the call order to ensure `OntClassAR` is called first"
+        )
 
 OntPropertyDataType = Literal[
     "monolingualtext",
@@ -62,6 +68,12 @@ def uri2id(uri: str) -> str:
             "The method is set when its store is initialized. Check the call order to ensure `OntPropertyAR` is called first"
         )
 
+    @staticmethod
+    def id2uri(id: str) -> str:
+        """Convert property ID to property URI."""
+        raise NotImplementedError(
+            "The method is set when its store is initialized. Check the call order to ensure `OntPropertyAR` is called first"
+        )
 
 PROP_AR = None
 CLASS_AR = None
@@ -89,6 +101,7 @@ def OntPropertyAR() -> Mapping[str, OntProperty]:
         func = import_func(cfg["constructor"])
         PROP_AR = func(**cfg["args"])
         OntProperty.uri2id = import_func(cfg["uri2id"])
+        OntProperty.id2uri = import_func(cfg["id2uri"])
 
     return PROP_AR
 
@@ -102,5 +115,5 @@ def OntClassAR() -> Mapping[str, OntClass]:
         func = import_func(cfg["constructor"])
         CLASS_AR = func(**cfg["args"])
         OntClass.uri2id = import_func(cfg["uri2id"])
-
+        OntClass.id2uri = import_func(cfg["id2uri"])
     return CLASS_AR