Coverage for oc_meta / run / merge / check_merged_brs_results.py: 30%
304 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 os
8import re
9import zipfile
10from functools import partial
11import multiprocessing
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.check_utils import has_next_chain_issues
22from oc_meta.run.merge.csv_utils import parse_merged_entities
24DATACITE = "http://purl.org/spar/datacite/"
25FABIO = "http://purl.org/spar/fabio/"
26PROV = Namespace("http://www.w3.org/ns/prov#")
27PRO = Namespace("http://purl.org/spar/pro/")
28DCTERMS = Namespace("http://purl.org/dc/terms/")
29FRBR = Namespace("http://purl.org/vocab/frbr/core#")
30PRISM = Namespace("http://prismstandard.org/namespaces/basic/2.0/")
33def read_csv(csv_file):
34 with open(csv_file, "r") as f:
35 reader = csv.DictReader(f)
36 return list(reader)
39def check_br_constraints(g: Dataset, entity):
40 issues = []
42 # Check types
43 types = list(g.objects(entity, RDF.type, unique=True))
44 if not types:
45 issues.append(f"Entity {entity} has no type")
46 elif len(types) > 2:
47 issues.append(f"Entity {entity} has more than two types")
48 elif URIRef(FABIO + "Expression") not in types:
49 issues.append(f"Entity {entity} is not a fabio:Expression")
51 # Check if entity is a journal issue or volume
52 is_journal_issue = URIRef(FABIO + "JournalIssue") in types
53 is_journal_volume = URIRef(FABIO + "JournalVolume") in types
55 # Check identifiers
56 identifiers = list(
57 g.objects(entity, URIRef(DATACITE + "hasIdentifier"), unique=True)
58 )
59 if not identifiers:
60 issues.append(f"Entity {entity} has no datacite:hasIdentifier")
62 # Check title (zero or one)
63 titles = list(g.objects(entity, DCTERMS.title, unique=True))
64 if len(titles) > 1:
65 issues.append(f"Entity {entity} has multiple titles")
67 # Check part of (zero or one)
68 part_of = list(g.objects(entity, FRBR.partOf, unique=True))
69 if len(part_of) > 1:
70 issues.append(f"Entity {entity} has multiple partOf relations")
72 # Check publication date (zero or one)
73 pub_dates = list(g.objects(entity, PRISM.publicationDate, unique=True))
74 if len(pub_dates) > 1:
75 issues.append(f"Entity {entity} has multiple publication dates")
77 # Check sequence identifier (zero or one)
78 seq_ids = list(
79 g.objects(entity, URIRef(FABIO + "hasSequenceIdentifier"), unique=True)
80 )
81 if len(seq_ids) > 1:
82 issues.append(f"Entity {entity} has multiple sequence identifiers")
83 elif seq_ids and not (is_journal_issue or is_journal_volume):
84 issues.append(
85 f"Entity {entity} has sequence identifier but is not a journal issue or volume"
86 )
88 return issues
91def check_entity_sparql(endpoint: str, entity_uri, is_surviving):
92 has_issues = False
94 exists_query = f"""
95 ASK {{
96 <{entity_uri}> ?p ?o .
97 }}
98 """
99 exists_results = execute_sparql(
100 endpoint, exists_query, max_retries=3, backoff_factor=1
101 )
103 exists = exists_results["boolean"]
104 if exists:
105 if not is_surviving:
106 tqdm.write(f"Error in SPARQL: Merged entity {entity_uri} still exists")
107 has_issues = True
108 elif is_surviving:
109 tqdm.write(f"Error in SPARQL: Surviving entity {entity_uri} does not exist")
110 has_issues = True
111 return has_issues
113 if not is_surviving:
114 referenced_query = f"""
115 ASK {{
116 ?s ?p <{entity_uri}> .
117 }}
118 """
119 referenced_results = execute_sparql(
120 endpoint, referenced_query, max_retries=3, backoff_factor=1
121 )
123 if referenced_results["boolean"]:
124 tqdm.write(
125 f"Error in SPARQL: Merged entity {entity_uri} is still referenced by other entities"
126 )
127 has_issues = True
129 if not exists:
130 return has_issues
132 if not exists:
133 return has_issues
135 types_query = f"""
136 SELECT ?type WHERE {{
137 <{entity_uri}> a ?type .
138 }}
139 """
140 types_results = execute_sparql(
141 endpoint, types_query, max_retries=3, backoff_factor=1
142 )
144 types = [result["type"]["value"] for result in types_results["results"]["bindings"]]
145 if not types:
146 tqdm.write(f"Error in SPARQL: Entity {entity_uri} has no type")
147 has_issues = True
148 elif len(types) > 2:
149 tqdm.write(f"Error in SPARQL: Entity {entity_uri} has more than two types")
150 has_issues = True
151 elif FABIO + "Expression" not in types:
152 tqdm.write(f"Error in SPARQL: Entity {entity_uri} is not a fabio:Expression")
153 has_issues = True
155 identifiers_query = f"""
156 SELECT ?identifier WHERE {{
157 <{entity_uri}> <{DATACITE}hasIdentifier> ?identifier .
158 }}
159 """
160 identifiers_results = execute_sparql(
161 endpoint, identifiers_query, max_retries=3, backoff_factor=1
162 )
164 identifiers = [
165 result["identifier"]["value"]
166 for result in identifiers_results["results"]["bindings"]
167 ]
168 if not identifiers:
169 tqdm.write(
170 f"Error in SPARQL: Entity {entity_uri} has no datacite:hasIdentifier"
171 )
172 has_issues = True
174 if is_surviving and check_duplicate_contributor_roles(endpoint, entity_uri):
175 has_issues = True
177 if is_surviving and check_duplicate_identifiers(endpoint, entity_uri):
178 has_issues = True
180 if is_surviving and check_has_next_integrity(endpoint, entity_uri):
181 has_issues = True
183 return has_issues
186def check_duplicate_identifiers(endpoint: str, entity_uri):
187 query = f"""
188 PREFIX datacite: <{DATACITE}>
189 PREFIX literal: <http://www.essepuntato.it/2010/06/literalreification/>
190 SELECT ?scheme ?value (COUNT(DISTINCT ?identifier) AS ?count) WHERE {{
191 <{entity_uri}> datacite:hasIdentifier ?identifier .
192 ?identifier datacite:usesIdentifierScheme ?scheme .
193 ?identifier literal:hasLiteralValue ?value .
194 }}
195 GROUP BY ?scheme ?value
196 HAVING (COUNT(DISTINCT ?identifier) > 1)
197 """
198 results = execute_sparql(endpoint, query, max_retries=3, backoff_factor=1)
200 has_issues = False
201 for result in results["results"]["bindings"]:
202 has_issues = True
203 scheme = result["scheme"]["value"].split("/")[-1]
204 tqdm.write(
205 f"Error in SPARQL: Entity {entity_uri} has {result['count']['value']} "
206 f"duplicate identifiers with scheme {scheme} and value {result['value']['value']}"
207 )
208 return has_issues
211def check_duplicate_contributor_roles(endpoint: str, entity_uri):
212 query = f"""
213 PREFIX pro: <{PRO}>
214 PREFIX frbr: <{FRBR}>
215 SELECT ?container ?roleType ?agent (COUNT(DISTINCT ?ar) AS ?count) WHERE {{
216 {{ BIND(<{entity_uri}> AS ?container) }}
217 UNION
218 {{ <{entity_uri}> frbr:partOf+ ?container }}
219 ?container pro:isDocumentContextFor ?ar .
220 ?ar pro:withRole ?roleType .
221 ?ar pro:isHeldBy ?agent .
222 }}
223 GROUP BY ?container ?roleType ?agent
224 HAVING (COUNT(DISTINCT ?ar) > 1)
225 """
226 results = execute_sparql(endpoint, query, max_retries=3, backoff_factor=1)
228 has_issues = False
229 for result in results["results"]["bindings"]:
230 has_issues = True
231 role_type = result["roleType"]["value"].split("/")[-1]
232 tqdm.write(
233 f"Error in SPARQL: Entity {result['container']['value']} has "
234 f"{result['count']['value']} duplicate {role_type} roles held by "
235 f"{result['agent']['value']}"
236 )
237 return has_issues
240def check_has_next_integrity(endpoint: str, entity_uri):
241 issues = has_next_chain_issues(endpoint, f"BIND(<{entity_uri}> AS ?br)")
242 for issue in issues:
243 tqdm.write(f"Error in SPARQL: {issue}")
244 return bool(issues)
247def get_entity_triples(endpoint: str, entity_uri):
248 query = f"""
249 SELECT ?g ?s ?p ?o
250 WHERE {{
251 GRAPH ?g {{
252 {{
253 <{entity_uri}> ?p ?o .
254 BIND(<{entity_uri}> AS ?s)
255 }}
256 UNION
257 {{
258 ?s ?p <{entity_uri}> .
259 BIND(<{entity_uri}> AS ?o)
260 }}
261 }}
262 }}
263 """
264 results = execute_sparql(endpoint, query, max_retries=3, backoff_factor=1)
266 triples = []
267 for result in results["results"]["bindings"]:
268 graph = result["g"]["value"]
269 subject = URIRef(result["s"]["value"])
270 predicate = URIRef(result["p"]["value"])
272 obj_data = result["o"]
273 if obj_data["type"] == "uri":
274 obj = URIRef(obj_data["value"])
275 else:
276 datatype = obj_data.get("datatype")
277 obj = Literal(
278 obj_data["value"], datatype=URIRef(datatype) if datatype else None
279 )
281 triples.append((graph, subject, predicate, obj))
283 return triples
286def generate_update_query(merged_entity, surviving_entity, triples):
287 delete_query = "DELETE DATA {\n"
288 insert_query = "INSERT DATA {\n"
290 for graph, subject, predicate, obj in triples:
291 if subject == URIRef(merged_entity):
292 delete_query += (
293 f" GRAPH <{graph}> {{ <{subject}> <{predicate}> {obj.n3()} }}\n"
294 )
295 elif obj == URIRef(merged_entity):
296 delete_query += (
297 f" GRAPH <{graph}> {{ <{subject}> <{predicate}> <{obj}> }}\n"
298 )
299 insert_query += f" GRAPH <{graph}> {{ <{subject}> <{predicate}> <{surviving_entity}> }}\n"
301 delete_query += "}\n"
302 insert_query += "}\n"
304 combined_query = delete_query + "\n" + insert_query
305 return combined_query
308def process_csv(args, csv_file):
309 csv_path, rdf_dir, meta_config_path, sparql_endpoint, query_output_dir = args
310 csv_path = os.path.join(csv_path, csv_file)
311 data = read_csv(csv_path)
312 tasks = []
314 meta_editor = MetaEditor(meta_config_path, "")
316 for row in data:
317 if "Done" not in row or row["Done"] != "True":
318 continue
320 surviving_entity = row["surviving_entity"]
321 merged_entities = parse_merged_entities(row["merged_entities"])
322 all_entities = [surviving_entity] + merged_entities
324 for entity in all_entities:
325 file_path = find_rdf_file(
326 entity,
327 rdf_dir,
328 meta_editor.dir_split,
329 meta_editor.n_file_item,
330 zip_output=True,
331 )
332 tasks.append(
333 (
334 entity,
335 entity == surviving_entity,
336 file_path,
337 rdf_dir,
338 meta_config_path,
339 sparql_endpoint,
340 query_output_dir,
341 surviving_entity,
342 )
343 )
345 return tasks
348def process_file_group(args):
349 file_path, entities, sparql_endpoint, query_output_dir = args
351 if file_path is None:
352 for entity, is_surviving, surviving_entity in entities:
353 tqdm.write(f"Error: Could not find file for entity {entity}")
354 has_issues = check_entity_sparql(sparql_endpoint, entity, is_surviving)
355 if has_issues and not is_surviving:
356 triples = get_entity_triples(sparql_endpoint, entity)
357 combined_query = generate_update_query(
358 entity, surviving_entity, triples
359 )
360 query_file_path = os.path.join(
361 query_output_dir, f"update_{entity.split('/')[-1]}.sparql"
362 )
363 with open(query_file_path, "w") as f:
364 f.write(combined_query)
365 return
367 try:
368 with zipfile.ZipFile(file_path, "r") as zip_ref:
369 g = Dataset(default_union=True)
370 for filename in zip_ref.namelist():
371 with zip_ref.open(filename) as file:
372 g.parse(file, format="json-ld")
373 except FileNotFoundError:
374 for entity, is_surviving, surviving_entity in entities:
375 tqdm.write(f"Error: File not found for entity {entity}")
376 has_issues = check_entity_sparql(sparql_endpoint, entity, is_surviving)
377 if has_issues and not is_surviving:
378 triples = get_entity_triples(sparql_endpoint, entity)
379 combined_query = generate_update_query(
380 entity, surviving_entity, triples
381 )
382 query_file_path = os.path.join(
383 query_output_dir, f"update_{entity.split('/')[-1]}.sparql"
384 )
385 with open(query_file_path, "w") as f:
386 f.write(combined_query)
387 return
389 prov_file_path = file_path.replace(".zip", "") + "/prov/se.zip"
390 prov_graph = None
391 try:
392 with zipfile.ZipFile(prov_file_path, "r") as zip_ref:
393 prov_graph = Dataset(default_union=True)
394 for filename in zip_ref.namelist():
395 with zip_ref.open(filename) as file:
396 prov_graph.parse(file, format="json-ld")
397 except FileNotFoundError:
398 prov_graph = None
399 except zipfile.BadZipFile:
400 prov_graph = "badzip"
402 for entity, is_surviving, surviving_entity in entities:
403 if (URIRef(entity), None, None) not in g:
404 if is_surviving:
405 tqdm.write(
406 f"Error in file {file_path}: Surviving entity {entity} does not exist"
407 )
408 else:
409 if not is_surviving:
410 tqdm.write(
411 f"Error in file {file_path}: Merged entity {entity} still exists"
412 )
413 else:
414 br_issues = check_br_constraints(g, URIRef(entity))
415 for issue in br_issues:
416 tqdm.write(f"Error in file {file_path}: {issue}")
418 if prov_graph is None:
419 tqdm.write(f"Error: Provenance file not found for entity {entity}")
420 elif prov_graph == "badzip":
421 tqdm.write(f"Error: Invalid zip file for provenance of entity {entity}")
422 else:
423 check_entity_provenance(
424 URIRef(entity), is_surviving, prov_graph, prov_file_path
425 )
427 has_issues = check_entity_sparql(sparql_endpoint, entity, is_surviving)
429 if has_issues and not is_surviving:
430 triples = get_entity_triples(sparql_endpoint, entity)
431 combined_query = generate_update_query(entity, surviving_entity, triples)
432 query_file_path = os.path.join(
433 query_output_dir, f"update_{entity.split('/')[-1]}.sparql"
434 )
435 with open(query_file_path, "w") as f:
436 f.write(combined_query)
439def check_entity_provenance(entity_uri, is_surviving, prov_graph, prov_file_path):
440 def extract_snapshot_number(snapshot_uri):
441 match = re.search(r"/prov/se/(\d+)$", str(snapshot_uri))
442 if match:
443 return int(match.group(1))
444 return 0
446 snapshots = list(
447 prov_graph.subjects(PROV.specializationOf, entity_uri, unique=True)
448 )
449 if len(snapshots) <= 1:
450 tqdm.write(
451 f"Error in provenance file {prov_file_path}: Less than two snapshots found for entity {entity_uri}"
452 )
453 return
455 snapshots.sort(key=extract_snapshot_number)
456 for i, snapshot in enumerate(snapshots):
457 snapshot_number = extract_snapshot_number(snapshot)
458 if snapshot_number != i + 1:
459 tqdm.write(
460 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} has unexpected number {snapshot_number}, expected {i + 1}"
461 )
463 gen_time = prov_graph.value(snapshot, PROV.generatedAtTime)
464 if gen_time is None:
465 tqdm.write(
466 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} has no generation time"
467 )
469 if i < len(snapshots) - 1 or not is_surviving:
470 invalidation_time = prov_graph.value(snapshot, PROV.invalidatedAtTime)
471 if invalidation_time is None:
472 tqdm.write(
473 f"Error in provenance file {prov_file_path}: Non-last snapshot {snapshot} has no invalidation time"
474 )
475 elif (
476 is_surviving
477 and prov_graph.value(snapshot, PROV.invalidatedAtTime) is not None
478 ):
479 tqdm.write(
480 f"Error in provenance file {prov_file_path}: Last snapshot of surviving entity {snapshot} should not have invalidation time"
481 )
483 description = prov_graph.value(snapshot, DCTERMS.description)
484 is_merge_snapshot = description and "has been merged with" in str(description)
486 derived_from = list(prov_graph.objects(snapshot, PROV.wasDerivedFrom))
487 if i == 0: # First snapshot
488 if derived_from:
489 tqdm.write(
490 f"Error in provenance file {prov_file_path}: First snapshot {snapshot} should not have prov:wasDerivedFrom relation"
491 )
492 elif is_merge_snapshot:
493 if len(derived_from) < 2:
494 tqdm.write(
495 f"Error in provenance file {prov_file_path}: Merge snapshot {snapshot} should be derived from at least two snapshots"
496 )
497 else: # Regular modification snapshot
498 if len(derived_from) != 1:
499 tqdm.write(
500 f"Error in provenance file {prov_file_path}: Regular modification snapshot {snapshot} should have exactly one prov:wasDerivedFrom relation"
501 )
502 else:
503 previous_snapshot = snapshots[i - 1]
504 if derived_from[0] != previous_snapshot:
505 tqdm.write(
506 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} is not derived from the previous snapshot"
507 )
509 if not is_surviving:
510 # Check if the last snapshot is invalidated for merged entities
511 last_snapshot = snapshots[-1]
512 invalidation = list(prov_graph.objects(last_snapshot, PROV.invalidatedAtTime))
513 if not invalidation:
514 tqdm.write(
515 f"Error in provenance file {prov_file_path}: Last snapshot {last_snapshot} of merged entity {entity_uri} is not invalidated"
516 )
519def main():
520 parser = argparse.ArgumentParser(
521 description="Check merge process success on files and SPARQL endpoint for bibliographic resources",
522 formatter_class=RichHelpFormatter,
523 )
524 parser.add_argument(
525 "csv_folder", type=str, help="Path to the folder containing CSV files"
526 )
527 parser.add_argument("rdf_dir", type=str, help="Path to the RDF directory")
528 parser.add_argument(
529 "--meta_config", type=str, required=True, help="Path to meta configuration file"
530 )
531 parser.add_argument(
532 "--query_output",
533 type=str,
534 required=True,
535 help="Path to the folder where SPARQL queries will be saved",
536 )
537 args = parser.parse_args()
539 with open(args.meta_config, "r") as config_file:
540 config = yaml.safe_load(config_file)
542 sparql_endpoint = config["triplestore_url"]
544 os.makedirs(args.query_output, exist_ok=True)
546 csv_files = [f for f in os.listdir(args.csv_folder) if f.endswith(".csv")]
548 # Use forkserver to avoid deadlocks when forking in a multi-threaded environment
549 ctx = multiprocessing.get_context("forkserver")
551 # Process CSV files to gather tasks
552 with ctx.Pool(processes=multiprocessing.cpu_count()) as pool:
553 process_csv_partial = partial(
554 process_csv,
555 (
556 args.csv_folder,
557 args.rdf_dir,
558 args.meta_config,
559 sparql_endpoint,
560 args.query_output,
561 ),
562 )
563 all_tasks_list = list(
564 tqdm(
565 pool.imap(process_csv_partial, csv_files),
566 total=len(csv_files),
567 desc="Processing CSV files",
568 )
569 )
571 # Flatten the list of lists into a single list
572 all_tasks = [task for sublist in all_tasks_list for task in sublist]
574 # Now we have a list of tasks (entity, is_surviving, file_path, rdf_dir, meta_config_path, sparql_endpoint, query_output_dir, surviving_entity)
575 # We want to group by file_path to open each file only once
576 tasks_by_file = {}
577 for (
578 entity,
579 is_surviving,
580 file_path,
581 rdf_dir,
582 meta_config_path,
583 sparql_endpoint,
584 query_output_dir,
585 surviving_entity,
586 ) in all_tasks:
587 if file_path not in tasks_by_file:
588 tasks_by_file[file_path] = []
589 tasks_by_file[file_path].append((entity, is_surviving, surviving_entity))
591 # Convert to a list of arguments for parallel processing
592 file_groups = [
593 (file_path, tasks, sparql_endpoint, args.query_output)
594 for file_path, tasks in tasks_by_file.items()
595 ]
597 # Process each file group in parallel
598 with ctx.Pool(processes=multiprocessing.cpu_count()) as pool:
599 list(
600 tqdm(
601 pool.imap(process_file_group, file_groups),
602 total=len(file_groups),
603 desc="Processing files",
604 )
605 )
608if __name__ == "__main__":
609 main()