"""
:Author: Daniel Mohr
:Email: daniel.mohr@dlr.de
:Date: 2021-03-23 (last change).
:License: GNU GENERAL PUBLIC LICENSE, Version 3, 29 June 2007.
"""
from .add_properties_to_object import add_properties_to_object
from .get_graph_item import get_graph_item
def create_context_schema(word):
"""
:Author: Daniel Mohr
:Date: 2021-03-16
"""
context = dict()
context["required"] = [word]
context["properties"] = {word: {"type": "string",
"enum": ["https://schema.org/" + word]}}
return context
def add_word(schemaorg_data, word):
"""
:Author: Daniel Mohr
:Date: 2021-03-23
This function generates a json schema from https://schema.org , which
desribes the given word.
:param schemaorg_data: json-ld data from https://schema.org as returned
from :func:`get_schema_org_data`.
:param word: word, which are a Schema.org Type (Schema.org vocabulary)
:return: return a list of:
* missing words
* json schema describing a json-ld for the given word
"""
# pylint: disable=too-many-branches
missing_words = []
schema = dict()
data = get_graph_item(schemaorg_data, word)
if data is None:
return [], None
47 ↛ 91line 47 didn't jump to line 91, because the condition on line 47 was never false if ("@type" in data) and (data["@type"] == "rdfs:Class"):
# found json object description
schema["definitions"] = dict()
schema["definitions"][word] = dict()
schema["definitions"][word]["type"] = "object"
52 ↛ 55line 52 didn't jump to line 55, because the condition on line 52 was never false if "rdfs:comment" in data:
schema["definitions"][word]["description"] = data["rdfs:comment"]
# schema["definitions"][word]["required"] = ["@context"]
schema["definitions"][word]["required"] = []
schema["definitions"][word]["properties"] = dict()
properties = schema["definitions"][word]["properties"]
# properties["@context"] = create_context_schema(word)
if "rdfs:subClassOf" in data:
sub_class_of = []
if isinstance(data["rdfs:subClassOf"], dict):
62 ↛ 71line 62 didn't jump to line 71, because the condition on line 62 was never false if data["rdfs:subClassOf"]["@id"].startswith('schema:'):
sub_class_of.append(data["rdfs:subClassOf"]["@id"])
64 ↛ 69line 64 didn't jump to line 69, because the condition on line 64 was never false elif isinstance(data["rdfs:subClassOf"], list):
for sco in data["rdfs:subClassOf"]:
66 ↛ 65line 66 didn't jump to line 65, because the condition on line 66 was never false if sco["@id"].startswith('schema:'):
sub_class_of.append(sco["@id"])
else:
raise NotImplementedError(
'data type of "rdfs:subClassOf" not handled')
71 ↛ 82line 71 didn't jump to line 82, because the condition on line 71 was never false if bool(sub_class_of): # len(sub_class_of) > 0
schema["definitions"][word]["allOf"] = list()
schema["definitions"][word]["allOf"].append(properties)
properties = schema["definitions"][word]["allOf"][0]
for sco in sub_class_of:
new_word = sco.split('schema:')[1]
schema["definitions"][word]["allOf"].append(
{"$ref": "#/definitions/" + new_word})
missing_words.append(new_word)
del schema["definitions"][word]["properties"]
# add other properties to: properties
add_properties_to_object(
schemaorg_data, word, properties, missing_words)
# items with: "schema:domainIncludes": {"@id": "schema:Person"}
# example: "@id": "schema:additionalName",
schema["required"] = ["@context"]
schema["properties"] = dict()
schema["properties"]["@context"] = create_context_schema(word)
89 ↛ 91line 89 didn't jump to line 91, because the condition on line 89 was never false if not bool(schema["definitions"][word]["required"]): # len(..) == 0
del schema["definitions"][word]["required"]
return missing_words, schema
|