Coverage for heritrace/routes/merge.py: 99%
254 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-13 22:01 +0000
1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5from __future__ import annotations
7from collections import defaultdict
8from typing import TYPE_CHECKING, Any
10import validators
11from flask import (
12 Blueprint,
13 Response,
14 current_app,
15 flash,
16 jsonify,
17 redirect,
18 render_template,
19 request,
20 url_for,
21)
22from flask_babel import gettext
23from flask_login import current_user, login_required
24from markupsafe import Markup
25from rdflib import URIRef
26from SPARQLWrapper import JSON
28from heritrace.apis.orcid import get_responsible_agent_uri
29from heritrace.editor import Editor, EndpointConfig
30from heritrace.extensions import (
31 get_counter_handler,
32 get_custom_filter,
33 get_dataset_endpoint,
34 get_dataset_is_quadstore,
35 get_provenance_endpoint,
36 get_sparql,
37)
38from heritrace.sparql import get_sparql_bindings
39from heritrace.utils.display_rules_utils import (
40 get_highest_priority_class,
41 get_similarity_properties,
42)
43from heritrace.utils.primary_source_utils import (
44 get_user_default_primary_source,
45 save_user_default_primary_source,
46)
47from heritrace.utils.shacl_utils import determine_shape_for_classes
48from heritrace.utils.sparql_utils import get_entity_types, import_entity_graph
49from heritrace.utils.virtuoso_utils import is_virtuoso
51if TYPE_CHECKING:
52 from werkzeug.wrappers import Response as WerkzeugResponse
54merge_bp = Blueprint("merge", __name__)
57def get_entity_details(
58 entity_uri: URIRef,
59) -> tuple[dict[str, list[dict[str, Any]]] | None, list[str]]:
60 """
61 Fetches all properties (predicates and objects) for a given entity URI,
62 grouped by predicate, along with its types.
64 Args:
65 entity_uri: The URI of the entity to fetch details for.
67 Returns:
68 A tuple containing:
69 - A dictionary where keys are predicate URIs and values are lists of
70 object dictionaries (containing 'value', 'type', 'lang', 'datatype').
71 Returns None if an error occurs.
72 - A list of entity type URIs. Returns an empty list if an error occurs
73 or no types are found.
74 """
75 sparql = get_sparql()
76 custom_filter = get_custom_filter()
77 grouped_properties: dict[str, list[dict[str, Any]]] = {}
78 entity_types: list[str] = []
80 try:
81 entity_types = get_entity_types(entity_uri)
82 if not entity_types:
83 current_app.logger.warning("No types found for entity: %s", entity_uri)
85 query = f"""
86 SELECT DISTINCT ?p ?o WHERE {{
87 <{entity_uri}> ?p ?o .
88 }}
89 """
90 sparql.setQuery(query)
91 sparql.setReturnFormat(JSON)
92 results = sparql.query().convert()
94 bindings = get_sparql_bindings(results)
95 for binding in bindings:
96 predicate = binding["p"]["value"]
97 obj_node = binding["o"]
98 obj_details = {
99 "value": obj_node["value"],
100 "type": obj_node["type"],
101 "lang": obj_node.get("xml:lang"),
102 "datatype": obj_node.get("datatype"),
103 "readable_label": None,
104 }
105 if obj_details["type"] == "uri":
106 obj_types = get_entity_types(URIRef(obj_details["value"]))
107 obj_type = get_highest_priority_class(obj_types)
108 if obj_type:
109 obj_details["readable_label"] = custom_filter.human_readable_entity(
110 obj_details["value"], (obj_type, None)
111 )
112 else:
113 obj_details["readable_label"] = obj_details["value"]
114 else:
115 obj_details["readable_label"] = obj_details["value"]
117 if predicate not in grouped_properties:
118 grouped_properties[predicate] = []
119 grouped_properties[predicate].append(obj_details)
121 except Exception:
122 current_app.logger.exception(
123 "Error fetching details for %s",
124 entity_uri,
125 )
126 return None, []
127 else:
128 return grouped_properties, entity_types
131@merge_bp.route("/execute-merge", methods=["POST"])
132@login_required
133def execute_merge() -> WerkzeugResponse:
134 """
135 Handles the actual merging of two entities using the Editor class
136 to ensure provenance and data model agnosticism.
137 Entity 1 (keep) absorbs Entity 2 (delete).
138 """
139 entity1_uri_str = request.form.get("entity1_uri")
140 entity2_uri_str = request.form.get("entity2_uri")
141 primary_source = request.form.get("primary_source")
142 save_default_source = request.form.get("save_default_source") == "true"
144 # TODO(arcangelo): Implement CSRF validation
145 # if using Flask-WTF
147 if not entity1_uri_str or not entity2_uri_str:
148 flash(gettext("Missing entity URIs for merge."), "danger")
149 return redirect(url_for("main.catalogue"))
151 entity1_uri = URIRef(entity1_uri_str)
152 entity2_uri = URIRef(entity2_uri_str)
154 if primary_source and not validators.url(primary_source): # type: ignore[arg-type]
155 flash(gettext("Invalid primary source URL provided."), "danger")
156 return redirect(
157 url_for(
158 ".compare_and_merge", subject=entity1_uri, other_subject=entity2_uri
159 )
160 )
162 if save_default_source and primary_source and validators.url(primary_source): # type: ignore[arg-type]
163 save_user_default_primary_source(current_user.orcid, primary_source)
165 try:
166 custom_filter = get_custom_filter()
168 _, entity1_types = get_entity_details(entity1_uri)
169 _, entity2_types = get_entity_details(entity2_uri)
171 entity1_type = get_highest_priority_class(entity1_types)
172 entity2_type = get_highest_priority_class(entity2_types)
173 entity1_shape = determine_shape_for_classes(entity1_types)
174 entity2_shape = determine_shape_for_classes(entity2_types)
175 entity1_label = (
176 custom_filter.human_readable_entity(
177 entity1_uri, (entity1_type, entity1_shape)
178 )
179 if entity1_type
180 else entity1_uri
181 )
182 entity2_label = (
183 custom_filter.human_readable_entity(
184 entity2_uri, (entity2_type, entity2_shape)
185 )
186 if entity2_type
187 else entity2_uri
188 )
190 counter_handler = get_counter_handler()
191 resp_agent = get_responsible_agent_uri(current_user.orcid)
193 dataset_endpoint = get_dataset_endpoint()
194 provenance_endpoint = get_provenance_endpoint()
195 dataset_is_quadstore = get_dataset_is_quadstore()
197 editor = Editor(
198 EndpointConfig(
199 dataset=dataset_endpoint,
200 provenance=provenance_endpoint,
201 is_quadstore=dataset_is_quadstore,
202 ),
203 counter_handler,
204 resp_agent,
205 URIRef(current_app.config["PRIMARY_SOURCE"]),
206 current_app.config["DATASET_GENERATION_TIME"],
207 save_plugin=current_app.config.get("SAVE_PLUGIN"),
208 )
210 editor = import_entity_graph(editor, entity1_uri)
211 editor = import_entity_graph(editor, entity2_uri)
212 editor.merge(
213 keep_entity_uri=entity1_uri,
214 delete_entity_uri=entity2_uri,
215 primary_source=URIRef(primary_source) if primary_source else None,
216 )
218 entity1_url = url_for("entity.about", subject=entity1_uri)
219 entity2_url = url_for("entity.about", subject=entity2_uri)
220 flash_message_html = gettext(
221 "Entities merged successfully. "
222 "<a href='%(entity2_url)s' target='_blank'>%(entity2)s</a> "
223 "has been deleted and its references now point to "
224 "<a href='%(entity1_url)s' target='_blank'>%(entity1)s</a>.",
225 entity1=entity1_label,
226 entity2=entity2_label,
227 entity1_url=entity1_url,
228 entity2_url=entity2_url,
229 )
231 flash(Markup(flash_message_html), "success") # noqa: S704
233 return redirect(url_for("entity.about", subject=entity1_uri))
235 except ValueError as ve:
236 current_app.logger.warning("Merge attempt failed: %s", ve)
237 flash(str(ve), "warning")
238 return redirect(
239 url_for(
240 ".compare_and_merge", subject=entity1_uri, other_subject=entity2_uri
241 )
242 )
244 except Exception:
245 current_app.logger.exception(
246 "Error executing Editor merge for <%s> and <%s>",
247 entity1_uri,
248 entity2_uri,
249 )
250 flash(
251 gettext(
252 "An error occurred during the merge"
253 " operation. Please check the logs."
254 " No changes were made."
255 ),
256 "danger",
257 )
258 return redirect(
259 url_for(
260 ".compare_and_merge", subject=entity1_uri, other_subject=entity2_uri
261 )
262 )
265@merge_bp.route("/compare-and-merge")
266@login_required
267def compare_and_merge() -> str | WerkzeugResponse:
268 """
269 Route to display details of two entities side-by-side for merge confirmation.
270 """
271 entity1_uri_str = request.args.get("subject")
272 entity2_uri_str = request.args.get("other_subject")
273 custom_filter = get_custom_filter()
275 if not entity1_uri_str or not entity2_uri_str:
276 flash(
277 gettext("Two entities must be selected for merging/comparison."), "warning"
278 )
279 return redirect(url_for("main.catalogue"))
281 entity1_uri = URIRef(entity1_uri_str)
282 entity2_uri = URIRef(entity2_uri_str)
284 entity1_props, entity1_types = get_entity_details(entity1_uri)
285 entity2_props, entity2_types = get_entity_details(entity2_uri)
287 if entity1_props is None or entity2_props is None:
288 flash(
289 gettext("Could not retrieve details for one or both entities. Check logs."),
290 "danger",
291 )
292 return redirect(url_for("main.catalogue"))
294 entity1_type = get_highest_priority_class(entity1_types)
295 entity2_type = get_highest_priority_class(entity2_types)
296 entity1_shape = determine_shape_for_classes(entity1_types)
297 entity2_shape = determine_shape_for_classes(entity2_types)
298 entity1_label = (
299 custom_filter.human_readable_entity(entity1_uri, (entity1_type, entity1_shape))
300 if entity1_type
301 else entity1_uri
302 )
303 entity2_label = (
304 custom_filter.human_readable_entity(entity2_uri, (entity2_type, entity2_shape))
305 if entity2_type
306 else entity2_uri
307 )
309 entity1_data = {
310 "uri": entity1_uri,
311 "label": entity1_label,
312 "type_label": custom_filter.human_readable_class((entity1_type, entity1_shape)),
313 "type": entity1_type,
314 "shape": entity1_shape,
315 "properties": entity1_props,
316 }
317 entity2_data = {
318 "uri": entity2_uri,
319 "label": entity2_label,
320 "type_label": custom_filter.human_readable_class((entity2_type, entity2_shape)),
321 "type": entity2_type,
322 "shape": entity2_shape,
323 "properties": entity2_props,
324 }
326 default_primary_source = get_user_default_primary_source(current_user.orcid)
328 return render_template(
329 "entity/merge_confirm.jinja",
330 entity1=entity1_data,
331 entity2=entity2_data,
332 default_primary_source=default_primary_source,
333 )
336def _format_rdf_term(node: dict[str, str]) -> str | None:
337 value = node["value"]
338 value_type = node["type"]
339 if value_type == "uri":
340 return f"<{value}>"
341 if value_type in {"literal", "typed-literal"}:
342 datatype = node.get("datatype")
343 lang = node.get("xml:lang")
344 escaped_value = value.replace("\\", "\\\\").replace('"', '\\"')
345 if datatype:
346 return f'"{escaped_value}"^^<{datatype}>'
347 if lang:
348 return f'"{escaped_value}"@{lang}'
349 return f'"{escaped_value}"'
350 return None
353def _fetch_subject_values(
354 subject_uri: str,
355 similarity_config: list,
356) -> defaultdict[str, list[str]] | None:
357 sparql = get_sparql()
359 all_props_in_config: set[str] = set()
360 for item in similarity_config:
361 if isinstance(item, str):
362 all_props_in_config.add(item)
363 elif isinstance(item, dict) and "and" in item:
364 all_props_in_config.update(item["and"])
366 if not all_props_in_config:
367 current_app.logger.warning(
368 "Empty properties list derived from similarity config for type %s",
369 subject_uri,
370 )
371 return None
373 prop_uris_formatted_for_filter = [f"<{p}>" for p in all_props_in_config]
374 property_filter_for_subject = (
375 f"FILTER(?p IN ({', '.join(prop_uris_formatted_for_filter)}))"
376 )
378 fetch_comparison_values_query = f"""
379 SELECT DISTINCT ?p ?o WHERE {{
380 <{subject_uri}> ?p ?o .
381 {property_filter_for_subject}
382 }}
383 """
385 sparql.setQuery(fetch_comparison_values_query)
386 sparql.setReturnFormat(JSON)
387 subject_values_results = sparql.query().convert()
388 subject_bindings = get_sparql_bindings(subject_values_results)
390 if not subject_bindings:
391 return None
393 subject_values_by_prop: defaultdict[str, list[str]] = defaultdict(list)
394 for binding in subject_bindings:
395 formatted_value = _format_rdf_term(binding["o"])
396 if formatted_value:
397 subject_values_by_prop[binding["p"]["value"]].append(formatted_value)
399 return subject_values_by_prop
402def _build_similarity_pattern(prop_uri: str, values: list[str], index: int) -> str:
403 variable = f"?o_{index}"
404 if not is_virtuoso():
405 return (
406 f"VALUES {variable} {{ {' '.join(values)} }} "
407 f"?similar <{prop_uri}> {variable} ."
408 )
409 return (
410 f"?similar <{prop_uri}> {variable} . "
411 f"FILTER({variable} IN ({', '.join(values)}))"
412 )
415def _build_union_blocks(
416 similarity_config: list,
417 subject_values_by_prop: defaultdict[str, list[str]],
418 subject_uri: str,
419) -> list[str]:
420 union_blocks: list[str] = []
421 var_counter = 0
423 for condition in similarity_config:
424 if isinstance(condition, str):
425 prop_values = subject_values_by_prop.get(condition)
426 if prop_values:
427 var_counter += 1
428 pattern = _build_similarity_pattern(condition, prop_values, var_counter)
429 union_blocks.append(f" {{ {pattern} }}")
430 elif isinstance(condition, dict) and "and" in condition:
431 block = _build_and_block(
432 condition["and"], subject_values_by_prop, subject_uri, var_counter
433 )
434 if block is not None:
435 text, var_counter = block
436 union_blocks.append(text)
437 else:
438 var_counter += len(condition["and"])
440 return union_blocks
443def _build_and_block(
444 and_props: list[str],
445 subject_values_by_prop: defaultdict[str, list[str]],
446 subject_uri: str,
447 var_counter: int,
448) -> tuple[str, int] | None:
449 if not all(p in subject_values_by_prop for p in and_props):
450 current_app.logger.debug(
451 "Skipping AND group %s because"
452 " subject %s lacks values for"
453 " all its properties.",
454 and_props,
455 subject_uri,
456 )
457 return None
459 and_patterns = []
460 for prop_uri in and_props:
461 prop_values = subject_values_by_prop[prop_uri]
462 var_counter += 1
463 and_patterns.append(
464 f" {_build_similarity_pattern(prop_uri, prop_values, var_counter)}"
465 )
467 patterns_str = "\n".join(and_patterns)
468 return f" {{\n{patterns_str}\n }}", var_counter
471def _execute_similarity_query(
472 union_blocks: list[str],
473 entity_type: str,
474 subject_uri: str,
475 limit: int,
476 offset: int,
477) -> tuple[list[str], bool]:
478 sparql = get_sparql()
479 similarity_query_body = " UNION ".join(union_blocks)
481 query_limit = limit + 1
482 final_query = f"""
483 SELECT DISTINCT ?similar WHERE {{
484 ?similar a <{entity_type}> .
485 FILTER(?similar != <{subject_uri}>)
486 {{
487 {similarity_query_body}
488 }}
489 }} ORDER BY ?similar OFFSET {offset} LIMIT {query_limit}
490 """
492 sparql.setQuery(final_query)
493 sparql.setReturnFormat(JSON)
494 results = sparql.query().convert()
496 bindings = get_sparql_bindings(results)
497 candidate_uris = [item["similar"]["value"] for item in bindings]
499 has_more = len(candidate_uris) > limit
500 return candidate_uris[:limit], has_more
503def _transform_results(
504 uris: list[str],
505 entity_type: str,
506 shape_uri: str | None,
507) -> list[dict[str, str]]:
508 custom_filter = get_custom_filter()
509 transformed: list[dict[str, str]] = []
510 for uri in uris:
511 readable_label = (
512 custom_filter.human_readable_entity(uri, (entity_type, shape_uri))
513 if entity_type
514 else uri
515 )
516 transformed.append({"uri": uri, "label": readable_label or uri})
517 return transformed
520@merge_bp.route("/find_similar", methods=["GET"])
521@login_required
522def find_similar_resources() -> Response | tuple[Response, int]: # noqa: PLR0911
523 subject_uri = request.args.get("subject_uri")
524 entity_type = request.args.get("entity_type")
525 shape_uri = request.args.get("shape_uri")
526 try:
527 limit = int(request.args.get("limit", 5))
528 offset = int(request.args.get("offset", 0))
529 except ValueError:
530 return jsonify(
531 {"status": "error", "message": gettext("Invalid limit or offset parameter")}
532 ), 400
534 if not subject_uri or not entity_type:
535 return jsonify(
536 {
537 "status": "error",
538 "message": gettext(
539 "Missing required parameters (subject_uri, entity_type)"
540 ),
541 }
542 ), 400
544 if limit <= 0 or offset < 0:
545 return jsonify(
546 {
547 "status": "error",
548 "message": gettext("Limit must be positive and offset non-negative"),
549 }
550 ), 400
552 try:
553 entity_key = (entity_type, shape_uri)
554 similarity_config = get_similarity_properties(entity_key)
556 if not similarity_config or not isinstance(similarity_config, list):
557 return jsonify({"status": "success", "results": [], "has_more": False})
559 subject_values_by_prop = _fetch_subject_values(subject_uri, similarity_config)
560 if subject_values_by_prop is None:
561 return jsonify({"status": "success", "results": [], "has_more": False})
563 union_blocks = _build_union_blocks(
564 similarity_config, subject_values_by_prop, subject_uri
565 )
566 if not union_blocks:
567 return jsonify({"status": "success", "results": [], "has_more": False})
569 result_uris, has_more = _execute_similarity_query(
570 union_blocks, entity_type, subject_uri, limit, offset
571 )
572 transformed_results = _transform_results(result_uris, entity_type, shape_uri)
574 return jsonify(
575 {
576 "status": "success",
577 "results": transformed_results,
578 "has_more": has_more,
579 }
580 )
582 except Exception:
583 current_app.logger.exception(
584 "Error finding similar resources for %s", subject_uri
585 )
586 return jsonify(
587 {
588 "status": "error",
589 "message": gettext("An error occurred while finding similar resources"),
590 }
591 ), 500