Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Renderizza i campi di un json-ld
import yaml
from os.path import basename
from pyld import jsonld
import logging
log = logging.getLogger()
logging.basicConfig(level=logging.INFO)
url = "https://ontopia-lodview.prod.pdnd.italia.it/onto/CPV/Person"
nulla = lambda *a, **kw: None
def jsonld_expand(url):
url = url.replace("w3id.org/italia", "ontopia-lodview.pdnd.italia.it")
return jsonld.expand(url)
def flower(property_name):
if not property_name:
return property_name
s = property_name[0].lower()
if len(property_name) >= 2:
s += property_name[1:]
return s
def test_parse_field():
item = {
"http://www.w3.org/2002/07/owl#onProperty": [
{"@id": "https://w3id.org/italia/onto/CPV/hasBirthPlace"}
]
}
s = parse_property(item)
def test_parse_field_1():
item = {
"@id": "_:b10",
"@type": ["http://www.w3.org/2002/07/owl#Restriction"],
"http://www.w3.org/2002/07/owl#onDataRange": [
{"@id": "http://www.w3.org/2001/XMLSchema#dateTime"}
],
"http://www.w3.org/2002/07/owl#onProperty": [
{"@id": "https://w3id.org/italia/onto/CPV/dateOfBirth"}
],
"http://www.w3.org/2002/07/owl#qualifiedCardinality": [
{
"@type": "http://www.w3.org/2001/XMLSchema#nonNegativeInteger",
"@value": "1",
}
],
}
s = parse_property(item)
assert s == "dateOfBirth"
t = parse_type(item)
assert t["type"] == "http://www.w3.org/2001/XMLSchema#dateTime"
def test_parse_field_2():
item = {
"@id": "_:b0",
"@type": ["http://www.w3.org/2002/07/owl#Restriction"],
"http://www.w3.org/2002/07/owl#allValuesFrom": [
{"@id": "https://w3id.org/italia/onto/CPV/Person"}
],
"http://www.w3.org/2002/07/owl#onProperty": [
{"@id": "https://w3id.org/italia/onto/CPV/hasRelationshipWith"}
],
}
s = parse_property(item)
assert s == "relationshipWith"
t = parse_type(item)
assert t["items"] == "https://w3id.org/italia/onto/CPV/Person"
def test_parse_field_3():
item = {
"@id": "https://w3id.org/italia/onto/CPV/taxCode",
"@type": ["http://www.w3.org/2002/07/owl#DatatypeProperty"],
"http://www.w3.org/2000/01/rdf-schema#comment": [
{"@language": "it", "@value": "Il codice fiscale di una persona."},
{"@language": "en", "@value": "The tax code of a person."},
],
"http://www.w3.org/2000/01/rdf-schema#domain": [
{"@id": "https://w3id.org/italia/onto/CPV/Person"}
],
"http://www.w3.org/2000/01/rdf-schema#isDefinedBy": [
{"@id": "https://w3id.org/italia/onto/CPV"}
],
"http://www.w3.org/2000/01/rdf-schema#label": [
{"@language": "en", "@value": "tax code"},
{"@language": "it", "@value": "codice fiscale"},
],
"http://www.w3.org/2000/01/rdf-schema#range": [
{"@id": "http://www.w3.org/2000/01/rdf-schema#Literal"}
],
"http://www.w3.org/2000/01/rdf-schema#subPropertyOf": [
{"@id": "https://w3id.org/italia/onto/l0/identifier"}
],
"http://www.w3.org/2002/07/owl#versionInfo": [
{"@language": "it", "@value": "stabile"},
{"@language": "en", "@value": "stable"},
],
}
t = parse_type(item)
assert t["type"] == "http://www.w3.org/2000/01/rdf-schema#Literal"
def _get_id(x, label):
ret = x.get(label)[0]["@id"]
if "italia" in ret:
components[basename(ret)] = ret
return ret
def parse_vc(item, **kwargs):
ret = {}
if "https://w3id.org/italia/onto/l0/controlledVocabulary" in item:
vc_url = _get_id(item, "https://w3id.org/italia/onto/l0/controlledVocabulary")
vc = jsonld_expand(vc_url)
labels = vc[0]["http://www.w3.org/2004/02/skos/core#hasTopConcept"]
for lang in "it en".split():
ret[f"x-extensible-enum-{lang}"] = [
parse_vc_label(item, lang, **kwargs) for item in labels
]
ret["vc"] = vc_url
return ret
def parse_vc_label(
item, language="it", backref="http://www.w3.org/2004/02/skos/core#prefLabel"
):
url = item["@id"]
schema = jsonld_expand(url)
if backref not in schema[0]:
backref = "http://www.w3.org/2004/02/skos/core#prefLabel"
ret = [
x["@value"]
for x in schema[0][backref]
if x.get("@language", language) == language
]
if len(ret):
return ret[0]
return []
def parse_type(item):
type_map = {
"http://www.w3.org/2002/07/owl#onDataRange": lambda x: {
"type": _get_id(x, "http://www.w3.org/2002/07/owl#onDataRange")
},
"http://www.w3.org/2002/07/owl#onClass": lambda x: {
"type": _get_id(x, "http://www.w3.org/2002/07/owl#onClass")
},
"http://www.w3.org/2002/07/owl#allValuesFrom": lambda x: {
"type": "array",
"items": _get_id(x, "http://www.w3.org/2002/07/owl#allValuesFrom"),
},
"http://www.w3.org/2000/01/rdf-schema#range": lambda x: {
"type": _get_id(x, "http://www.w3.org/2000/01/rdf-schema#range")
},
}
data_type = None
for k, f in type_map.items():
if k in item:
data_type = f(item)
if not data_type:
url = parse_property(item, strip=False)
nulla(f"Can't find data_type dereferencing {item} to {url}")
data_type = dereference(url=url)
kwargs = {"backref": item["@id"]} if "@id" in item else {}
enum = parse_vc(item, **kwargs) # , backref=item['@id'])
if enum:
data_type.update(enum)
# components[basename(data_type['type'])] = data_type['type']
return data_type
def parse_property(item, strip=True):
ON_PROPERTY = "http://www.w3.org/2002/07/owl#onProperty"
if ON_PROPERTY not in item:
return None
property = item[ON_PROPERTY][0]
assert "@id" in property
property_fqdn = property["@id"]
if not strip:
return property_fqdn
property_name = basename(property_fqdn)
if property_name.startswith("has"):
property_name = property_name[3:]
return flower(property_name)
components = {}
def dereference(schema=None, url=None, components=None):
log.info(f"url: {url}")
if url and not schema:
if not url.startswith("http"):
return url
if "XMLSchema" in url:
return url
schema = jsonld_expand(url)
if not schema:
return {}
from collections import defaultdict
nulla("dereferencing", yaml.dump(schema))
ret = defaultdict(list)
for x in schema:
try:
p = parse_property(x)
t = parse_type(x)
if p:
ret[p] = t
# log.info(yaml.dump({p: t}))
else:
if not ret:
return t
else:
pass # ret['resti'].append(t)
except Exception as e:
raise
return {"xxx": dict(ret)}
ret = dereference(schema=None, url=url)
log.info(yaml.dump(ret))
c = dict()
for k, v in dict(**components).items():
c[k] = dereference(url=v)
log.info(yaml.dump(c))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.