Coverage for oc_meta/run/merge/check_merged_ras

1# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>

3# SPDX-License-Identifier: ISC

5import argparse

6import csv

7import multiprocessing

8import os

9import re

10import zipfile

11from functools import partial

13import filelock

14import yaml

15from rdflib import RDF, Dataset, Literal, Namespace, URIRef

16from rich_argparse import RichHelpFormatter

17from tqdm import tqdm

19from oc_meta.core.editor import MetaEditor

20from oc_meta.lib.file_manager import find_rdf_file

21from oc_meta.lib.sparql import execute_sparql

22from oc_meta.run.merge.check_utils import has_next_chain_issues

23from oc_meta.run.merge.csv_utils import parse_merged_entities

25DATACITE = "http://purl.org/spar/datacite/"

26FOAF = Namespace("http://xmlns.com/foaf/0.1/")

27PROV = Namespace("http://www.w3.org/ns/prov#")

28PRO = Namespace("http://purl.org/spar/pro/")

29DCTERMS = Namespace("http://purl.org/dc/terms/")

32def read_csv(csv_file):

33 with open(csv_file, "r") as f:

34 reader = csv.DictReader(f)

35 return list(reader)

38def check_agent_constraints(g: Dataset, entity):

39 issues = []

41 # Check type (must be exactly one: foaf:Agent)

42 types = list(g.objects(entity, RDF.type, unique=True))

43 if not types:

44 issues.append(f"Entity {entity} has no type")

45 elif len(types) > 1:

46 issues.append(f"Entity {entity} has multiple types")

47 elif URIRef(FOAF + "Agent") not in types:

48 issues.append(f"Entity {entity} is not a foaf:Agent")

50 # Check identifiers

51 identifiers = list(

52 g.objects(entity, URIRef(DATACITE + "hasIdentifier"), unique=True)

53 )

54 if not identifiers:

55 issues.append(f"Entity {entity} has no datacite:hasIdentifier")

57 # Check name properties (must have at least one)

58 has_name = False

59 names = list(g.objects(entity, FOAF.name, unique=True))

60 given_names = list(g.objects(entity, FOAF.givenName, unique=True))

61 family_names = list(g.objects(entity, FOAF.familyName, unique=True))

63 if names or given_names or family_names:

64 has_name = True

66 if not has_name:

67 issues.append(

68 f"Entity {entity} has no name properties (name, givenName, or familyName)"

69 )

71 return issues

74def check_has_next_integrity(endpoint: str, entity_uri):

75 # The roles the agent holds live in other resources' files, so the

76 # oco:hasNext chains touched by an agent merge are only visible to SPARQL.

77 issues = has_next_chain_issues(

78 endpoint,

79 f"?held_role pro:isHeldBy <{entity_uri}> . ?br pro:isDocumentContextFor ?held_role .",

80 )

81 for issue in issues:

82 tqdm.write(f"Error in SPARQL: {issue}")

83 return bool(issues)

86def check_entity_sparql(endpoint: str, entity_uri, is_surviving):

87 has_issues = False

89 exists_query = f"""

90 ASK {{

91 <{entity_uri}> ?p ?o .

92 }}

93 """

94 exists_results = execute_sparql(

95 endpoint, exists_query, max_retries=3, backoff_factor=1

96 )

98 exists = exists_results["boolean"]

99 if exists:

100 if not is_surviving:

101 tqdm.write(f"Error in SPARQL: Merged entity {entity_uri} still exists")

102 has_issues = True

103 elif is_surviving:

104 tqdm.write(f"Error in SPARQL: Surviving entity {entity_uri} does not exist")

105 return True

106

107 if not is_surviving:

108 referenced_query = f"""

109 ASK {{

110 ?s ?p <{entity_uri}> .

111 }}

112 """

113 referenced_results = execute_sparql(

114 endpoint, referenced_query, max_retries=3, backoff_factor=1

115 )

116 if referenced_results["boolean"]:

117 tqdm.write(

118 f"Error in SPARQL: Merged responsible agent {entity_uri} is still referenced by other entities"

119 )

