File size: 6,535 Bytes
06555b5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 |
import re
import json
import math
import typing
import decimal
import pathlib
import datetime
import collections
import attr
from rdflib import Graph, URIRef, Literal
from rfc3986 import URIReference
from isodate.duration import Duration
from .utils import is_url
__all__ = ['group_triples', 'to_json', 'Triple', 'format_value']
def format_value(value, col):
"""
Format values as JSON-LD literals.
"""
if isinstance(value, (datetime.date, datetime.datetime, datetime.time)):
res = value.isoformat()
if col and col.datatype.base == 'time':
res = res.split('T')[-1]
if col and col.datatype.base == 'date':
res = re.sub('T[0-9.:]+', '', res)
if isinstance(value, (datetime.datetime, datetime.time)):
stamp, _, milliseconds = res.partition('.')
return '{}.{}'.format(stamp, milliseconds.rstrip('0')) if milliseconds \
else stamp.replace('+00:00', 'Z')
return res # pragma: no cover
if isinstance(value, datetime.timedelta):
return col.datatype.formatted(value)
if isinstance(value, Duration):
return col.datatype.formatted(value)
if isinstance(value, decimal.Decimal):
value = float(value)
if isinstance(value, URIReference):
return value.unsplit()
if isinstance(value, bytes):
return col.datatype.formatted(value)
if isinstance(value, pathlib.Path):
return str(value)
if isinstance(value, float):
return 'NaN' if math.isnan(value) else (
'{}INF'.format('-' if value < 0 else '') if math.isinf(value) else value)
return value
@attr.s
class Triple:
"""
A table cell's data as RDF triple.
"""
about = attr.ib()
property = attr.ib()
value = attr.ib()
def as_rdflib_triple(self):
return (
URIRef(self.about),
URIRef(self.property),
URIRef(self.value) if is_url(self.value) else Literal(self.value))
@classmethod
def from_col(cls, table, col, row, prop, val, rownum):
"""
"""
_name = col.header if col else None
propertyUrl = col.propertyUrl if col else table.inherit('propertyUrl')
if propertyUrl:
prop = table.expand(propertyUrl, row, _row=rownum, _name=_name, qname=True)
is_type = prop == 'rdf:type'
valueUrl = col.valueUrl if col else table.inherit('valueUrl')
if valueUrl:
val = table.expand(
valueUrl, row, _row=rownum, _name=_name, qname=is_type, uri=not is_type)
val = format_value(val, col)
s = None
aboutUrl = col.aboutUrl if col else None
if aboutUrl:
s = table.expand(aboutUrl, row, _row=rownum, _name=_name) or s
return cls(about=s, property=prop, value=val)
def frame(data: list) -> list:
"""
Inline referenced items to force a deterministic graph layout.
.. see:: https://w3c.github.io/json-ld-framing/#introduction
"""
items, refs = collections.OrderedDict(), {}
for item in data:
itemid = item.get('@id')
if itemid:
items[itemid] = item
for vs in item.values():
for v in [vs] if not isinstance(vs, list) else vs:
if isinstance(v, dict):
refid = v.get('@id')
if refid:
refs.setdefault(refid, (v, []))[1].append(item)
for ref, subjects in refs.values():
if len(subjects) == 1 and ref['@id'] in items:
ref.update(items.pop(ref['@id']))
return list(items.values())
def to_json(obj, flatten_list=False):
"""
Simplify JSON-LD data by refactoring trivial objects.
"""
if isinstance(obj, dict):
if '@value' in obj:
obj = obj['@value']
if len(obj) == 1 and '@id' in obj:
obj = obj['@id']
if isinstance(obj, dict):
return {
'@type' if k == 'rdf:type' else k: to_json(v, flatten_list=flatten_list)
for k, v in obj.items()}
if isinstance(obj, list):
if len(obj) == 1 and flatten_list:
return to_json(obj[0], flatten_list=flatten_list)
return [to_json(v, flatten_list=flatten_list) for v in obj]
return obj
def group_triples(triples: typing.Iterable[Triple]) -> typing.List[dict]:
"""
Group and frame triples into a `list` of JSON objects.
"""
merged = []
for triple in triples:
if isinstance(triple.value, list):
for t in merged:
if t.property == triple.property and isinstance(t.value, list):
t.value.extend(triple.value)
break
else:
merged.append(triple)
else:
merged.append(triple)
grouped = collections.OrderedDict()
triples = []
# First pass: get top-level properties.
for triple in merged:
if triple.about is None and triple.property == '@id':
grouped[triple.property] = triple.value
else:
if not triple.about:
# For test48
if triple.property in grouped:
if not isinstance(grouped[triple.property], list):
grouped[triple.property] = [grouped[triple.property]]
grouped[triple.property].append(triple.value)
else:
grouped[triple.property] = triple.value
else:
triples.append(triple)
if not triples:
return [grouped]
g = Graph()
for triple in triples:
g.add(triple.as_rdflib_triple())
if '@id' in grouped:
for prop, val in grouped.items():
if prop != '@id':
g.add(Triple(about=grouped['@id'], property=prop, value=val).as_rdflib_triple())
res = g.serialize(format='json-ld')
# Frame and simplify the resulting objects, augment with list index:
res = [(i, to_json(v, flatten_list=True)) for i, v in enumerate(frame(json.loads(res)))]
# Sort the objects making sure the one with the row's aboutUrl as @id comes first:
res = [k[1] for k in sorted(
res, key=lambda o: -1 if o[1].get('@id') == grouped.get('@id') else o[0])]
# If there's no aboutUrl for the row and we have only one object from triples, we just merge
# the properties into a single object.
if grouped and ('@id' not in grouped) and len(res) == 1:
grouped.update(res[0])
return [grouped]
return res
|