feat: Wikidata knowledge panel

app/i18n.py

@@ -44,9 +44,7 @@ def active_translation(lang=None):
     # restore context
     _current_language.reset(token)
 
-
-def get_current_translation():
-    """function to get translation object for current language"""
+def get_current_lang():
     lang = _current_language.get()
     if lang is None:
         # warn
@@ -56,7 +54,12 @@ def get_current_translation():
             stack_info=True,
         )
         lang = DEFAULT_LANGUAGE
-    return get_translation(lang)
+    return lang
+
+
+def get_current_translation():
+    """function to get translation object for current language"""
+    return get_translation(get_current_lang())
 
 
 def translate(message):

app/knowledge_panels.py

@@ -1,8 +1,7 @@
 from typing import Union
 from urllib.parse import urlencode
-
 from .models import HungerGameFilter, country_to_ISO_code, facet_plural
-from .off import data_quality, hungergame, last_edit
+from .off import data_quality, hungergame, last_edit, wikidata_helper
 
 
 def hunger_game_kp(
@@ -134,3 +133,36 @@ def last_edits_kp(
             ],
         },
     }
+
+
+def wikidata_kp(facet: str, value: str):
+    """
+    Return knowledge panel for wikidata
+    """
+    query = {}
+    if value:
+        query["tagtype"] = facet_plural(facet=facet)
+        query["fields"] = "wikidata"
+        query["tags"] = value
+
+    entities = wikidata_helper(query=query, value=value)
+    return {
+        "WikiData": {
+            "title": "wiki-data",
+            "subtitle": entities.description_tag,
+            "source_url": f"https://www.wikidata.org/wiki/{entities.entity_id}",
+            "elements": [
+                {
+                    "element_type": "text",
+                    "text_element": entities.label_tag,
+                    "image_url": entities.image_url,
+                },
+                {
+                    "element_type": "links",
+                    "wikipedia": entities.wikipedia_relation,
+                    "open_street_map": entities.OSM_relation,
+                    "INAO": entities.INAO_relation,
+                },
+            ],
+        },
+    }

app/main.py

@@ -1,11 +1,15 @@
 import logging
 from typing import Union
-
 from fastapi import FastAPI
-
+from .knowledge_panels import (
+    data_quality_kp,
+    hunger_game_kp,
+    last_edits_kp,
+    wikidata_kp,
+)
+from .models import FacetName, HungerGameFilter, Taxonomies
 from .i18n import active_translation
-from .knowledge_panels import data_quality_kp, hunger_game_kp, last_edits_kp
-from .models import FacetName, HungerGameFilter
+
 
 app = FastAPI()
 
@@ -26,7 +30,7 @@ def knowledge_panel(
 ):
     """
     FacetName is the model that have list of values
-    facet_value are the list of values connecting to FacetName eg:- category/beer, here beer is the value
+    facet_tag are the list of values connecting to FacetName eg:- category/beer, here beer is the value
     """
     with active_translation(lang_code):
         panels = []
@@ -48,5 +52,10 @@ def knowledge_panel(
             )
         except Exception:
             logging.exception("error occued while appending last-edits-kp")
+        try:
+            if facet_tag in Taxonomies.list():
+                panels.append(wikidata_kp(facet=facet_tag, value=value_tag))
+        except Exception:
+            logging.exception("error occurred while appending wikidata-kp")
 
         return {"knowledge_panels": panels}

app/models.py

@@ -6,23 +6,32 @@
 
 class FacetName(str, Enum):
     country = "country"
-    nutrition_grade = "nutrition-grade"
-    nova_group = "nova-group"
+    nutrition_grade = "nutrition_grade"
+    nova_group = "nova_group"
     brand = "brand"
     category = "category"
     label = "label"
     packaging = "packaging"
     origin_of_ingredient = "origin"
-    manufacturing_place = "manufacturing-place"
-    packager_code = "packager-code"
+    manufacturing_place = "manufacturing_place"
+    packager_code = "packager_code"
     ingredient = "ingredient"
     additive = "additive"
     vitamin = "vitamin"
     mineral = "mineral"
-    amino_acid = "amino-acid"
+    amino_acid = "amino_acid"
     nucleotide = "nucleotide"
     allergen = "allergen"
     trace = "trace"
+    language = "language"
+    contributor = "contributor"
+    state = "state"
+    data_source = "data_source"
+    entry_date = "entry_date"
+    last_edit_date = "last_edit_date"
+    last_check_date = "last_check_date"
+    other_nutritional_substances = "other_nutritional_substances"
+    team = "team"
 
     @staticmethod
     def list():