120 has_issues = True

121 if not exists:

122 return has_issues

123

124 if not exists:

125 return has_issues

126

127 types_query = f"""

128 SELECT ?type WHERE {{

129 <{entity_uri}> a ?type .

130 }}

131 """

132 types_results = execute_sparql(

133 endpoint, types_query, max_retries=3, backoff_factor=1

134 )

135

136 types = [result["type"]["value"] for result in types_results["results"]["bindings"]]

137 if not types:

138 tqdm.write(f"Error in SPARQL: Entity {entity_uri} has no type")

139 has_issues = True

140 elif len(types) > 1:

141 tqdm.write(f"Error in SPARQL: Entity {entity_uri} has multiple types")

142 has_issues = True

143 elif FOAF + "Agent" not in types:

144 tqdm.write(f"Error in SPARQL: Entity {entity_uri} is not a foaf:Agent")

145 has_issues = True

146

147 names_query = f"""

148 SELECT ?name ?givenName ?familyName WHERE {{

149 OPTIONAL {{ <{entity_uri}> <{FOAF}name> ?name }}

150 OPTIONAL {{ <{entity_uri}> <{FOAF}givenName> ?givenName }}

151 OPTIONAL {{ <{entity_uri}> <{FOAF}familyName> ?familyName }}

152 }}

153 """

154 names_results = execute_sparql(

155 endpoint, names_query, max_retries=3, backoff_factor=1

156 )

157

158 has_name = False

159 for result in names_results["results"]["bindings"]:

160 if any(key in result for key in ["name", "givenName", "familyName"]):

161 has_name = True

162 break

163

164 if not has_name:

165 tqdm.write(f"Error in SPARQL: Entity {entity_uri} has no name properties")

166 has_issues = True

167

168 identifiers_query = f"""

169 SELECT ?identifier WHERE {{

170 <{entity_uri}> <{DATACITE}hasIdentifier> ?identifier .

171 }}

172 """

173 identifiers_results = execute_sparql(

174 endpoint, identifiers_query, max_retries=3, backoff_factor=1

175 )

176

177 identifiers = [

178 result["identifier"]["value"]

179 for result in identifiers_results["results"]["bindings"]

180 ]

181 if not identifiers:

182 tqdm.write(

183 f"Error in SPARQL: Entity {entity_uri} has no datacite:hasIdentifier"

184 )

185 has_issues = True

186

187 if is_surviving and check_has_next_integrity(endpoint, entity_uri):

188 has_issues = True

189

190 return has_issues

191

192

193def get_entity_triples(endpoint: str, entity_uri):

194 query = f"""

195 SELECT ?g ?s ?p ?o

196 WHERE {{

197 GRAPH ?g {{

198 {{

199 <{entity_uri}> ?p ?o .

200 BIND(<{entity_uri}> AS ?s)

201 }}

202 UNION

203 {{

204 ?s ?p <{entity_uri}> .

205 BIND(<{entity_uri}> AS ?o)

206 }}

207 }}

208 }}

209 """

210 results = execute_sparql(endpoint, query, max_retries=3, backoff_factor=1)

211

212 triples = []

213 for result in results["results"]["bindings"]:

214 graph = result["g"]["value"]

215 subject = URIRef(result["s"]["value"])

216 predicate = URIRef(result["p"]["value"])

217

218 obj_data = result["o"]

219 if obj_data["type"] == "uri":

220 obj = URIRef(obj_data["value"])

221 else:

222 datatype = obj_data.get("datatype")

223 obj = Literal(

224 obj_data["value"], datatype=URIRef(datatype) if datatype else None

225 )

226

227 triples.append((graph, subject, predicate, obj))

228

229 return triples

230

231

232def check_entity_provenance(

233 entity_uri, is_surviving, prov_graph: Dataset, prov_file_path

234):

235 def extract_snapshot_number(snapshot_uri):

236 match = re.search(r"/prov/se/(\d+)$", str(snapshot_uri))

237 if match:

