Coverage for src/time_agnostic_library/ocdm_converter.py: 100%
249 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-03 21:17 +0000
1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import gzip
6import re
7from collections import defaultdict
8from collections.abc import Callable, Generator
9from concurrent.futures import ThreadPoolExecutor
10from contextlib import contextmanager
11from datetime import datetime
12from pathlib import Path
14PROV_NS = "http://www.w3.org/ns/prov#"
15OCO_NS = "https://w3id.org/oc/ontology/"
16DCTERMS_NS = "http://purl.org/dc/terms/"
17XSD_NS = "http://www.w3.org/2001/XMLSchema#"
19_TRIPLE_LEN = 3
21# Regex to parse an N-Triples line into (subject, predicate, object) in a single
22# C-level pass. Falls back to the character-by-character parser on mismatch.
23#
24# N-Triples line format: <subject> <predicate> <object> .
25#
26# Group 1 - subject: URI <http://...> or blank node _:id
27# Group 2 - predicate: always a URI <http://...>
28# Group 3 - object, one of:
29# - URI: <http://...>
30# - literal: "text" optionally followed by @lang or ^^<datatype>
31# - blank node: _:id
32_NT_RE = re.compile(
33 r"(<[^>]+>|_:\S+)\s+" # group 1: subject (URI or blank node)
34 r"(<[^>]+>)\s+" # group 2: predicate (URI)
35 r"(<[^>]+>" # group 3 option a: URI object
36 r'|"(?:[^"\\]|\\.)*"' # group 3 option b: quoted literal (handles escapes)
37 r"(?:@[a-zA-Z-]+|\^\^<[^>]+>)?" # optional language tag or datatype
38 r"|_:\S+)" # group 3 option c: blank node object
39 r"\s*\.\s*$" # trailing dot and whitespace
40)
43def parse_ntriples_line(
44 line: str,
45 term_normalizer: Callable[[str], str] | None = None,
46) -> tuple[str, str, str] | None:
47 line = line.strip()
48 if not line or line.startswith("#"):
49 return None
50 m = _NT_RE.match(line)
51 if m:
52 subject, obj = m.group(1), m.group(3)
53 if term_normalizer:
54 subject = term_normalizer(subject)
55 obj = term_normalizer(obj)
56 return (subject, m.group(2), obj)
57 if line.endswith(" ."):
58 line = line[:-2]
59 elif line.endswith("."):
60 line = line[:-1]
61 line = line.strip()
62 parts = []
63 i = 0
64 while i < len(line) and len(parts) < _TRIPLE_LEN:
65 if line[i] == "<":
66 end = line.index(">", i)
67 parts.append(line[i : end + 1])
68 i = end + 1
69 elif line[i] == '"':
70 j = i + 1
71 while j < len(line):
72 if line[j] == "\\" and j + 1 < len(line):
73 j += 2
74 continue
75 if line[j] == '"':
76 break
77 j += 1
78 end_quote = j
79 rest_start = end_quote + 1
80 if rest_start < len(line) and line[rest_start : rest_start + 2] == "^^":
81 dt_start = rest_start + 2
82 if dt_start < len(line) and line[dt_start] == "<":
83 dt_end = line.index(">", dt_start)
84 parts.append(line[i : dt_end + 1])
85 i = dt_end + 1
86 else:
87 space = line.find(" ", dt_start)
88 if space == -1:
89 parts.append(line[i:])
90 i = len(line)
91 else:
92 parts.append(line[i:space])
93 i = space
94 elif rest_start < len(line) and line[rest_start] == "@":
95 space = line.find(" ", rest_start)
96 if space == -1:
97 parts.append(line[i:])
98 i = len(line)
99 else:
100 parts.append(line[i:space])
101 i = space
102 else:
103 parts.append(line[i : end_quote + 1])
104 i = end_quote + 1
105 elif line[i] == "_":
106 space = line.find(" ", i)
107 if space == -1:
108 parts.append(line[i:])
109 i = len(line)
110 else:
111 parts.append(line[i:space])
112 i = space
113 elif line[i] == " " or line[i] == "\t":
114 i += 1
115 else:
116 space = line.find(" ", i)
117 if space == -1:
118 parts.append(line[i:])
119 i = len(line)
120 else:
121 parts.append(line[i:space])
122 i = space
123 if len(parts) == _TRIPLE_LEN:
124 subject, obj = parts[0], parts[2]
125 if term_normalizer:
126 subject = term_normalizer(subject)
127 obj = term_normalizer(obj)
128 return (subject, parts[1], obj)
129 return None
132def extract_subject_uri(s_term: str) -> str:
133 if s_term.startswith("<") and s_term.endswith(">"):
134 return s_term[1:-1]
135 return s_term
138def _open_ntriples(filepath: Path):
139 if filepath.suffix == ".gz":
140 return gzip.open(filepath, "rt", encoding="utf-8", errors="replace")
141 return filepath.open(encoding="utf-8", errors="replace")
144def _open_for_writing(filepath: Path):
145 if filepath.suffix == ".gz":
146 return gzip.open(filepath, "wt", encoding="utf-8", compresslevel=6)
147 return filepath.open("w", encoding="utf-8")
150def read_ntriples_file(
151 filepath: Path,
152 term_normalizer: Callable[[str], str] | None = None,
153) -> list[tuple[str, str, str]]:
154 triples = []
155 with _open_ntriples(filepath) as f:
156 for line in f:
157 parsed = parse_ntriples_line(line, term_normalizer)
158 if parsed:
159 triples.append(parsed)
160 return triples
163def group_triples_by_subject(
164 triples: list[tuple[str, str, str]],
165) -> dict[str, set[tuple[str, str]]]:
166 by_subject: dict[str, set[tuple[str, str]]] = defaultdict(set)
167 for s, p, o in triples:
168 uri = extract_subject_uri(s)
169 by_subject[uri].add((p, o))
170 return by_subject
173def _read_and_group(
174 filepath: Path,
175 term_normalizer: Callable[[str], str] | None = None,
176) -> dict[str, set[tuple[str, str]]]:
177 by_subject: dict[str, set[tuple[str, str]]] = defaultdict(set)
178 match = _NT_RE.match
179 with _open_ntriples(filepath) as f:
180 for line in f:
181 m = match(line)
182 if m:
183 s, p, obj = m.groups()
184 if term_normalizer:
185 s = term_normalizer(s)
186 obj = term_normalizer(obj)
187 uri = s[1:-1] if s[0] == "<" else s
188 else:
189 parsed = parse_ntriples_line(line, term_normalizer)
190 if not parsed:
191 continue
192 s, p, obj = parsed
193 uri = s[1:-1] if s[0] == "<" and s[-1] == ">" else s
194 by_subject[uri].add((p, obj))
195 return by_subject
198def _format_timestamp(dt: datetime) -> str:
199 return dt.strftime("%Y-%m-%dT%H:%M:%S+00:00")
202def _build_update_query(
203 entity_uri: str,
204 data_graph_uri: str,
205 deleted_po: set[tuple[str, str]],
206 added_po: set[tuple[str, str]],
207) -> str:
208 parts = []
209 if deleted_po:
210 triples = " ".join(f"<{entity_uri}> {p} {o} ." for p, o in deleted_po)
211 parts.append(f"DELETE DATA {{ GRAPH <{data_graph_uri}> {{ {triples} }} }}")
212 if added_po:
213 triples = " ".join(f"<{entity_uri}> {p} {o} ." for p, o in added_po)
214 parts.append(f"INSERT DATA {{ GRAPH <{data_graph_uri}> {{ {triples} }} }}")
215 return "; ".join(parts)
218def _escape_sparql_for_nquads(query: str) -> str:
219 escaped = query.replace("\\", "\\\\")
220 escaped = escaped.replace('"', '\\"')
221 escaped = escaped.replace("\n", "\\n")
222 escaped = escaped.replace("\r", "\\r")
223 return escaped.replace("\t", "\\t")
226class _ProvenanceWriter:
227 """Serializes the provenance of every entity while the diffs are computed."""
229 def __init__(
230 self,
231 out,
232 data_graph_uri: str,
233 agent_uri: str,
234 timestamps: list[datetime],
235 ) -> None:
236 self._out = out
237 self._data_graph_uri = data_graph_uri
238 self._agent_uri = agent_uri
239 self._timestamps = timestamps
240 self._changes: dict[str, int] = {}
241 # Entities whose last change left them without triples, and the version
242 # it happened at. A reappearance removes the entry, so what is left at
243 # the end are the entities to mark as invalidated.
244 self._emptied_at: dict[str, int] = {}
246 def created(self, entity_uri: str) -> None:
247 """Record an entity the conversion has not seen before."""
248 if entity_uri in self._changes:
249 return
250 self._changes[entity_uri] = 0
251 prov_graph = f"<{entity_uri}/prov/>"
252 se1_uri = f"<{entity_uri}/prov/se/1>"
253 t0 = _format_timestamp(self._timestamps[0])
254 self._out.write(
255 f"{se1_uri} <{PROV_NS}specializationOf> <{entity_uri}> {prov_graph} .\n"
256 f'{se1_uri} <{PROV_NS}generatedAtTime> "{t0}"^^<{XSD_NS}dateTime> '
257 f"{prov_graph} .\n"
258 f"{se1_uri} <{PROV_NS}wasAttributedTo> <{self._agent_uri}> {prov_graph} .\n"
259 f'{se1_uri} <{DCTERMS_NS}description> "The entity has been created." '
260 f"{prov_graph} .\n"
261 )
263 def changed(
264 self,
265 entity_uri: str,
266 version_idx: int,
267 deleted_po: set[tuple[str, str]],
268 added_po: set[tuple[str, str]],
269 *,
270 emptied: bool,
271 ) -> None:
272 se_num = self._changes[entity_uri] + 2
273 self._changes[entity_uri] += 1
274 prov_graph = f"<{entity_uri}/prov/>"
275 se_uri = f"<{entity_uri}/prov/se/{se_num}>"
276 prev_se_uri = f"<{entity_uri}/prov/se/{se_num - 1}>"
277 timestamp = _format_timestamp(self._timestamps[version_idx])
278 update_query = _build_update_query(
279 entity_uri, self._data_graph_uri, deleted_po, added_po
280 )
281 escaped_query = _escape_sparql_for_nquads(update_query)
283 self._out.write(
284 f"{se_uri} <{PROV_NS}specializationOf> <{entity_uri}> {prov_graph} .\n"
285 f'{se_uri} <{PROV_NS}generatedAtTime> "{timestamp}"'
286 f"^^<{XSD_NS}dateTime> {prov_graph} .\n"
287 f"{se_uri} <{PROV_NS}wasAttributedTo> <{self._agent_uri}> {prov_graph} .\n"
288 f'{se_uri} <{OCO_NS}hasUpdateQuery> "{escaped_query}" {prov_graph} .\n'
289 f"{se_uri} <{DCTERMS_NS}description> "
290 f'"The entity has been modified." {prov_graph} .\n'
291 f"{se_uri} <{PROV_NS}wasDerivedFrom> {prev_se_uri} {prov_graph} .\n"
292 )
294 if emptied:
295 self._emptied_at[entity_uri] = version_idx
296 else:
297 self._emptied_at.pop(entity_uri, None)
299 def finish(self) -> None:
300 """Mark as invalidated the entities that never came back."""
301 for entity_uri, version_idx in self._emptied_at.items():
302 se_num = self._changes[entity_uri] + 1
303 self._out.write(
304 f"<{entity_uri}/prov/se/{se_num}> <{PROV_NS}invalidatedAtTime> "
305 f'"{_format_timestamp(self._timestamps[version_idx])}"'
306 f"^^<{XSD_NS}dateTime> <{entity_uri}/prov/> .\n"
307 )
310class OCDMConverter:
311 def __init__(
312 self,
313 data_graph_uri: str,
314 agent_uri: str,
315 term_normalizer: Callable[[str], str] | None = None,
316 ):
317 self.data_graph_uri = data_graph_uri
318 self.agent_uri = agent_uri
319 self.term_normalizer = term_normalizer
321 def convert_from_ic(
322 self,
323 ic_files: list[Path],
324 timestamps: list[datetime],
325 dataset_output: Path,
326 provenance_output: Path,
327 ) -> None:
328 prev_by_subject: dict[str, set[tuple[str, str]]] = {}
329 latest_by_subject: dict[str, set[tuple[str, str]]] = {}
331 # Prefetch pipeline: a single background thread reads and parses the
332 # next IC file while the main thread diffs the current pair. Two
333 # consecutive versions are all a conversion ever holds.
334 with (
335 self._open_provenance(provenance_output, timestamps) as writer,
336 ThreadPoolExecutor(max_workers=1) as executor,
337 ):
338 future = executor.submit(_read_and_group, ic_files[0], self.term_normalizer)
340 for version_idx in range(len(ic_files)):
341 cur_by_subject = future.result()
343 if version_idx + 1 < len(ic_files):
344 future = executor.submit(
345 _read_and_group,
346 ic_files[version_idx + 1],
347 self.term_normalizer,
348 )
350 for entity_uri in cur_by_subject:
351 writer.created(entity_uri)
353 if version_idx > 0:
354 for entity_uri in prev_by_subject.keys() | cur_by_subject.keys():
355 prev_po = prev_by_subject.get(entity_uri, set())
356 cur_po = cur_by_subject.get(entity_uri, set())
357 deleted_po = prev_po - cur_po
358 added_po = cur_po - prev_po
359 if deleted_po or added_po:
360 writer.changed(
361 entity_uri,
362 version_idx,
363 deleted_po,
364 added_po,
365 emptied=not cur_po,
366 )
368 prev_by_subject = cur_by_subject
369 if version_idx == len(ic_files) - 1:
370 latest_by_subject = cur_by_subject
372 self._write_dataset(dataset_output, latest_by_subject)
374 def convert_from_cb(
375 self,
376 initial_snapshot: Path,
377 changesets: list[tuple[Path, Path]],
378 timestamps: list[datetime],
379 dataset_output: Path,
380 provenance_output: Path,
381 ) -> None:
382 current_state: dict[str, set[tuple[str, str]]] = defaultdict(
383 set, _read_and_group(initial_snapshot, self.term_normalizer)
384 )
386 # Read the added and deleted files of each changeset in parallel.
387 with (
388 self._open_provenance(provenance_output, timestamps) as writer,
389 ThreadPoolExecutor(max_workers=2) as executor,
390 ):
391 for entity_uri in current_state:
392 writer.created(entity_uri)
394 for changeset_idx, (added_file, deleted_file) in enumerate(changesets):
395 version_idx = changeset_idx + 1
397 fut_del = executor.submit(
398 _read_and_group, deleted_file, self.term_normalizer
399 )
400 fut_add = executor.submit(
401 _read_and_group, added_file, self.term_normalizer
402 )
403 deleted_by_subject = fut_del.result()
404 added_by_subject = fut_add.result()
406 for entity_uri in deleted_by_subject.keys() | added_by_subject.keys():
407 writer.created(entity_uri)
408 deleted_po = deleted_by_subject.get(entity_uri, set())
409 added_po = added_by_subject.get(entity_uri, set())
411 current_state[entity_uri] -= deleted_po
412 current_state[entity_uri] |= added_po
414 emptied = not current_state[entity_uri]
415 if emptied:
416 del current_state[entity_uri]
418 if deleted_po or added_po:
419 writer.changed(
420 entity_uri,
421 version_idx,
422 deleted_po,
423 added_po,
424 emptied=emptied,
425 )
427 self._write_dataset(dataset_output, current_state)
429 @contextmanager
430 def _open_provenance(
431 self, provenance_output: Path, timestamps: list[datetime]
432 ) -> Generator[_ProvenanceWriter]:
433 provenance_output.parent.mkdir(parents=True, exist_ok=True)
434 with _open_for_writing(Path(provenance_output)) as out:
435 writer = _ProvenanceWriter(
436 out, self.data_graph_uri, self.agent_uri, timestamps
437 )
438 yield writer
439 writer.finish()
441 def _write_dataset(
442 self, dataset_output: Path, latest_by_subject: dict[str, set[tuple[str, str]]]
443 ) -> None:
444 dataset_output.parent.mkdir(parents=True, exist_ok=True)
445 with _open_for_writing(Path(dataset_output)) as out:
446 for entity_uri in sorted(latest_by_subject):
447 out.writelines(
448 f"<{entity_uri}> {p} {o} <{self.data_graph_uri}> .\n"
449 for p, o in sorted(latest_by_subject[entity_uri])
450 )