Coverage for heritrace/routes/merge.py: 99%
249 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +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
50if TYPE_CHECKING:
51 from werkzeug.wrappers import Response as WerkzeugResponse
53merge_bp = Blueprint("merge", __name__)
56def get_entity_details(
57 entity_uri: URIRef,
58) -> tuple[dict[str, list[dict[str, Any]]] | None, list[str]]:
59 """
60 Fetches all properties (predicates and objects) for a given entity URI,
61 grouped by predicate, along with its types.
63 Args:
64 entity_uri: The URI of the entity to fetch details for.
66 Returns:
67 A tuple containing:
68 - A dictionary where keys are predicate URIs and values are lists of
69 object dictionaries (containing 'value', 'type', 'lang', 'datatype').
70 Returns None if an error occurs.
71 - A list of entity type URIs. Returns an empty list if an error occurs
72 or no types are found.
73 """
74 sparql = get_sparql()
75 custom_filter = get_custom_filter()
76 grouped_properties: dict[str, list[dict[str, Any]]] = {}
77 entity_types: list[str] = []
79 try:
80 entity_types = get_entity_types(entity_uri)
81 if not entity_types:
82 current_app.logger.warning("No types found for entity: %s", entity_uri)
84 query = f"""
85 SELECT DISTINCT ?p ?o WHERE {{
86 <{entity_uri}> ?p ?o .
87 }}
88 """
89 sparql.setQuery(query)
90 sparql.setReturnFormat(JSON)
91 results = sparql.query().convert()
93 bindings = get_sparql_bindings(results)
94 for binding in bindings:
95 predicate = binding["p"]["value"]
96 obj_node = binding["o"]
97 obj_details = {
98 "value": obj_node["value"],
99 "type": obj_node["type"],
100 "lang": obj_node.get("xml:lang"),
101 "datatype": obj_node.get("datatype"),
102 "readable_label": None,
103 }
104 if obj_details["type"] == "uri":
105 obj_types = get_entity_types(URIRef(obj_details["value"]))
106 obj_type = get_highest_priority_class(obj_types)
107 if obj_type:
108 obj_details["readable_label"] = custom_filter.human_readable_entity(
109 obj_details["value"], (obj_type, None)
110 )
111 else:
112 obj_details["readable_label"] = obj_details["value"]
113 else:
114 obj_details["readable_label"] = obj_details["value"]
116 if predicate not in grouped_properties:
117 grouped_properties[predicate] = []
118 grouped_properties[predicate].append(obj_details)
120 except Exception:
121 current_app.logger.exception(
122 "Error fetching details for %s",
123 entity_uri,
124 )
125 return None, []
126 else:
127 return grouped_properties, entity_types
130@merge_bp.route("/execute-merge", methods=["POST"])
131@login_required
132def execute_merge() -> WerkzeugResponse:
133 """
134 Handles the actual merging of two entities using the Editor class
135 to ensure provenance and data model agnosticism.
136 Entity 1 (keep) absorbs Entity 2 (delete).
137 """
138 entity1_uri_str = request.form.get("entity1_uri")
139 entity2_uri_str = request.form.get("entity2_uri")
140 primary_source = request.form.get("primary_source")
141 save_default_source = request.form.get("save_default_source") == "true"
143 # TODO(arcangelo): Implement CSRF validation
144 # if using Flask-WTF
146 if not entity1_uri_str or not entity2_uri_str:
147 flash(gettext("Missing entity URIs for merge."), "danger")
148 return redirect(url_for("main.catalogue"))
150 entity1_uri = URIRef(entity1_uri_str)
151 entity2_uri = URIRef(entity2_uri_str)
153 if primary_source and not validators.url(primary_source): # type: ignore[arg-type]
154 flash(gettext("Invalid primary source URL provided."), "danger")
155 return redirect(
156 url_for(
157 ".compare_and_merge", subject=entity1_uri, other_subject=entity2_uri
158 )
159 )
161 if save_default_source and primary_source and validators.url(primary_source): # type: ignore[arg-type]
162 save_user_default_primary_source(current_user.orcid, primary_source)
164 try:
165 custom_filter = get_custom_filter()
167 _, entity1_types = get_entity_details(entity1_uri)
168 _, entity2_types = get_entity_details(entity2_uri)
170 entity1_type = get_highest_priority_class(entity1_types)
171 entity2_type = get_highest_priority_class(entity2_types)
172 entity1_shape = determine_shape_for_classes(entity1_types)
173 entity2_shape = determine_shape_for_classes(entity2_types)
174 entity1_label = (
175 custom_filter.human_readable_entity(
176 entity1_uri, (entity1_type, entity1_shape)
177 )
178 if entity1_type
179 else entity1_uri
180 )
181 entity2_label = (
182 custom_filter.human_readable_entity(
183 entity2_uri, (entity2_type, entity2_shape)
184 )
185 if entity2_type
186 else entity2_uri
187 )
189 counter_handler = get_counter_handler()
190 resp_agent = get_responsible_agent_uri(current_user.orcid)
192 dataset_endpoint = get_dataset_endpoint()
193 provenance_endpoint = get_provenance_endpoint()
194 dataset_is_quadstore = get_dataset_is_quadstore()
196 editor = Editor(
197 EndpointConfig(
198 dataset=dataset_endpoint,
199 provenance=provenance_endpoint,
200 is_quadstore=dataset_is_quadstore,
201 ),
202 counter_handler,
203 resp_agent,
204 URIRef(current_app.config["PRIMARY_SOURCE"]),
205 current_app.config["DATASET_GENERATION_TIME"],
206 save_plugin=current_app.config.get("SAVE_PLUGIN"),
207 )
209 editor = import_entity_graph(editor, entity1_uri)
210 editor = import_entity_graph(editor, entity2_uri)
211 editor.merge(
212 keep_entity_uri=entity1_uri,
213 delete_entity_uri=entity2_uri,
214 primary_source=URIRef(primary_source) if primary_source else None,
215 )
217 entity1_url = url_for("entity.about", subject=entity1_uri)
218 entity2_url = url_for("entity.about", subject=entity2_uri)
219 flash_message_html = gettext(
220 "Entities merged successfully. "
221 "<a href='%(entity2_url)s' target='_blank'>%(entity2)s</a> "
222 "has been deleted and its references now point to "
223 "<a href='%(entity1_url)s' target='_blank'>%(entity1)s</a>.",
224 entity1=entity1_label,
225 entity2=entity2_label,
226 entity1_url=entity1_url,
227 entity2_url=entity2_url,
228 )
230 flash(Markup(flash_message_html), "success") # noqa: S704
232 return redirect(url_for("entity.about", subject=entity1_uri))
234 except ValueError as ve:
235 current_app.logger.warning("Merge attempt failed: %s", ve)
236 flash(str(ve), "warning")
237 return redirect(
238 url_for(
239 ".compare_and_merge", subject=entity1_uri, other_subject=entity2_uri
240 )
241 )
243 except Exception:
244 current_app.logger.exception(
245 "Error executing Editor merge for <%s> and <%s>",
246 entity1_uri,
247 entity2_uri,
248 )
249 flash(
250 gettext(
251 "An error occurred during the merge"
252 " operation. Please check the logs."
253 " No changes were made."
254 ),
255 "danger",
256 )
257 return redirect(
258 url_for(
259 ".compare_and_merge", subject=entity1_uri, other_subject=entity2_uri
260 )
261 )
264@merge_bp.route("/compare-and-merge")
265@login_required
266def compare_and_merge() -> str | WerkzeugResponse:
267 """
268 Route to display details of two entities side-by-side for merge confirmation.
269 """
270 entity1_uri_str = request.args.get("subject")
271 entity2_uri_str = request.args.get("other_subject")
272 custom_filter = get_custom_filter()
274 if not entity1_uri_str or not entity2_uri_str:
275 flash(
276 gettext("Two entities must be selected for merging/comparison."), "warning"
277 )
278 return redirect(url_for("main.catalogue"))
280 entity1_uri = URIRef(entity1_uri_str)
281 entity2_uri = URIRef(entity2_uri_str)
283 entity1_props, entity1_types = get_entity_details(entity1_uri)
284 entity2_props, entity2_types = get_entity_details(entity2_uri)
286 if entity1_props is None or entity2_props is None:
287 flash(
288 gettext("Could not retrieve details for one or both entities. Check logs."),
289 "danger",
290 )
291 return redirect(url_for("main.catalogue"))
293 entity1_type = get_highest_priority_class(entity1_types)
294 entity2_type = get_highest_priority_class(entity2_types)
295 entity1_shape = determine_shape_for_classes(entity1_types)
296 entity2_shape = determine_shape_for_classes(entity2_types)
297 entity1_label = (
298 custom_filter.human_readable_entity(entity1_uri, (entity1_type, entity1_shape))
299 if entity1_type
300 else entity1_uri
301 )
302 entity2_label = (
303 custom_filter.human_readable_entity(entity2_uri, (entity2_type, entity2_shape))
304 if entity2_type
305 else entity2_uri
306 )
308 entity1_data = {
309 "uri": entity1_uri,
310 "label": entity1_label,
311 "type_label": custom_filter.human_readable_class((entity1_type, entity1_shape)),
312 "type": entity1_type,
313 "shape": entity1_shape,
314 "properties": entity1_props,
315 }
316 entity2_data = {
317 "uri": entity2_uri,
318 "label": entity2_label,
319 "type_label": custom_filter.human_readable_class((entity2_type, entity2_shape)),
320 "type": entity2_type,
321 "shape": entity2_shape,
322 "properties": entity2_props,
323 }
325 default_primary_source = get_user_default_primary_source(current_user.orcid)
327 return render_template(
328 "entity/merge_confirm.jinja",
329 entity1=entity1_data,
330 entity2=entity2_data,
331 default_primary_source=default_primary_source,
332 )
335def _format_rdf_term(node: dict[str, str]) -> str | None:
336 value = node["value"]
337 value_type = node["type"]
338 if value_type == "uri":
339 return f"<{value}>"
340 if value_type in {"literal", "typed-literal"}:
341 datatype = node.get("datatype")
342 lang = node.get("xml:lang")
343 escaped_value = value.replace("\\", "\\\\").replace('"', '\\"')
344 if datatype:
345 return f'"{escaped_value}"^^<{datatype}>'
346 if lang:
347 return f'"{escaped_value}"@{lang}'
348 return f'"{escaped_value}"'
349 return None
352def _fetch_subject_values(
353 subject_uri: str,
354 similarity_config: list,
355) -> defaultdict[str, list[str]] | None:
356 sparql = get_sparql()
358 all_props_in_config: set[str] = set()
359 for item in similarity_config:
360 if isinstance(item, str):
361 all_props_in_config.add(item)
362 elif isinstance(item, dict) and "and" in item:
363 all_props_in_config.update(item["and"])
365 if not all_props_in_config:
366 current_app.logger.warning(
367 "Empty properties list derived from similarity config for type %s",
368 subject_uri,
369 )
370 return None
372 prop_uris_formatted_for_filter = [f"<{p}>" for p in all_props_in_config]
373 property_filter_for_subject = (
374 f"FILTER(?p IN ({', '.join(prop_uris_formatted_for_filter)}))"
375 )
377 fetch_comparison_values_query = f"""
378 SELECT DISTINCT ?p ?o WHERE {{
379 <{subject_uri}> ?p ?o .
380 {property_filter_for_subject}
381 }}
382 """
384 sparql.setQuery(fetch_comparison_values_query)
385 sparql.setReturnFormat(JSON)
386 subject_values_results = sparql.query().convert()
387 subject_bindings = get_sparql_bindings(subject_values_results)
389 if not subject_bindings:
390 return None
392 subject_values_by_prop: defaultdict[str, list[str]] = defaultdict(list)
393 for binding in subject_bindings:
394 formatted_value = _format_rdf_term(binding["o"])
395 if formatted_value:
396 subject_values_by_prop[binding["p"]["value"]].append(formatted_value)
398 return subject_values_by_prop
401def _build_union_blocks(
402 similarity_config: list,
403 subject_values_by_prop: defaultdict[str, list[str]],
404 subject_uri: str,
405) -> list[str]:
406 union_blocks: list[str] = []
407 var_counter = 0
409 for condition in similarity_config:
410 if isinstance(condition, str):
411 prop_values = subject_values_by_prop.get(condition)
412 if prop_values:
413 var_counter += 1
414 values_filter = ", ".join(prop_values)
415 union_blocks.append(
416 f" {{ ?similar <{condition}>"
417 f" ?o_{var_counter} ."
418 f" FILTER(?o_{var_counter}"
419 f" IN ({values_filter})) }}"
420 )
421 elif isinstance(condition, dict) and "and" in condition:
422 block = _build_and_block(
423 condition["and"], subject_values_by_prop, subject_uri, var_counter
424 )
425 if block is not None:
426 text, var_counter = block
427 union_blocks.append(text)
428 else:
429 var_counter += len(condition["and"])
431 return union_blocks
434def _build_and_block(
435 and_props: list[str],
436 subject_values_by_prop: defaultdict[str, list[str]],
437 subject_uri: str,
438 var_counter: int,
439) -> tuple[str, int] | None:
440 if not all(p in subject_values_by_prop for p in and_props):
441 current_app.logger.debug(
442 "Skipping AND group %s because"
443 " subject %s lacks values for"
444 " all its properties.",
445 and_props,
446 subject_uri,
447 )
448 return None
450 and_patterns = []
451 for prop_uri in and_props:
452 prop_values = subject_values_by_prop[prop_uri]
453 var_counter += 1
454 values_filter = ", ".join(prop_values)
455 and_patterns.append(
456 f" ?similar <{prop_uri}>"
457 f" ?o_{var_counter} ."
458 f" FILTER(?o_{var_counter}"
459 f" IN ({values_filter})) ."
460 )
462 patterns_str = "\n".join(and_patterns)
463 return f" {{\n{patterns_str}\n }}", var_counter
466def _execute_similarity_query(
467 union_blocks: list[str],
468 entity_type: str,
469 subject_uri: str,
470 limit: int,
471 offset: int,
472) -> tuple[list[str], bool]:
473 sparql = get_sparql()
474 similarity_query_body = " UNION ".join(union_blocks)
476 query_limit = limit + 1
477 final_query = f"""
478 SELECT DISTINCT ?similar WHERE {{
479 ?similar a <{entity_type}> .
480 FILTER(?similar != <{subject_uri}>)
481 {{
482 {similarity_query_body}
483 }}
484 }} ORDER BY ?similar OFFSET {offset} LIMIT {query_limit}
485 """
487 sparql.setQuery(final_query)
488 sparql.setReturnFormat(JSON)
489 results = sparql.query().convert()
491 bindings = get_sparql_bindings(results)
492 candidate_uris = [item["similar"]["value"] for item in bindings]
494 has_more = len(candidate_uris) > limit
495 return candidate_uris[:limit], has_more
498def _transform_results(
499 uris: list[str],
500 entity_type: str,
501 shape_uri: str | None,
502) -> list[dict[str, str]]:
503 custom_filter = get_custom_filter()
504 transformed: list[dict[str, str]] = []
505 for uri in uris:
506 readable_label = (
507 custom_filter.human_readable_entity(uri, (entity_type, shape_uri))
508 if entity_type
509 else uri
510 )
511 transformed.append({"uri": uri, "label": readable_label or uri})
512 return transformed
515@merge_bp.route("/find_similar", methods=["GET"])
516@login_required
517def find_similar_resources() -> Response | tuple[Response, int]: # noqa: PLR0911
518 subject_uri = request.args.get("subject_uri")
519 entity_type = request.args.get("entity_type")
520 shape_uri = request.args.get("shape_uri")
521 try:
522 limit = int(request.args.get("limit", 5))
523 offset = int(request.args.get("offset", 0))
524 except ValueError:
525 return jsonify(
526 {"status": "error", "message": gettext("Invalid limit or offset parameter")}
527 ), 400
529 if not subject_uri or not entity_type:
530 return jsonify(
531 {
532 "status": "error",
533 "message": gettext(
534 "Missing required parameters (subject_uri, entity_type)"
535 ),
536 }
537 ), 400
539 if limit <= 0 or offset < 0:
540 return jsonify(
541 {
542 "status": "error",
543 "message": gettext("Limit must be positive and offset non-negative"),
544 }
545 ), 400
547 try:
548 entity_key = (entity_type, shape_uri)
549 similarity_config = get_similarity_properties(entity_key)
551 if not similarity_config or not isinstance(similarity_config, list):
552 return jsonify({"status": "success", "results": [], "has_more": False})
554 subject_values_by_prop = _fetch_subject_values(subject_uri, similarity_config)
555 if subject_values_by_prop is None:
556 return jsonify({"status": "success", "results": [], "has_more": False})
558 union_blocks = _build_union_blocks(
559 similarity_config, subject_values_by_prop, subject_uri
560 )
561 if not union_blocks:
562 return jsonify({"status": "success", "results": [], "has_more": False})
564 result_uris, has_more = _execute_similarity_query(
565 union_blocks, entity_type, subject_uri, limit, offset
566 )
567 transformed_results = _transform_results(result_uris, entity_type, shape_uri)
569 return jsonify(
570 {
571 "status": "success",
572 "results": transformed_results,
573 "has_more": has_more,
574 }
575 )
577 except Exception:
578 current_app.logger.exception(
579 "Error finding similar resources for %s", subject_uri
580 )
581 return jsonify(
582 {
583 "status": "error",
584 "message": gettext("An error occurred while finding similar resources"),
585 }
586 ), 500