Coverage for oc_meta / run / meta / check_results.py: 91%
385 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import argparse
6import multiprocessing
7import os
8import sys
9import zipfile
10from concurrent.futures import ProcessPoolExecutor, as_completed
11from dataclasses import dataclass, field
12from datetime import datetime
13from typing import Callable, Dict, List, Set
15import orjson
16import polars as pl
17import yaml
18from rich.progress import (
19 BarColumn,
20 Progress,
21 TaskProgressColumn,
22 TextColumn,
23 TimeElapsedColumn,
24)
25from rich_argparse import RichHelpFormatter
26from oc_meta.constants import (
27 BR_ID_SCHEMAS,
28 QLEVER_BATCH_SIZE,
29 QLEVER_MAX_WORKERS,
30 RA_ID_SCHEMAS,
31)
32from oc_meta.lib.cleaner import normalize_hyphens, normalize_id
33from oc_meta.lib.console import EMATimeRemainingColumn, console
34from oc_meta.lib.file_manager import collect_files, find_rdf_file
35from oc_meta.lib.master_of_regex import RE_SEMICOLON_IN_PEOPLE_FIELD, split_name_and_ids
36from oc_meta.lib.sparql import run_queries_parallel
38MAX_RETRIES = 10
39RETRY_BACKOFF = 2
40DATACITE_PREFIX = "http://purl.org/spar/datacite/"
41RECOGNIZED_SCHEMAS = BR_ID_SCHEMAS | RA_ID_SCHEMAS
43_SPACE_PATTERN = "[\t\xa0\u200b\u202f\u2003\u2005\u2009]"
44_ID_COLUMNS = ["id", "author", "editor", "publisher", "venue"]
45_STAT_FIELDS = (
46 "total_rows",
47 "rows_with_ids",
48 "total_identifiers",
49 "omid_schema_identifiers",
50 "identifiers_skipped_invalid",
51 "identifiers_skipped_unverifiable",
52 "identifiers_with_omids",
53 "identifiers_without_omids",
54 "identifiers_with_omid_mismatch",
55 "data_graphs_found",
56 "data_graphs_missing",
57 "prov_graphs_found",
58 "prov_graphs_missing",
59 "omids_with_provenance",
60 "omids_without_provenance",
61)
64@dataclass
65class FileResult:
66 file: str
67 total_rows: int = 0
68 rows_with_ids: int = 0
69 total_identifiers: int = 0
70 omid_schema_identifiers: int = 0
71 identifiers_skipped_invalid: int = 0
72 identifiers_skipped_unverifiable: int = 0
73 identifiers_with_omids: int = 0
74 identifiers_without_omids: int = 0
75 identifiers_with_omid_mismatch: int = 0
76 data_graphs_found: int = 0
77 data_graphs_missing: int = 0
78 prov_graphs_found: int = 0
79 prov_graphs_missing: int = 0
80 omids_with_provenance: int = 0
81 omids_without_provenance: int = 0
82 errors: list = field(default_factory=list)
83 id_key_to_omids: dict = field(default_factory=dict)
84 id_key_locations: dict = field(default_factory=dict)
87def check_provenance_existence(
88 omids: List[str],
89 prov_endpoint_url: str,
90 workers: int = QLEVER_MAX_WORKERS,
91 progress_callback: Callable[[int], None] | None = None,
92) -> Dict[str, bool]:
93 if not omids:
94 return {}
96 prov_results = {omid: False for omid in omids}
98 batch_queries = []
99 batch_sizes = []
100 for i in range(0, len(omids), QLEVER_BATCH_SIZE):
101 batch = omids[i : i + QLEVER_BATCH_SIZE]
102 values_entries = " ".join(f"<{omid}/prov/se/1>" for omid in batch)
103 query = f"""
104 SELECT ?snapshot WHERE {{
105 VALUES ?snapshot {{ {values_entries} }}
106 ?snapshot <http://www.w3.org/ns/prov#specializationOf> ?o .
107 }}
108 """
109 batch_queries.append(query)
110 batch_sizes.append(len(batch))
112 all_bindings = run_queries_parallel(
113 prov_endpoint_url,
114 batch_queries,
115 batch_sizes,
116 workers,
117 progress_callback,
118 max_retries=MAX_RETRIES,
119 backoff_factor=RETRY_BACKOFF,
120 )
122 for bindings in all_bindings:
123 for result in bindings:
124 snapshot_uri = result["snapshot"]["value"]
125 omid = snapshot_uri.rsplit("/prov/se/1", 1)[0]
126 prov_results[omid] = True
128 return prov_results
131def check_omids_existence(
132 identifiers: List[Dict[str, str]],
133 endpoint_url: str,
134 workers: int = QLEVER_MAX_WORKERS,
135 progress_callback: Callable[[int], None] | None = None,
136) -> Dict[str, Set[str]]:
137 if not identifiers:
138 return {}
140 found_omids: Dict[str, Set[str]] = {}
142 batch_queries = []
143 batch_sizes = []
144 for i in range(0, len(identifiers), QLEVER_BATCH_SIZE):
145 batch = identifiers[i : i + QLEVER_BATCH_SIZE]
147 values_entries = []
148 for identifier in batch:
149 escaped_value = (
150 identifier["value"].replace("\\", "\\\\").replace('"', '\\"')
151 )
152 values_entries.append(
153 f'("{escaped_value}"^^xsd:string datacite:{identifier["schema"]})'
154 )
156 query = f"""
157 PREFIX datacite: <http://purl.org/spar/datacite/>
158 PREFIX literal: <http://www.essepuntato.it/2010/06/literalreification/>
159 PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
161 SELECT ?val ?scheme ?entity
162 WHERE {{
163 VALUES (?val ?scheme) {{ {" ".join(values_entries)} }}
164 ?id_entity literal:hasLiteralValue ?val ;
165 datacite:usesIdentifierScheme ?scheme .
166 ?entity datacite:hasIdentifier ?id_entity .
167 }}
168 """
169 batch_queries.append(query)
170 batch_sizes.append(len(batch))
172 all_bindings = run_queries_parallel(
173 endpoint_url,
174 batch_queries,
175 batch_sizes,
176 workers,
177 progress_callback,
178 max_retries=MAX_RETRIES,
179 backoff_factor=RETRY_BACKOFF,
180 )
182 for bindings in all_bindings:
183 for result in bindings:
184 entity = result["entity"]["value"]
185 val = result["val"]["value"]
186 scheme_uri = result["scheme"]["value"]
187 scheme = (
188 scheme_uri[len(DATACITE_PREFIX) :]
189 if scheme_uri.startswith(DATACITE_PREFIX)
190 else scheme_uri
191 )
192 id_key = f"{scheme}:{val}"
193 if id_key not in found_omids:
194 found_omids[id_key] = set()
195 found_omids[id_key].add(entity)
197 return found_omids
200def find_prov_file(data_zip_path: str) -> str | None:
201 base_dir = os.path.dirname(data_zip_path)
202 file_name = os.path.splitext(os.path.basename(data_zip_path))[0]
203 prov_dir = os.path.join(base_dir, file_name, "prov")
204 prov_file = os.path.join(prov_dir, "se.zip")
205 return prov_file if os.path.exists(prov_file) else None
208def _check_zip_file(args: tuple) -> tuple[str, dict[str, tuple[bool, bool]]]:
209 zip_path, omids = args
211 raw_data = b""
212 with zipfile.ZipFile(zip_path, "r") as z:
213 json_files = [f for f in z.namelist() if f.endswith(".json")]
214 if json_files:
215 with z.open(json_files[0]) as f:
216 raw_data = f.read()
218 raw_prov = b""
219 prov_path = find_prov_file(zip_path)
220 if prov_path:
221 with zipfile.ZipFile(prov_path, "r") as z:
222 json_files = [f for f in z.namelist() if f.endswith(".json")]
223 if json_files:
224 with z.open(json_files[0]) as f:
225 raw_prov = f.read()
227 results: dict[str, tuple[bool, bool]] = {}
228 for omid in omids:
229 omid_bytes = (omid + '"').encode()
230 results[omid] = (
231 omid_bytes in raw_data,
232 (omid + '/prov/"').encode() in raw_prov,
233 )
234 return zip_path, results
237def _extract_entity_groups(cell: str, col: str, base_iri: str) -> list[dict]:
238 base = base_iri.rstrip("/")
240 def parse_tokens(token_str: str) -> dict:
241 group: dict = {"omid_uri": None, "recognized": [], "unverifiable": []}
242 for token in token_str.strip().split():
243 colon_pos = token.find(":")
244 if colon_pos <= 0:
245 continue
246 schema = token[:colon_pos].lower()
247 value = normalize_hyphens(token[colon_pos + 1 :])
248 if schema == "omid":
249 group["omid_uri"] = f"{base}/{value}"
250 elif schema in RECOGNIZED_SCHEMAS:
251 group["recognized"].append((schema, value))
252 else:
253 group["unverifiable"].append((schema, value))
254 return group
256 if col == "id":
257 return [parse_tokens(cell)]
259 groups = []
260 for element in RE_SEMICOLON_IN_PEOPLE_FIELD.split(cell):
261 element = element.strip()
262 if not element:
263 continue
264 _, ids_str = split_name_and_ids(element)
265 if not ids_str:
266 continue
267 groups.append(parse_tokens(ids_str))
268 return groups
271def process_csv_file(
272 args: tuple, workers: int = QLEVER_MAX_WORKERS, progress=None, task_id=None
273) -> FileResult:
274 (
275 csv_file,
276 endpoint_url,
277 prov_endpoint_url,
278 rdf_dir,
279 dir_split_number,
280 items_per_file,
281 zip_output_rdf,
282 base_iri,
283 ) = args
285 result = FileResult(file=os.path.basename(csv_file))
287 if progress and task_id is not None:
288 progress.update(task_id, detail="Phase 1/5: Reading CSV")
290 with open(csv_file, "rb") as f:
291 raw = f.read()
292 df = pl.read_csv(
293 raw.replace(b"\0", b""), columns=_ID_COLUMNS, infer_schema_length=0
294 )
295 df = df.with_columns(
296 [
297 pl.col(c)
298 .str.replace_all(_SPACE_PATTERN, " ")
299 .str.replace_all(" ", " ", literal=True)
300 for c in _ID_COLUMNS
301 ]
302 )
304 result.total_rows = len(df)
305 col_lists = {col: df[col].to_list() for col in _ID_COLUMNS}
306 del df
308 # entity URIs from CSV OMID tokens — used for data graph + prov verification
309 all_entity_uris: set[str] = set()
310 entity_uri_to_info: dict[str, tuple[int, str]] = {} # uri → (row_num, col)
312 # recognized IDs for SPARQL lookup
313 recognized_ids: list[dict] = []
314 recognized_id_set: set[str] = set()
315 recognized_id_to_csv_omid: dict[str, str | None] = {}
316 recognized_id_occurrences: dict[str, list[tuple[int, str]]] = {}
317 recognized_id_meta: dict[str, tuple[str, str]] = {}
319 phase1_task = None
320 if progress:
321 phase1_task = progress.add_task(
322 " Phase 1/5: Extracting identifiers", total=result.total_rows, detail=""
323 )
325 for row_idx in range(result.total_rows):
326 row_has_ids = False
327 row_num = row_idx + 1
329 for col in _ID_COLUMNS:
330 cell = col_lists[col][row_idx]
331 if not cell:
332 continue
334 groups = _extract_entity_groups(cell, col, base_iri)
335 for group in groups:
336 n_tokens = (
337 (1 if group["omid_uri"] else 0)
338 + len(group["recognized"])
339 + len(group["unverifiable"])
340 )
341 if n_tokens == 0:
342 continue
344 row_has_ids = True
345 result.total_identifiers += n_tokens
347 if group["omid_uri"]:
348 result.omid_schema_identifiers += 1
349 all_entity_uris.add(group["omid_uri"])
350 if group["omid_uri"] not in entity_uri_to_info:
351 entity_uri_to_info[group["omid_uri"]] = (row_num, col)
353 result.identifiers_skipped_unverifiable += len(group["unverifiable"])
355 for schema, value in group["recognized"]:
356 normalized = normalize_id(f"{schema}:{value}")
357 if not normalized:
358 result.identifiers_skipped_invalid += 1
359 continue
360 norm_schema, norm_value = normalized.split(":", 1)
361 id_key = normalized
362 if id_key not in recognized_id_set:
363 recognized_id_set.add(id_key)
364 recognized_ids.append(
365 {"schema": norm_schema, "value": norm_value}
366 )
367 recognized_id_to_csv_omid[id_key] = group["omid_uri"]
368 recognized_id_meta[id_key] = (norm_schema, norm_value)
369 recognized_id_occurrences[id_key] = []
370 recognized_id_occurrences[id_key].append((row_num, col))
372 if row_has_ids:
373 result.rows_with_ids += 1
375 if progress and phase1_task is not None:
376 progress.advance(phase1_task)
378 if progress and phase1_task is not None:
379 progress.update(phase1_task, visible=False)
381 del col_lists
383 phase2_task = None
384 if progress:
385 phase2_task = progress.add_task(
386 " Phase 2/5: Querying DB", total=len(recognized_ids), detail=""
387 )
389 def on_id_batch(batch_size: int):
390 if progress and phase2_task is not None:
391 progress.advance(phase2_task, batch_size)
393 identifier_cache = check_omids_existence(
394 recognized_ids, endpoint_url, workers=workers, progress_callback=on_id_batch
395 )
397 if progress and phase2_task is not None:
398 progress.update(phase2_task, visible=False)
400 omids_by_file: dict[str, set[str]] = {}
401 path_exists_cache: dict[str, bool] = {}
402 omid_to_id_info: dict[str, tuple[int, str, str]] = {}
403 csv_basename = os.path.basename(csv_file)
405 phase3_task = None
406 if progress:
407 phase3_task = progress.add_task(
408 " Phase 3/5: Mapping OMIDs", total=len(recognized_id_set), detail=""
409 )
411 for id_key in recognized_id_set:
412 occurrences = recognized_id_occurrences[id_key]
413 expected_omid = recognized_id_to_csv_omid[id_key]
414 schema, value = recognized_id_meta[id_key]
416 if id_key in identifier_cache:
417 entity_uris = identifier_cache[id_key]
418 result.id_key_to_omids[id_key] = entity_uris
419 result.id_key_locations[id_key] = [
420 {"file": csv_basename, "row": r, "column": c} for r, c in occurrences
421 ]
423 if expected_omid and expected_omid not in entity_uris:
424 result.identifiers_with_omid_mismatch += len(occurrences)
425 for row_num, col in occurrences:
426 result.errors.append(
427 {
428 "type": "omid_mismatch",
429 "schema": schema,
430 "value": value,
431 "expected_omid": expected_omid,
432 "found_omids": sorted(entity_uris),
433 "file": csv_basename,
434 "row": row_num,
435 "column": col,
436 }
437 )
438 else:
439 result.identifiers_with_omids += len(occurrences)
440 else:
441 result.identifiers_without_omids += len(occurrences)
442 for row_num, col in occurrences:
443 result.errors.append(
444 {
445 "type": "identifier_not_in_triplestore",
446 "schema": schema,
447 "value": value,
448 "file": csv_basename,
449 "row": row_num,
450 "column": col,
451 }
452 )
454 if progress and phase3_task is not None:
455 progress.advance(phase3_task)
457 if progress and phase3_task is not None:
458 progress.update(phase3_task, visible=False)
460 # Build file map from all entity URIs collected from CSV OMIDs
461 for entity_uri in all_entity_uris:
462 if entity_uri not in omid_to_id_info:
463 if entity_uri in entity_uri_to_info:
464 row_num, col = entity_uri_to_info[entity_uri]
465 omid_to_id_info[entity_uri] = (row_num, col, entity_uri)
466 zip_path = find_rdf_file(
467 entity_uri, rdf_dir, dir_split_number, items_per_file, zip_output_rdf
468 )
469 if zip_path not in path_exists_cache:
470 path_exists_cache[zip_path] = os.path.exists(zip_path)
471 if path_exists_cache[zip_path]:
472 if zip_path in omids_by_file:
473 omids_by_file[zip_path].add(entity_uri)
474 else:
475 omids_by_file[zip_path] = {entity_uri}
477 total_rdf_files = len(omids_by_file)
478 total_omids = len(all_entity_uris)
480 prov_future = None
481 prov_executor = None
482 if total_omids > 0:
483 prov_executor = ProcessPoolExecutor(
484 max_workers=1, mp_context=multiprocessing.get_context("forkserver")
485 )
486 prov_future = prov_executor.submit(
487 check_provenance_existence,
488 list(all_entity_uris),
489 prov_endpoint_url,
490 workers,
491 )
493 phase4_task = None
494 if progress:
495 phase4_task = progress.add_task(
496 " Phase 4/5: Checking RDF files", total=total_rdf_files, detail=""
497 )
499 zip_args = [(zp, list(omids)) for zp, omids in omids_by_file.items()]
501 def _apply_zip_results(zip_results: dict[str, tuple[bool, bool]]) -> None:
502 for _, (data_found, prov_found) in zip_results.items():
503 if data_found:
504 result.data_graphs_found += 1
505 else:
506 result.data_graphs_missing += 1
507 if prov_found:
508 result.prov_graphs_found += 1
509 else:
510 result.prov_graphs_missing += 1
512 if zip_args and workers > 1:
513 with ProcessPoolExecutor(
514 max_workers=min(len(zip_args), workers),
515 mp_context=multiprocessing.get_context("forkserver"),
516 ) as executor:
517 for future in as_completed(
518 {executor.submit(_check_zip_file, a): a for a in zip_args}
519 ):
520 _zip_path, zip_results = future.result()
521 _apply_zip_results(zip_results)
522 if progress and phase4_task is not None:
523 progress.advance(phase4_task)
524 else:
525 for a in zip_args:
526 _zip_path, zip_results = _check_zip_file(a)
527 _apply_zip_results(zip_results)
528 if progress and phase4_task is not None:
529 progress.advance(phase4_task)
531 if progress and phase4_task is not None:
532 progress.update(phase4_task, visible=False)
534 phase5_task = None
535 if progress:
536 phase5_task = progress.add_task(
537 " Phase 5/5: Checking provenance", total=total_omids, detail=""
538 )
540 prov_results: Dict[str, bool] = {}
541 if prov_future and prov_executor:
542 prov_results = prov_future.result()
543 prov_executor.shutdown(wait=False)
545 if progress and phase5_task is not None:
546 progress.advance(phase5_task, total_omids)
547 progress.update(phase5_task, visible=False)
549 for omid, has_prov in prov_results.items():
550 if has_prov:
551 result.omids_with_provenance += 1
552 else:
553 result.omids_without_provenance += 1
554 if omid in omid_to_id_info:
555 row_num, col, id_key = omid_to_id_info[omid]
556 result.errors.append(
557 {
558 "type": "missing_provenance",
559 "omid": omid,
560 "identifier": id_key,
561 "file": csv_basename,
562 "row": row_num,
563 }
564 )
566 return result
569def main():
570 parser = argparse.ArgumentParser(
571 description="Check MetaProcess results by verifying output CSV identifiers against the triplestore",
572 formatter_class=RichHelpFormatter,
573 )
574 parser.add_argument("meta_config", help="Path to meta_config.yaml file")
575 parser.add_argument("output", help="Output file path for results (JSON)")
576 parser.add_argument(
577 "--csv",
578 help="Path to a single output CSV file or directory of output CSVs. "
579 "Defaults to {base_output_dir}/csv/ from meta_config.",
580 default=None,
581 )
582 parser.add_argument(
583 "--workers",
584 type=int,
585 default=QLEVER_MAX_WORKERS,
586 help=f"Max parallel SPARQL workers (default: {QLEVER_MAX_WORKERS})",
587 )
588 args = parser.parse_args()
590 with open(args.meta_config, "r", encoding="utf-8") as f:
591 config = yaml.safe_load(f)
593 if args.csv:
594 csv_source = args.csv
595 if os.path.isfile(csv_source):
596 csv_files = [csv_source]
597 else:
598 csv_files = collect_files(csv_source, pattern="*.csv")
599 else:
600 output_csv_dir = os.path.join(config["base_output_dir"], "csv")
601 csv_files = collect_files(output_csv_dir, pattern="*.csv")
603 base_output_dir = config["base_output_dir"]
604 output_rdf_dir = os.path.join(base_output_dir, "rdf")
605 endpoint_url = config["triplestore_url"]
606 prov_endpoint_url = config["provenance_triplestore_url"]
607 base_iri = config["base_iri"].rstrip("/") + "/"
609 if not os.path.exists(output_rdf_dir):
610 console.print(f"RDF directory not found at {output_rdf_dir}")
611 return
613 if not csv_files:
614 console.print("No CSV files found")
615 return
617 console.print(f"Found {len(csv_files)} CSV files to process")
619 output_dir = os.path.dirname(args.output) or "."
620 os.makedirs(output_dir, exist_ok=True)
622 all_file_results: list[FileResult] = []
624 process_args = [
625 (
626 f,
627 endpoint_url,
628 prov_endpoint_url,
629 output_rdf_dir,
630 config["dir_split_number"],
631 config["items_per_file"],
632 config["zip_output_rdf"],
633 base_iri,
634 )
635 for f in csv_files
636 ]
638 with Progress(
639 TextColumn("[progress.description]{task.description}"),
640 BarColumn(),
641 TextColumn("[cyan]{task.completed:.0f}/{task.total:.0f}"),
642 TaskProgressColumn(),
643 TimeElapsedColumn(),
644 EMATimeRemainingColumn(),
645 TextColumn("[cyan]{task.fields[detail]}"),
646 ) as progress:
647 task = progress.add_task(
648 "Processing CSV files", total=len(csv_files), detail=""
649 )
651 for idx, proc_args in enumerate(process_args):
652 current_file = os.path.basename(csv_files[idx])
653 progress.update(task, detail=f"[{idx + 1}/{len(csv_files)}] {current_file}")
654 file_result = process_csv_file(
655 proc_args, workers=args.workers, progress=progress, task_id=task
656 )
657 all_file_results.append(file_result)
658 progress.advance(task)
660 errors: list[dict] = []
661 merged_id_key_to_omids: dict[str, set[str]] = {}
662 merged_id_key_locations: dict[str, list[dict]] = {}
664 for fr in all_file_results:
665 errors.extend(fr.errors)
666 for id_key, omids in fr.id_key_to_omids.items():
667 if id_key in merged_id_key_to_omids:
668 merged_id_key_to_omids[id_key].update(omids)
669 else:
670 merged_id_key_to_omids[id_key] = set(omids)
671 for id_key, locs in fr.id_key_locations.items():
672 if id_key in merged_id_key_locations:
673 merged_id_key_locations[id_key].extend(locs)
674 else:
675 merged_id_key_locations[id_key] = list(locs)
677 warnings: list[dict] = []
678 for id_key, omids in merged_id_key_to_omids.items():
679 if len(omids) > 1:
680 warnings.append(
681 {
682 "type": "multiple_omids",
683 "identifier": id_key,
684 "omid_count": len(omids),
685 "omids": sorted(omids),
686 "occurrences": merged_id_key_locations[id_key],
687 }
688 )
690 summary = {k: 0 for k in _STAT_FIELDS}
691 files_output = []
692 for fr in all_file_results:
693 file_dict = {k: getattr(fr, k) for k in ("file",) + _STAT_FIELDS}
694 files_output.append(file_dict)
695 for k in _STAT_FIELDS:
696 summary[k] += getattr(fr, k)
698 status = "PASS" if not errors else "FAIL"
700 report = {
701 "status": status,
702 "timestamp": datetime.now().isoformat(),
703 "config_path": os.path.abspath(args.meta_config),
704 "total_files_processed": len(csv_files),
705 "files": files_output,
706 "summary": summary,
707 "errors": errors,
708 "warnings": warnings,
709 }
711 with open(args.output, "wb") as f:
712 f.write(orjson.dumps(report, option=orjson.OPT_INDENT_2))
714 console.print(f"Status: {status}")
715 console.print(f"Errors: {len(errors)}, Warnings: {len(warnings)}")
716 console.print(f"Results written to: {args.output}")
718 if errors:
719 sys.exit(1)
722if __name__ == "__main__":
723 main() # pragma: no cover