@@ -41,6 +50,30 @@ def list():
         return [c.value for c in HungerGameFilter]
 
 
+class Taxonomies(str, Enum):
+    country = "country"
+    nova_group = "nova_group"
+    brand = "brand"
+    category = "category"
+    label = "label"
+    packaging = "packaging"
+    ingredient = "ingredient"
+    additive = "additive"
+    vitamin = "vitamin"
+    mineral = "mineral"
+    amino_acid = "amino_acid"
+    nucleotide = "nucleotide"
+    allergen = "allergen"
+    state = "state"
+    origin_of_ingredient = "origin"
+    language = "language"
+    other_nutritional_substances = "other_nutritional_substances"
+
+    @staticmethod
+    def list():
+        return [c.value for c in Taxonomies]
+
+
 def country_to_ISO_code(value: str):
     """
     Helper function that return ISO code for country

app/off.py

@@ -1,8 +1,10 @@
+from collections import namedtuple
 from urllib.parse import urljoin
 
 import requests
 
-from .i18n import translate as _
+from .i18n import DEFAULT_LANGUAGE, get_current_lang, translate as _
+from .wikidata_utils import get_wikidata_entity, wikidata_props
 
 
 def data_quality(url, path):
@@ -83,6 +85,72 @@ def last_edit(url, query):
     return text, description, title
 
 
+Entities = namedtuple(
+    "Entities",
+    [
+        "label_tag",
+        "description_tag",
+        "image_url",
+        "entity_id",
+        "OSM_relation",
+        "INAO_relation",
+        "wikipedia_relation",
+    ],
+)
+
+
+def in_lang(data, lang, suffix=""):
+    """retrieve an entry where key is lang + suffix
+    with an eventual fallback to DEFAULT_LANGUAGE
+    """
+    try:
+        return data[lang + suffix]
+    except KeyError:
+        if lang == DEFAULT_LANGUAGE:
+            raise
+        return data[DEFAULT_LANGUAGE + suffix]
+
+
+def wikidata_helper(query, value):
+    """
+    Helper function to return wikidata eg:label,description,image_url
+    """
+    lang = get_current_lang()
+    url = "https://world.openfoodfacts.org/api/v2/taxonomy"
+    response_API = requests.get(url, params=query)
+    data = response_API.json()
+    tag = data[value]
+    entity_id = in_lang(tag["wikidata"], lang)
+    entity = get_wikidata_entity(entity_id=entity_id)
+    if wikidata_props.image_prop in entity:
+        image = entity[wikidata_props.image_prop]
+        image_url = image.image_url
+    else:
+        image_url = ""
+    wiki_links = in_lang(entity.attributes["sitelinks"], lang, "wiki")
+    wikipedia_relation = wiki_links.get("url")
+    if wikidata_props.INAO_prop in entity:
+        INAO = entity[wikidata_props.INAO_prop]
+        INAO_relation = "https://www.inao.gouv.fr/produit/{}".format(INAO)
+    else:
+        INAO_relation = ""
+    if wikidata_props.OSM_prop in entity:
+        osm = entity[wikidata_props.OSM_prop]
+        OSM_relation = "https://www.openstreetmap.org/relation/{}".format(osm)
+    else:
+        OSM_relation = ""
+    entities = Entities(
+        in_lang(entity.label, lang),
+        in_lang(entity.description, lang),
+        image_url,
+        entity_id,
+        OSM_relation,
+        INAO_relation,
+        wikipedia_relation,
+    )
+    return entities
+
+
 def hungergame():
     """Helper function for making Translation easy"""
     description = _("Answer robotoff questions about")

app/wikidata_utils.py

@@ -0,0 +1,31 @@
+from functools import cached_property
+
+import wikidata.client
+
+
+class WikiDataProperties:
+    """some useful properties"""
+
+    @cached_property
+    def _client(self):
+        return wikidata.client.Client()
+
+    @cached_property
+    def image_prop(self):
+        return self._client.get("P18")
+
+    @cached_property
+    def OSM_prop(self):
+        return self._client.get("P402")
+
+    @cached_property
+    def INAO_prop(self):
+        return self._client.get("P3895")
+
+
+wikidata_props = WikiDataProperties()
+
+
+def get_wikidata_entity(entity_id: str):
+    client = wikidata.client.Client()
+    return client.get(entity_id)