238 return int(match.group(1))

239 return 0

240

241 snapshots = list(

242 prov_graph.subjects(PROV.specializationOf, entity_uri, unique=True)

243 )

244 if len(snapshots) <= 1:

245 tqdm.write(

246 f"Error in provenance file {prov_file_path}: Less than two snapshots found for entity {entity_uri}"

247 )

248 return

249

250 snapshots.sort(key=extract_snapshot_number)

251 for i, snapshot in enumerate(snapshots):

252 snapshot_number = extract_snapshot_number(snapshot)

253 if snapshot_number != i + 1:

254 tqdm.write(

255 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} has unexpected number {snapshot_number}, expected {i + 1}"

256 )

257

258 gen_time = prov_graph.value(snapshot, PROV.generatedAtTime)

259 if gen_time is None:

260 tqdm.write(

261 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} has no generation time"

262 )

263

264 if i < len(snapshots) - 1 or not is_surviving:

265 invalidation_time = prov_graph.value(snapshot, PROV.invalidatedAtTime)

266 if invalidation_time is None:

267 tqdm.write(

268 f"Error in provenance file {prov_file_path}: Non-last snapshot {snapshot} has no invalidation time"

269 )

270 elif (

271 is_surviving

272 and prov_graph.value(snapshot, PROV.invalidatedAtTime) is not None

273 ):

274 tqdm.write(

275 f"Error in provenance file {prov_file_path}: Last snapshot of surviving entity {snapshot} should not have invalidation time"

276 )

277

278 description = prov_graph.value(snapshot, DCTERMS.description)

279 is_merge_snapshot = description and "has been merged with" in str(description)

280

281 derived_from = list(

282 prov_graph.objects(snapshot, PROV.wasDerivedFrom, unique=True)

283 )

284 if i == 0: # First snapshot

285 if derived_from:

286 tqdm.write(

287 f"Error in provenance file {prov_file_path}: First snapshot {snapshot} should not have prov:wasDerivedFrom relation"

288 )

289 elif is_merge_snapshot:

290 if len(derived_from) < 2:

291 tqdm.write(

292 f"Error in provenance file {prov_file_path}: Merge snapshot {snapshot} should be derived from at least two snapshots"

293 )

294 else: # Regular modification snapshot

295 if len(derived_from) != 1:

296 tqdm.write(

297 f"Error in provenance file {prov_file_path}: Regular modification snapshot {snapshot} should have exactly one prov:wasDerivedFrom relation"

298 )

299 else:

300 previous_snapshot = snapshots[i - 1]

301 if derived_from[0] != previous_snapshot:

302 tqdm.write(

303 f"Error in provenance file {prov_file_path}: Snapshot {snapshot} is not derived from the previous snapshot"

304 )

305

306

307def process_file_group(args):

308 file_path, entities, sparql_endpoint, query_output_dir = args

309

310 if file_path is None:

311 for entity, is_surviving, surviving_entity in entities:

312 tqdm.write(f"Error: Could not find file for entity {entity}")

313 has_issues = check_entity_sparql(sparql_endpoint, entity, is_surviving)

314 if has_issues and not is_surviving:

315 triples = get_entity_triples(sparql_endpoint, entity)

316 combined_query = generate_update_query(

317 entity, surviving_entity, triples

318 )

319 query_file_path = os.path.join(

320 query_output_dir, f"update_{entity.split('/')[-1]}.sparql"

321 )

322 with open(query_file_path, "w") as f:

323 f.write(combined_query)

324 return

325

326 data_lock_file = f"{file_path}.lock"

327 prov_lock_file = f"{file_path.replace('.zip', '')}/prov/se.zip.lock"

328

329 data_lock = filelock.FileLock(data_lock_file)

330 prov_lock = filelock.FileLock(prov_lock_file)

331

332 try:

333 with data_lock.acquire(timeout=60):

334 try:

335 with zipfile.ZipFile(file_path, "r") as zip_ref:

336 g = Dataset(default_union=True)

337 for filename in zip_ref.namelist():

