Coverage for oc_meta / run / merge / check_merged_ids_results.py: 45%
247 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import argparse
6import csv
7import multiprocessing
8import os
9import re
10import zipfile
11from functools import partial
13import yaml
14from rdflib import RDF, Dataset, Literal, Namespace, URIRef
15from rich_argparse import RichHelpFormatter
16from oc_meta.lib.sparql import execute_sparql
17from tqdm import tqdm
19from oc_meta.core.editor import MetaEditor
20from oc_meta.lib.file_manager import find_rdf_file
21from oc_meta.run.merge.csv_utils import parse_merged_entities
23DATACITE = "http://purl.org/spar/datacite/"
24LITERAL_REIFICATION = "http://www.essepuntato.it/2010/06/literalreification/"
25PROV = Namespace("http://www.w3.org/ns/prov#")
28def read_csv(csv_file):
29 with open(csv_file, "r") as f:
30 reader = csv.DictReader(f)
31 return list(reader)
34def check_provenance(prov_file_path, entity_uri, is_surviving):
35 def extract_snapshot_number(snapshot_uri):
36 match = re.search(r"/prov/se/(\d+)$", str(snapshot_uri))
37 if match:
38 return int(match.group(1))
39 return 0 # Return 0 if no match found, this will put invalid URIs at the start
41 def is_merge_snapshot(g: Dataset, snapshot):
42 description = g.value(snapshot, URIRef("http://purl.org/dc/terms/description"))
43 if description:
44 # Check if the description indicates a merge operation
45 return "has been merged with" in str(description)
46 return False
48 try:
49 with zipfile.ZipFile(prov_file_path, "r") as zip_ref:
50 g = Dataset(default_union=True)
51 for filename in zip_ref.namelist():
52 with zip_ref.open(filename) as file:
53 g.parse(file, format="json-ld")
55 entity = URIRef(entity_uri)
56 snapshots = list(g.subjects(PROV.specializationOf, entity, unique=True))
58 if len(snapshots) <= 1:
59 tqdm.write(
60 f"Error in provenance file {prov_file_path}: Less than two snapshots found for entity {entity_uri}"
61 )
62 return
64 # Sort snapshots by their URI number
65 snapshots.sort(key=extract_snapshot_number)
67 for i, snapshot in enumerate(snapshots):
68 snapshot_number = extract_snapshot_number(snapshot)
69 if snapshot_number != i + 1:
70 tqdm.write(
71 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} has unexpected number {snapshot_number}, expected {i + 1}"
72 )
74 gen_time = g.value(snapshot, PROV.generatedAtTime)
75 if gen_time is None:
76 tqdm.write(
77 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} has no generation time"
78 )
80 if i < len(snapshots) - 1 or not is_surviving:
81 invalidation_time = g.value(snapshot, PROV.invalidatedAtTime)
82 if invalidation_time is None:
83 tqdm.write(
84 f"Error in provenance file {prov_file_path}: Non-last snapshot {snapshot} has no invalidation time"
85 )
86 elif (
87 is_surviving
88 and g.value(snapshot, PROV.invalidatedAtTime) is not None
89 ):
90 tqdm.write(
91 f"Error in provenance file {prov_file_path}: Last snapshot of surviving entity {snapshot} should not have invalidation time"
92 )
94 # Check prov:wasDerivedFrom
95 derived_from = list(
96 g.objects(snapshot, PROV.wasDerivedFrom, unique=True)
97 )
98 if i == 0: # First snapshot
99 if derived_from:
100 tqdm.write(
101 f"Error in provenance file {prov_file_path}: First snapshot {snapshot} should not have prov:wasDerivedFrom relation"
102 )
103 else: # All other snapshots
104 # Check if this is a merge snapshot
105 is_merge = is_merge_snapshot(g, snapshot)
107 if is_merge:
108 if len(derived_from) < 2:
109 tqdm.write(
110 f"Error in provenance file {prov_file_path}: Merge snapshot {snapshot} should be derived from more than one snapshot"
111 )
112 else:
113 if len(derived_from) != 1:
114 tqdm.write(
115 f"Error in provenance file {prov_file_path}: Non-merge snapshot {snapshot} should have exactly one prov:wasDerivedFrom relation, but has {len(derived_from)}"
116 )
117 elif derived_from[0] != snapshots[i - 1]:
118 tqdm.write(
119 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} is not derived from the previous snapshot"
120 )
122 if not is_surviving:
123 # Check if the last snapshot is invalidated for merged entities
124 last_snapshot = snapshots[-1]
125 if g.value(last_snapshot, PROV.invalidatedAtTime) is None:
126 tqdm.write(
127 f"Error in provenance file {prov_file_path}: Last snapshot {last_snapshot} of merged entity {entity_uri} is not invalidated"
128 )
130 except FileNotFoundError:
131 tqdm.write(f"Error: Provenance file not found for entity {entity_uri}")
132 except zipfile.BadZipFile:
133 tqdm.write(f"Error: Invalid zip file for provenance of entity {entity_uri}")
136def check_identifier_constraints(g: Dataset, entity):
137 issues = []
139 types = list(g.objects(entity, RDF.type, unique=True))
140 if not types:
141 issues.append(f"Entity {entity} has no type")
142 elif len(types) > 1:
143 issues.append(f"Entity {entity} has multiple types")
144 elif URIRef(DATACITE + "Identifier") not in types:
145 issues.append(f"Entity {entity} is not a datacite:Identifier")
147 identifier_scheme = list(
148 g.objects(entity, URIRef(DATACITE + "usesIdentifierScheme"), unique=True)
149 )
150 literal_value = list(
151 g.objects(entity, URIRef(LITERAL_REIFICATION + "hasLiteralValue"), unique=True)
152 )
154 if len(identifier_scheme) != 1:
155 issues.append(
156 f"Entity {entity} should have exactly one usesIdentifierScheme, found {len(identifier_scheme)}"
157 )
158 elif not isinstance(identifier_scheme[0], URIRef):
159 issues.append(
160 f"Entity {entity}'s usesIdentifierScheme should be a URIRef, found {type(identifier_scheme[0])}"
161 )
163 if len(literal_value) != 1:
164 issues.append(
165 f"Entity {entity} should have exactly one hasLiteralValue, found {len(literal_value)}"
166 )
167 elif not isinstance(literal_value[0], Literal):
168 issues.append(
169 f"Entity {entity}'s hasLiteralValue should be a Literal, found {type(literal_value[0])}"
170 )
172 return issues
175def check_entity_file(file_path, entity_uri, is_surviving):
176 try:
177 with zipfile.ZipFile(file_path, "r") as zip_ref:
178 g = Dataset(default_union=True)
179 for filename in zip_ref.namelist():
180 with zip_ref.open(filename) as file:
181 g.parse(file, format="json-ld")
182 except FileNotFoundError:
183 tqdm.write(f"Error: File not found for entity {entity_uri}")
184 return
185 except zipfile.BadZipFile:
186 tqdm.write(f"Error: Invalid zip file for entity {entity_uri}")
187 return
189 entity = URIRef(entity_uri)
190 if (entity, None, None) not in g:
191 if is_surviving:
192 tqdm.write(
193 f"Error in file {file_path}: Surviving entity {entity_uri} does not exist"
194 )
195 elif not is_surviving:
196 tqdm.write(
197 f"Error in file {file_path}: Merged entity {entity_uri} still exists"
198 )
199 else:
200 for issue in check_identifier_constraints(g, entity):
201 tqdm.write(f"Error in file {file_path}: {issue}")
203 # Check provenance
204 prov_file_path = file_path.replace(".zip", "") + "/prov/se.zip"
205 check_provenance(prov_file_path, entity_uri, is_surviving)
208def check_entity_sparql(endpoint: str, entity_uri, is_surviving):
209 has_issues = False
211 exists_query = f"""
212 ASK {{
213 <{entity_uri}> ?p ?o .
214 }}
215 """
216 exists_results = execute_sparql(
217 endpoint, exists_query, max_retries=3, backoff_factor=1
218 )
220 exists = exists_results["boolean"]
221 if exists:
222 if not is_surviving:
223 tqdm.write(f"Error in SPARQL: Merged entity {entity_uri} still exists")
224 has_issues = True
225 elif is_surviving:
226 tqdm.write(f"Error in SPARQL: Surviving entity {entity_uri} does not exist")
227 has_issues = True
228 return has_issues
230 if not is_surviving:
231 referenced_query = f"""
232 ASK {{
233 ?s ?p <{entity_uri}> .
234 }}
235 """
236 referenced_results = execute_sparql(
237 endpoint, referenced_query, max_retries=3, backoff_factor=1
238 )
240 if referenced_results["boolean"]:
241 tqdm.write(
242 f"Error in SPARQL: Merged entity {entity_uri} is still referenced by other entities"
243 )
244 has_issues = True
246 if not exists:
247 return has_issues
249 if not exists:
250 return has_issues
252 types_query = f"""
253 SELECT ?type WHERE {{
254 <{entity_uri}> a ?type .
255 }}
256 """
257 types_results = execute_sparql(
258 endpoint, types_query, max_retries=3, backoff_factor=1
259 )
261 types = [result["type"]["value"] for result in types_results["results"]["bindings"]]
262 if not types:
263 tqdm.write(f"Error in SPARQL: Entity {entity_uri} has no type")
264 has_issues = True
265 elif len(types) > 1:
266 tqdm.write(f"Error in SPARQL: Entity {entity_uri} has multiple types")
267 has_issues = True
268 elif DATACITE + "Identifier" not in types:
269 tqdm.write(f"Error in SPARQL: Entity {entity_uri} is not a datacite:Identifier")
270 has_issues = True
272 identifier_query = f"""
273 SELECT DISTINCT ?scheme ?value WHERE {{
274 OPTIONAL {{ <{entity_uri}> <{DATACITE}usesIdentifierScheme> ?scheme }}
275 OPTIONAL {{ <{entity_uri}> <{LITERAL_REIFICATION}hasLiteralValue> ?value }}
276 }}
277 """
278 identifier_results = execute_sparql(
279 endpoint, identifier_query, max_retries=3, backoff_factor=1
280 )
282 schemes = sorted(
283 {
284 result["scheme"]["value"]
285 for result in identifier_results["results"]["bindings"]
286 if "scheme" in result
287 }
288 )
289 values = sorted(
290 {
291 result["value"]["value"]
292 for result in identifier_results["results"]["bindings"]
293 if "value" in result
294 }
295 )
297 if len(schemes) != 1:
298 tqdm.write(
299 f"Error in SPARQL: Entity {entity_uri} should have exactly one usesIdentifierScheme, found {len(schemes)}"
300 )
301 has_issues = True
302 elif not schemes[0].startswith("http"):
303 tqdm.write(
304 f"Error in SPARQL: Entity {entity_uri}'s usesIdentifierScheme should be a URIRef, found {schemes[0]}"
305 )
306 has_issues = True
308 if len(values) != 1:
309 tqdm.write(
310 f"Error in SPARQL: Entity {entity_uri} should have exactly one hasLiteralValue, found {len(values)}"
311 )
312 has_issues = True
314 return has_issues
317def process_csv(args, csv_file):
318 csv_path, rdf_dir, meta_config_path, sparql_endpoint, query_output_dir = args
319 csv_path = os.path.join(csv_path, csv_file)
320 data = read_csv(csv_path)
321 tasks = []
323 meta_editor = MetaEditor(meta_config_path, "")
325 for row in data:
326 if "Done" not in row or row["Done"] != "True":
327 continue
329 surviving_entity = row["surviving_entity"]
330 merged_entities = parse_merged_entities(row["merged_entities"])
331 all_entities = [surviving_entity] + merged_entities
333 for entity in all_entities:
334 file_path = find_rdf_file(
335 entity,
336 rdf_dir,
337 meta_editor.dir_split,
338 meta_editor.n_file_item,
339 zip_output=True,
340 )
341 tasks.append(
342 (
343 entity,
344 entity == surviving_entity,
345 file_path,
346 rdf_dir,
347 meta_config_path,
348 sparql_endpoint,
349 query_output_dir,
350 surviving_entity,
351 )
352 )
354 return tasks
357def get_entity_triples(endpoint: str, entity_uri):
358 query = f"""
359 SELECT ?g ?s ?p ?o
360 WHERE {{
361 GRAPH ?g {{
362 {{
363 <{entity_uri}> ?p ?o .
364 BIND(<{entity_uri}> AS ?s)
365 }}
366 UNION
367 {{
368 ?s ?p <{entity_uri}> .
369 BIND(<{entity_uri}> AS ?o)
370 }}
371 }}
372 }}
373 """
374 results = execute_sparql(endpoint, query, max_retries=3, backoff_factor=1)
376 triples = []
377 for result in results["results"]["bindings"]:
378 graph = result["g"]["value"]
379 subject = URIRef(result["s"]["value"])
380 predicate = URIRef(result["p"]["value"])
382 obj_data = result["o"]
383 if obj_data["type"] == "uri":
384 obj = URIRef(obj_data["value"])
385 else:
386 datatype = obj_data.get("datatype")
387 obj = Literal(
388 obj_data["value"], datatype=URIRef(datatype) if datatype else None
389 )
391 triples.append((graph, subject, predicate, obj))
393 return triples
396def generate_update_query(merged_entity, surviving_entity, triples):
397 delete_query = "DELETE DATA {\n"
398 insert_query = "INSERT DATA {\n"
400 for graph, subject, predicate, obj in triples:
401 if subject == URIRef(merged_entity):
402 delete_query += (
403 f" GRAPH <{graph}> {{ <{subject}> <{predicate}> {obj.n3()} }}\n"
404 )
405 elif obj == URIRef(merged_entity):
406 delete_query += (
407 f" GRAPH <{graph}> {{ <{subject}> <{predicate}> <{obj}> }}\n"
408 )
409 insert_query += f" GRAPH <{graph}> {{ <{subject}> <{predicate}> <{surviving_entity}> }}\n"
411 delete_query += "}\n"
412 insert_query += "}\n"
414 combined_query = delete_query + "\n" + insert_query
415 return combined_query
418def process_entity(args):
419 (
420 entity,
421 is_surviving,
422 file_path,
423 rdf_dir,
424 meta_config_path,
425 sparql_endpoint,
426 query_output_dir,
427 surviving_entity,
428 ) = args
430 if file_path is None:
431 tqdm.write(f"Error: Could not find file for entity {entity}")
432 else:
433 check_entity_file(file_path, entity, is_surviving)
435 has_issues = check_entity_sparql(sparql_endpoint, entity, is_surviving)
437 if has_issues and not is_surviving:
438 triples = get_entity_triples(sparql_endpoint, entity)
439 combined_query = generate_update_query(entity, surviving_entity, triples)
441 query_file_path = os.path.join(
442 query_output_dir, f"update_{entity.split('/')[-1]}.sparql"
443 )
444 with open(query_file_path, "w") as f:
445 f.write(combined_query)
448def main():
449 parser = argparse.ArgumentParser(
450 description="Check merge process success on files and SPARQL endpoint",
451 formatter_class=RichHelpFormatter,
452 )
453 parser.add_argument(
454 "csv_folder", type=str, help="Path to the folder containing CSV files"
455 )
456 parser.add_argument("rdf_dir", type=str, help="Path to the RDF directory")
457 parser.add_argument(
458 "--meta_config", type=str, required=True, help="Path to meta configuration file"
459 )
460 parser.add_argument(
461 "--query_output",
462 type=str,
463 required=True,
464 help="Path to the folder where SPARQL queries will be saved",
465 )
466 args = parser.parse_args()
468 with open(args.meta_config, "r") as config_file:
469 config = yaml.safe_load(config_file)
471 sparql_endpoint = config["triplestore_url"]
473 os.makedirs(args.query_output, exist_ok=True)
475 csv_files = [f for f in os.listdir(args.csv_folder) if f.endswith(".csv")]
477 # Use forkserver to avoid deadlocks when forking in a multi-threaded environment
478 ctx = multiprocessing.get_context("forkserver")
480 # Process CSV files in parallel
481 with ctx.Pool(processes=multiprocessing.cpu_count()) as pool:
482 process_csv_partial = partial(
483 process_csv,
484 (
485 args.csv_folder,
486 args.rdf_dir,
487 args.meta_config,
488 sparql_endpoint,
489 args.query_output,
490 ),
491 )
492 all_tasks = list(
493 tqdm(
494 pool.imap(process_csv_partial, csv_files),
495 total=len(csv_files),
496 desc="Processing CSV files",
497 )
498 )
500 # Flatten the list of lists into a single list
501 all_tasks = [task for sublist in all_tasks for task in sublist]
503 # Process entities in parallel
504 with ctx.Pool(processes=multiprocessing.cpu_count()) as pool:
505 list(
506 tqdm(
507 pool.imap(process_entity, all_tasks),
508 total=len(all_tasks),
509 desc="Processing entities",
510 )
511 )
514if __name__ == "__main__":
515 main()