Coverage for heritrace/routes/entity/_history.py: 88%
199 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +0000
1# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import re
6from datetime import datetime
8from flask import abort, render_template
9from flask_babel import gettext
10from flask_login import login_required
11from rdflib import RDF, Graph, Literal, URIRef
12from SPARQLWrapper import JSON
13from time_agnostic_library.agnostic_entity import AgnosticEntity
15from heritrace.extensions import (
16 get_change_tracking_config,
17 get_custom_filter,
18 get_dataset_is_quadstore,
19 get_display_rules_use_inverse_relations,
20 get_provenance_sparql,
21)
22from heritrace.routes.entity._blueprint import entity_bp
23from heritrace.routes.entity._rendering import generate_modification_text
24from heritrace.routes.entity._types import _DATETIME_MIN_UTC, HistoryContext
25from heritrace.sparql import get_sparql_bindings
26from heritrace.utils.converters import convert_to_datetime
27from heritrace.utils.display_rules_utils import (
28 get_grouped_triples,
29 get_highest_priority_class,
30)
31from heritrace.utils.shacl_utils import determine_shape_for_entity_triples
32from heritrace.utils.shacl_validation import get_valid_predicates
33from heritrace.utils.sparql_utils import (
34 convert_to_rdflib_graphs,
35 determine_shape_for_classes,
36 get_triples_from_graph,
37 parse_sparql_update,
38)
39from heritrace.utils.uri_utils import is_valid_url
42@entity_bp.route("/entity-history/<path:entity_uri>")
43@login_required
44def entity_history(entity_uri: str) -> str:
45 entity_uri_ref = URIRef(entity_uri)
46 custom_filter = get_custom_filter()
47 change_tracking_config = get_change_tracking_config()
49 agnostic_entity = AgnosticEntity(
50 res=entity_uri,
51 config=change_tracking_config,
52 include_related_objects=True,
53 include_merged_entities=True,
54 include_reverse_relations=get_display_rules_use_inverse_relations(),
55 )
56 history, provenance = agnostic_entity.get_history(include_prov_metadata=True)
57 history = convert_to_rdflib_graphs(history, is_quadstore=get_dataset_is_quadstore())
59 sorted_metadata = sorted(
60 provenance[entity_uri].items(),
61 key=lambda x: convert_to_datetime(x[1]["generatedAtTime"]) or _DATETIME_MIN_UTC,
62 )
63 sorted_timestamps: list[str] = [
64 dt.isoformat()
65 for _, meta in sorted_metadata
66 if (dt := convert_to_datetime(meta["generatedAtTime"])) is not None
67 ]
69 latest_metadata = sorted_metadata[-1][1] if sorted_metadata else None
70 is_latest_deletion = (
71 latest_metadata
72 and "invalidatedAtTime" in latest_metadata
73 and latest_metadata["invalidatedAtTime"]
74 )
75 if is_latest_deletion and len(sorted_timestamps) > 1:
76 context_snapshot = history[entity_uri][sorted_timestamps[-2]]
77 else:
78 context_snapshot = history[entity_uri][sorted_timestamps[-1]]
80 entity_classes = [
81 str(triple[2])
82 for triple in get_triples_from_graph(
83 context_snapshot, (entity_uri_ref, RDF.type, None)
84 )
85 ]
86 highest_priority_class = get_highest_priority_class(entity_classes)
88 snapshot_entity_shape = determine_shape_for_entity_triples(
89 list(get_triples_from_graph(context_snapshot, (entity_uri_ref, None, None)))
90 )
92 events = []
93 for i, (_snapshot_uri, metadata) in enumerate(sorted_metadata):
94 date = convert_to_datetime(metadata["generatedAtTime"])
95 if date is None:
96 msg = "date must not be None"
97 raise AssertionError(msg)
98 snapshot_graph = history[entity_uri][date.isoformat()]
100 responsible_agent = custom_filter.format_agent_reference(
101 metadata["wasAttributedTo"]
102 )
103 primary_source = custom_filter.format_source_reference(
104 metadata["hadPrimarySource"]
105 )
107 history_ctx = HistoryContext(
108 entity_uri=entity_uri,
109 highest_priority_class=highest_priority_class,
110 entity_shape=snapshot_entity_shape,
111 history=history,
112 sorted_timestamps=sorted_timestamps,
113 custom_filter=custom_filter,
114 )
116 description = _format_snapshot_description(
117 metadata,
118 history_ctx,
119 context_snapshot,
120 i,
121 )
122 modifications = metadata.get("hasUpdateQuery", "")
123 modification_text = ""
124 if modifications:
125 parsed_modifications = parse_sparql_update(modifications)
126 modification_text = generate_modification_text(
127 parsed_modifications,
128 history_ctx,
129 snapshot_graph,
130 date.isoformat(),
131 )
133 can_restore = len(sorted_metadata) > 1 and i + 1 < len(sorted_metadata)
134 restore_button = ""
135 if can_restore:
136 restore_label = gettext("Restore")
137 restore_action = (
138 f"/restore-version/{entity_uri}/{metadata['generatedAtTime']}"
139 )
140 restore_button = f"""
141 <form action='{restore_action}'
142 method='post'
143 class='d-inline restore-form'>
144 <button type='submit'
145 class='btn btn-success restore-btn'>
146 <i class='bi
147 bi-arrow-counterclockwise
148 me-1'></i>{restore_label}
149 </button>
150 </form>
151 """
153 event = {
154 "start_date": {
155 "year": date.year,
156 "month": date.month,
157 "day": date.day,
158 "hour": date.hour,
159 "minute": date.minute,
160 "second": date.second,
161 },
162 "text": {
163 "headline": gettext("Snapshot") + " " + str(i + 1),
164 "text": (
165 f"<p><strong>"
166 f"{gettext('Responsible agent')}"
167 f":</strong>"
168 f" {responsible_agent}</p>"
169 f"<p><strong>"
170 f"{gettext('Primary source')}"
171 f":</strong>"
172 f" {primary_source}</p>"
173 f"<p><strong>"
174 f"{gettext('Description')}"
175 f":</strong>"
176 f" {description}</p>"
177 f'<div class="modifications mb-3">'
178 f"{modification_text}"
179 f"</div>"
180 f'<div class="d-flex gap-2 mt-2">'
181 f"<a href='/entity-version/"
182 f"{entity_uri}/"
183 f"{metadata['generatedAtTime']}'"
184 f" class='btn btn-outline-primary"
185 f" view-version'"
186 f" target='_self'>"
187 f"{gettext('View version')}</a>"
188 f"{restore_button}"
189 f"</div>"
190 ),
191 },
192 "autolink": False,
193 }
195 if i + 1 < len(sorted_metadata):
196 next_date = convert_to_datetime(
197 sorted_metadata[i + 1][1]["generatedAtTime"]
198 )
199 if next_date is None:
200 msg = "next_date must not be None"
201 raise AssertionError(msg)
202 event["end_date"] = {
203 "year": next_date.year,
204 "month": next_date.month,
205 "day": next_date.day,
206 "hour": next_date.hour,
207 "minute": next_date.minute,
208 "second": next_date.second,
209 }
211 events.append(event)
213 entity_label = custom_filter.human_readable_entity(
214 entity_uri, (highest_priority_class, snapshot_entity_shape), context_snapshot
215 )
217 timeline_data = {
218 "entityUri": entity_uri,
219 "entityLabel": entity_label,
220 "entityClasses": list(entity_classes),
221 "entityShape": snapshot_entity_shape,
222 "events": events,
223 }
225 return render_template("entity/history.jinja", timeline_data=timeline_data)
228def _format_snapshot_description(
229 metadata: dict,
230 ctx: HistoryContext,
231 context_snapshot: Graph,
232 current_index: int,
233) -> str:
234 description = metadata.get("description", "")
235 is_merge_snapshot = False
236 was_derived_from = metadata.get("wasDerivedFrom")
237 if isinstance(was_derived_from, list) and len(was_derived_from) > 1:
238 is_merge_snapshot = True
240 if is_merge_snapshot:
241 match = re.search(
242 r"merged with ['\u2018\u2019]?([^'\u2018\u2019<>\s]+)"
243 r"['\u2018\u2019]?",
244 description,
245 )
246 if match:
247 potential_merged_uri = match.group(1)
248 if is_valid_url(potential_merged_uri):
249 merged_entity_uri_from_desc = potential_merged_uri
250 merged_entity_label = None
251 if current_index > 0:
252 previous_snapshot_timestamp = ctx.sorted_timestamps[
253 current_index - 1
254 ]
255 previous_snapshot_graph = ctx.history.get(ctx.entity_uri, {}).get(
256 previous_snapshot_timestamp
257 )
258 if previous_snapshot_graph:
259 raw_merged_entity_classes = [
260 str(o)
261 for s, p, o in get_triples_from_graph(
262 previous_snapshot_graph,
263 (URIRef(merged_entity_uri_from_desc), RDF.type, None),
264 )
265 ]
266 highest_priority_merged_class = (
267 get_highest_priority_class(raw_merged_entity_classes)
268 if raw_merged_entity_classes
269 else None
270 )
272 shape = determine_shape_for_classes(raw_merged_entity_classes)
273 merged_entity_label = ctx.custom_filter.human_readable_entity(
274 merged_entity_uri_from_desc,
275 (highest_priority_merged_class, shape),
276 previous_snapshot_graph,
277 )
278 if (
279 merged_entity_label
280 and merged_entity_label != merged_entity_uri_from_desc
281 ):
282 description = description.replace(
283 match.group(0), f"merged with '{merged_entity_label}'"
284 )
286 shape = (
287 determine_shape_for_classes([ctx.highest_priority_class])
288 if ctx.highest_priority_class
289 else None
290 )
291 entity_label_for_desc = ctx.custom_filter.human_readable_entity(
292 ctx.entity_uri, (ctx.highest_priority_class, shape), context_snapshot
293 )
294 if entity_label_for_desc and entity_label_for_desc != ctx.entity_uri:
295 description = description.replace(
296 f"'{ctx.entity_uri}'", f"'{entity_label_for_desc}'"
297 )
299 return description
302def _resolve_timestamp(entity_uri: str, timestamp: str) -> tuple[str, datetime]:
303 try:
304 return timestamp, datetime.fromisoformat(timestamp)
305 except ValueError:
306 pass
308 provenance_sparql = get_provenance_sparql()
309 query_timestamp = f"""
310 SELECT ?generation_time
311 WHERE {{
312 <{entity_uri}/prov/se/{timestamp}>
313 <http://www.w3.org/ns/prov#generatedAtTime>
314 ?generation_time.
315 }}
316 """
317 provenance_sparql.setQuery(query_timestamp)
318 provenance_sparql.setReturnFormat(JSON)
319 try:
320 bindings = get_sparql_bindings(provenance_sparql.queryAndConvert())
321 generation_time = bindings[0]["generation_time"]["value"]
322 except IndexError:
323 abort(404)
324 return generation_time, datetime.fromisoformat(generation_time)
327def _find_closest_metadata(
328 entity_metadata: dict,
329 timestamp_dt: datetime,
330 latest_timestamp: str,
331) -> tuple[dict | None, dict | None]:
332 closest_metadata = None
333 min_time_diff = None
334 latest_metadata = None
336 for meta in entity_metadata.values():
337 meta_time = convert_to_datetime(meta["generatedAtTime"])
338 if meta_time is None:
339 msg = "meta_time must not be None"
340 raise AssertionError(msg)
341 time_diff = abs((meta_time - timestamp_dt).total_seconds())
343 if (
344 closest_metadata is None
345 or min_time_diff is None
346 or time_diff < min_time_diff
347 ):
348 closest_metadata = meta
349 min_time_diff = time_diff
351 if meta["generatedAtTime"] == latest_timestamp:
352 latest_metadata = meta
354 return closest_metadata, latest_metadata
357def _compute_version_navigation(
358 snapshot_times: list[datetime],
359 timestamp_dt: datetime,
360) -> tuple[str | None, str | None]:
361 next_snapshot_timestamp = None
362 prev_snapshot_timestamp = None
364 for snap_time in snapshot_times:
365 if snap_time > timestamp_dt:
366 next_snapshot_timestamp = snap_time.isoformat()
367 break
369 for snap_time in reversed(snapshot_times):
370 if snap_time < timestamp_dt:
371 prev_snapshot_timestamp = snap_time.isoformat()
372 break
374 return prev_snapshot_timestamp, next_snapshot_timestamp
377def _prepare_modifications(
378 closest_metadata: dict,
379 ctx: HistoryContext,
380 context_version: Graph,
381 closest_timestamp: str,
382 sorted_timestamps: list[str],
383) -> tuple[str, dict]:
384 modifications = ""
385 if closest_metadata.get("hasUpdateQuery"):
386 sparql_query = closest_metadata["hasUpdateQuery"]
387 parsed_modifications = parse_sparql_update(sparql_query)
388 modifications = generate_modification_text(
389 parsed_modifications,
390 ctx,
391 context_version,
392 closest_timestamp,
393 )
395 try:
396 current_index = sorted_timestamps.index(closest_timestamp)
397 except ValueError:
398 current_index = -1
400 if closest_metadata.get("description"):
401 formatted_description = _format_snapshot_description(
402 closest_metadata,
403 ctx,
404 context_version,
405 current_index,
406 )
407 closest_metadata["description"] = formatted_description
409 return modifications, closest_metadata
412@entity_bp.route("/entity-version/<path:entity_uri>/<timestamp>")
413@login_required
414def entity_version(entity_uri: str, timestamp: str) -> str:
415 entity_uri_ref = URIRef(entity_uri)
416 custom_filter = get_custom_filter()
417 change_tracking_config = get_change_tracking_config()
419 timestamp, timestamp_dt = _resolve_timestamp(entity_uri, timestamp)
421 agnostic_entity = AgnosticEntity(
422 res=entity_uri,
423 config=change_tracking_config,
424 include_related_objects=True,
425 include_merged_entities=True,
426 include_reverse_relations=get_display_rules_use_inverse_relations(),
427 )
428 history, provenance = agnostic_entity.get_history(include_prov_metadata=True)
429 history = convert_to_rdflib_graphs(history, is_quadstore=get_dataset_is_quadstore())
430 main_entity_history = history.get(entity_uri, {})
431 sorted_timestamps = sorted(
432 main_entity_history.keys(),
433 key=lambda t: convert_to_datetime(t) or _DATETIME_MIN_UTC,
434 )
436 if not sorted_timestamps:
437 abort(404)
439 closest_timestamp = min(
440 sorted_timestamps,
441 key=lambda t: abs(
442 (convert_to_datetime(t) or _DATETIME_MIN_UTC).astimezone()
443 - timestamp_dt.astimezone()
444 ),
445 )
447 version = main_entity_history[closest_timestamp]
448 triples: list[tuple[URIRef, URIRef, URIRef | Literal]] = [
449 (URIRef(str(s)), URIRef(str(p)), URIRef(str(o)) if isinstance(o, URIRef) else o) # type: ignore[misc]
450 for s, p, o in get_triples_from_graph(version, (entity_uri_ref, None, None))
451 ]
453 entity_metadata = provenance.get(entity_uri, {})
454 latest_timestamp = max(sorted_timestamps)
455 closest_metadata, latest_metadata = _find_closest_metadata(
456 entity_metadata, timestamp_dt, latest_timestamp
457 )
459 if closest_metadata is None or latest_metadata is None:
460 abort(404)
462 is_deletion_snapshot = (
463 closest_timestamp == latest_timestamp
464 and "invalidatedAtTime" in latest_metadata
465 and latest_metadata["invalidatedAtTime"]
466 ) or len(triples) == 0
468 context_version = version
469 if is_deletion_snapshot and len(sorted_timestamps) > 1:
470 current_index = sorted_timestamps.index(closest_timestamp)
471 if current_index > 0:
472 context_version = main_entity_history[sorted_timestamps[current_index - 1]]
474 if is_deletion_snapshot and len(sorted_timestamps) > 1:
475 subject_classes = [
476 str(o)
477 for _, _, o in get_triples_from_graph(
478 context_version, (entity_uri_ref, RDF.type, None)
479 )
480 ]
481 else:
482 subject_classes = [
483 str(o)
484 for _, _, o in get_triples_from_graph(
485 version, (entity_uri_ref, RDF.type, None)
486 )
487 ]
489 highest_priority_class = get_highest_priority_class(subject_classes)
491 entity_shape = determine_shape_for_entity_triples(
492 list(get_triples_from_graph(context_version, (entity_uri_ref, None, None)))
493 )
495 _, _, _, _, _, valid_predicates_set = get_valid_predicates(
496 triples, highest_priority_class=URIRef(highest_priority_class or "")
497 )
499 grouped_triples, _relevant_properties = get_grouped_triples(
500 entity_uri_ref,
501 triples,
502 list(valid_predicates_set),
503 historical_snapshot=context_version,
504 entity_key=(highest_priority_class, entity_shape),
505 )
507 snapshot_times: list[datetime] = [
508 dt
509 for meta in entity_metadata.values()
510 if (dt := convert_to_datetime(meta["generatedAtTime"])) is not None
511 ]
512 snapshot_times = sorted(set(snapshot_times))
513 version_number = snapshot_times.index(timestamp_dt) + 1
515 prev_snapshot_timestamp, next_snapshot_timestamp = _compute_version_navigation(
516 snapshot_times, timestamp_dt
517 )
519 version_history_ctx = HistoryContext(
520 entity_uri=entity_uri,
521 highest_priority_class=highest_priority_class,
522 entity_shape=entity_shape,
523 history=history,
524 sorted_timestamps=sorted_timestamps,
525 custom_filter=custom_filter,
526 )
528 modifications, closest_metadata = _prepare_modifications(
529 closest_metadata,
530 version_history_ctx,
531 context_version,
532 closest_timestamp,
533 sorted_timestamps,
534 )
536 closest_timestamp = closest_metadata["generatedAtTime"]
538 return render_template(
539 "entity/version.jinja",
540 subject=entity_uri,
541 entity_type=highest_priority_class,
542 entity_shape=entity_shape,
543 metadata={closest_timestamp: closest_metadata},
544 timestamp=closest_timestamp,
545 next_snapshot_timestamp=next_snapshot_timestamp,
546 prev_snapshot_timestamp=prev_snapshot_timestamp,
547 modifications=modifications,
548 grouped_triples=grouped_triples,
549 version_number=version_number,
550 version=context_version,
551 )