338 with zip_ref.open(filename) as file:

339 g.parse(file, format="json-ld")

340 except FileNotFoundError:

341 for entity, is_surviving, surviving_entity in entities:

342 tqdm.write(f"Error: File not found for entity {entity}")

343 has_issues = check_entity_sparql(

344 sparql_endpoint, entity, is_surviving

345 )

346 if has_issues and not is_surviving:

347 triples = get_entity_triples(sparql_endpoint, entity)

348 combined_query = generate_update_query(

349 entity, surviving_entity, triples

350 )

351 query_file_path = os.path.join(

352 query_output_dir, f"update_{entity.split('/')[-1]}.sparql"

353 )

354 with open(query_file_path, "w") as f:

355 f.write(combined_query)

356 return

357

358 prov_file_path = file_path.replace(".zip", "") + "/prov/se.zip"

359 try:

360 with prov_lock.acquire(timeout=60):

361 try:

362 with zipfile.ZipFile(prov_file_path, "r") as zip_ref:

363 prov_graph = Dataset(default_union=True)

364 for filename in zip_ref.namelist():

365 with zip_ref.open(filename) as file:

366 prov_graph.parse(file, format="json-ld")

367 except FileNotFoundError:

368 prov_graph = None

369 except zipfile.BadZipFile:

370 prov_graph = "badzip"

371

372 for entity, is_surviving, surviving_entity in entities:

373 if (URIRef(entity), None, None) not in g:

374 if is_surviving:

375 tqdm.write(

376 f"Error in file {file_path}: Surviving entity {entity} does not exist"

377 )

378 else:

379 if not is_surviving:

380 tqdm.write(

381 f"Error in file {file_path}: Merged entity {entity} still exists"

382 )

383 else:

384 agent_issues = check_agent_constraints(

385 g, URIRef(entity)

386 )

387 for issue in agent_issues:

388 tqdm.write(f"Error in file {file_path}: {issue}")

389

390 if prov_graph is None:

391 tqdm.write(

392 f"Error: Provenance file not found for entity {entity}"

393 )

394 elif prov_graph == "badzip":

395 tqdm.write(

396 f"Error: Invalid zip file for provenance of entity {entity}"

397 )

398 else:

399 check_entity_provenance(

400 URIRef(entity), is_surviving, prov_graph, prov_file_path

401 )

402

403 has_issues = check_entity_sparql(

404 sparql_endpoint, entity, is_surviving

405 )

406 if has_issues and not is_surviving:

407 triples = get_entity_triples(sparql_endpoint, entity)

408 combined_query = generate_update_query(

409 entity, surviving_entity, triples

410 )

411 query_file_path = os.path.join(

412 query_output_dir,

413 f"update_{entity.split('/')[-1]}.sparql",

414 )

415 with open(query_file_path, "w") as f:

416 f.write(combined_query)

417

418 except filelock.Timeout:

419 tqdm.write(

420 f"Could not acquire lock for provenance file {prov_file_path} within timeout period"

421 )

422

423 except filelock.Timeout:

424 tqdm.write(

425 f"Could not acquire lock for data file {file_path} within timeout period"

426 )

427

428

429def generate_update_query(merged_entity, surviving_entity, triples):

430 delete_query = "DELETE DATA {\n"

431 insert_query = "INSERT DATA {\n"

432

433 for graph, subject, predicate, obj in triples:

434 if subject == URIRef(merged_entity):

435 delete_query += (

436 f" GRAPH <{graph}> {{ <{subject}> <{predicate}> {obj.n3()} }}\n"

437 )

438 elif obj == URIRef(merged_entity):

439 delete_query += (

440 f" GRAPH <{graph}> {{ <{subject}> <{predicate}> <{obj}> }}\n"

441 )

442 insert_query += f" GRAPH <{graph}> {{ <{subject}> <{predicate}> <{surviving_entity}> }}\n"

443

444 delete_query += "}\n"

445 insert_query += "}\n"

446

447 return delete_query + "\n" + insert_query

448

449

450def process_csv(args, csv_file):

451 csv_path, rdf_dir, meta_config_path, sparql_endpoint, query_output_dir = args

452 csv_path = os.path.join(csv_path, csv_file)

453 data = read_csv(csv_path)

454 tasks = []

455

456 meta_editor = MetaEditor(meta_config_path, "")

457

458 for row in data:

459 if "Done" not in row or row["Done"] != "True":

460 continue

461

462 surviving_entity = row["surviving_entity"]

463 merged_entities = parse_merged_entities(row["merged_entities"])

464 all_entities = [surviving_entity] + merged_entities

465

466 for entity in all_entities:

467 file_path = find_rdf_file(

468 entity,

469 rdf_dir,

470 meta_editor.dir_split,

471 meta_editor.n_file_item,

472 zip_output=True,

473 )

474 tasks.append(

475 (

476 entity,

477 entity == surviving_entity,

478 file_path,

479 rdf_dir,

480 meta_config_path,

481 sparql_endpoint,

482 query_output_dir,

483 surviving_entity,

484 )

485 )

486 return tasks

487

488

489def main():

490 parser = argparse.ArgumentParser(

491 description="Check merge process success on files and SPARQL endpoint for responsible agents",

492 formatter_class=RichHelpFormatter,

493 )

494 parser.add_argument(

495 "csv_folder", type=str, help="Path to the folder containing CSV files"

496 )

497 parser.add_argument("rdf_dir", type=str, help="Path to the RDF directory")

498 parser.add_argument(

499 "--meta_config", type=str, required=True, help="Path to meta configuration file"

500 )

501 parser.add_argument(

502 "--query_output",

503 type=str,

504 required=True,

505 help="Path to the folder where SPARQL queries will be saved",

506 )

507 args = parser.parse_args()

508

509 with open(args.meta_config, "r") as config_file:

510 config = yaml.safe_load(config_file)

511

512 sparql_endpoint = config["triplestore_url"]

513 os.makedirs(args.query_output, exist_ok=True)

514

515 csv_files = [f for f in os.listdir(args.csv_folder) if f.endswith(".csv")]

516

517 # Use forkserver to avoid deadlocks when forking in a multi-threaded environment

518 ctx = multiprocessing.get_context("forkserver")

519

520 # Process CSV files to gather tasks

521 with ctx.Pool(processes=multiprocessing.cpu_count()) as pool:

522 process_csv_partial = partial(

523 process_csv,

524 (

525 args.csv_folder,

526 args.rdf_dir,

527 args.meta_config,

528 sparql_endpoint,

529 args.query_output,

530 ),

531 )

532 all_tasks_list = list(

533 tqdm(

534 pool.imap(process_csv_partial, csv_files),

535 total=len(csv_files),

536 desc="Processing CSV files",

537 )

538 )

539

540 # Flatten the list of tasks

541 all_tasks = [task for sublist in all_tasks_list for task in sublist]

542

543 # Group tasks by file path

544 tasks_by_file = {}

545 for (

546 entity,

547 is_surviving,

548 file_path,

549 rdf_dir,

550 meta_config_path,

551 sparql_endpoint,

552 query_output_dir,

553 surviving_entity,

554 ) in all_tasks:

555 if file_path not in tasks_by_file:

556 tasks_by_file[file_path] = []

557 tasks_by_file[file_path].append((entity, is_surviving, surviving_entity))

558

559 # Convert to list of arguments for parallel processing

560 file_groups = [

561 (file_path, tasks, sparql_endpoint, args.query_output)

562 for file_path, tasks in tasks_by_file.items()

563 ]

564

565 # Process each file group in parallel

566 with ctx.Pool(processes=multiprocessing.cpu_count()) as pool:

567 list(

568 tqdm(

569 pool.imap(process_file_group, file_groups),

570 total=len(file_groups),

571 desc="Processing files",

572 )

573 )

574

575

576if __name__ == "__main__":

577 main()

Coverage for oc_meta / run / merge / check_merged_ras_results.py: 18%

287 statements