Coverage for oc_meta/run/meta/check_rdf

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

3# SPDX-License-Identifier: ISC

5"""Disk-only verification of MetaProcess RDF output over the whole curated CSV.

7Unlike ``check_results.py`` this script never queries the triplestore: it is meant

8for ``rdf_files_only`` runs, where the database is still stale. It scans every row of

9the produced curated CSV (and optionally cross-checks the original input) and verifies

10every referenced entity against the JSON-LD data and provenance zip files on disk.

12Rows are streamed and results aggregated incrementally with a bounded in-flight window,

13so neither the CSV, the worker futures, nor the per-row results are held in memory at

14once.

15"""

17from __future__ import annotations

19import argparse

20import csv

21import multiprocessing

22import os

23import sys

24import traceback

25from collections import Counter

26from collections.abc import Iterator

27from concurrent.futures import (

28 FIRST_COMPLETED,

29 ProcessPoolExecutor,

30 ThreadPoolExecutor,

31 as_completed,

32 wait,

33)

34from dataclasses import dataclass, field

35from datetime import datetime

36from typing import Optional

38import orjson

39import yaml

40from rich.progress import (

41 Progress,

42 SpinnerColumn,

43 TextColumn,

44 TimeElapsedColumn,

45)

46from rich_argparse import RichHelpFormatter

48from oc_meta.core.creator import Creator

49from oc_meta.lib.cleaner import normalize_id

50from oc_meta.lib.console import console

51from oc_meta.lib.file_manager import find_rdf_file

52from oc_meta.run.find.hasnext_anomalies import (

53 HAS_NEXT,

54 IS_DOC_CONTEXT_FOR,

55 find_anomalies,

56)

57from oc_meta.run.meta.check_results import _extract_entity_groups, find_prov_file

58from oc_meta.run.meta.generate_csv import URI_TYPE_DICT, load_json_from_file

60TITLE = "http://purl.org/dc/terms/title"

61PUB_DATE = "http://prismstandard.org/namespaces/basic/2.0/publicationDate"

62HAS_IDENTIFIER = "http://purl.org/spar/datacite/hasIdentifier"

63USES_SCHEME = "http://purl.org/spar/datacite/usesIdentifierScheme"

64HAS_LITERAL = "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue"

65WITH_ROLE = "http://purl.org/spar/pro/withRole"

66IS_HELD_BY = "http://purl.org/spar/pro/isHeldBy"

67SPECIALIZATION_OF = "http://www.w3.org/ns/prov#specializationOf"

68INVALIDATED = "http://www.w3.org/ns/prov#invalidatedAtTime"

69DATACITE_PREFIX = "http://purl.org/spar/datacite/"

71AGENT_COLUMNS = ("author", "editor", "publisher")

72CHECKS = ("data_graph", "identifier", "metadata", "agents", "provenance", "hasnext")

74_config: Optional[tuple[str, int, int]] = None

75_index: Optional[dict[str, str]] = None

78@dataclass

79class RowResult:

80 counts: Counter = field(default_factory=Counter)

81 errors: list[dict] = field(default_factory=list)

82 warnings: list[dict] = field(default_factory=list)

85class EntityCache:

86 """Per-row cache of data files, so one zip is read once even for many entities."""

88 def __init__(self, rdf_dir: str, dir_split: int, items_per_file: int) -> None:

89 self.rdf_dir = rdf_dir

90 self.dir_split = dir_split

91 self.items_per_file = items_per_file

92 self._files: dict[str, Optional[dict[str, dict]]] = {}

94 def _graphs(self, data_zip: str) -> Optional[dict[str, dict]]:

95 if data_zip not in self._files:

96 if not os.path.exists(data_zip):

97 self._files[data_zip] = None

98 else:

99 index: dict[str, dict] = {}

100 for graph in load_json_from_file(data_zip):

101 for entity in graph["@graph"]:

102 index[entity["@id"]] = entity

103 self._files[data_zip] = index

104 return self._files[data_zip]

105

106 def data_file(self, uri: str) -> str:

107 return find_rdf_file(

108 uri, self.rdf_dir, self.dir_split, self.items_per_file, zip_output=True

109 )

110

111 def get(self, uri: str) -> Optional[dict]:

112 graphs = self._graphs(self.data_file(uri))

113 if graphs is None:

114 return None

115 return graphs.get(uri)

116

117

118def _values(entity: dict, predicate: str) -> list[str]:

119 return [item["@value"] for item in entity.get(predicate, [])]

120

121

122def _ids(entity: dict, predicate: str) -> list[str]:

123 return [item["@id"] for item in entity.get(predicate, [])]

124

125

126def _entity_type_string(entity: dict) -> Optional[str]:

127 mapped = [URI_TYPE_DICT[t] for t in entity.get("@type", []) if t in URI_TYPE_DICT]

128 specific = [m for m in mapped if m]

129 if specific:

130 return specific[0]

131 return "" if mapped else None

132

133

134def _canonical_type(label: Optional[str]) -> Optional[str]:

135 """Collapse synonymous type labels (e.g. 'data file' / 'dataset') to one form.

136

137 The curated CSV and the fabio class can carry different labels for the same type,

138 so both sides are mapped through the creator's type vocabulary before comparison.

139 """

140 if not label:

141 return label

142 return Creator._TYPE_TO_METHOD.get(label, label)

143

144

145def _linked_identifiers(br: dict, cache: EntityCache) -> set[str]:

146 keys: set[str] = set()

147 for id_uri in _ids(br, HAS_IDENTIFIER):

148 id_entity = cache.get(id_uri)

149 if id_entity is None:

150 continue

151 schemes = _ids(id_entity, USES_SCHEME)

152 values = _values(id_entity, HAS_LITERAL)

153 if not schemes or not values:

154 continue

155 scheme = schemes[0]

156 scheme = (

157 scheme[len(DATACITE_PREFIX) :]

158 if scheme.startswith(DATACITE_PREFIX)

159 else scheme

160 )

161 normalized = normalize_id(f"{scheme}:{values[0]}")

162 if normalized:

163 keys.add(normalized)

164 return keys

165

166

167def _ar_chain_order(ar_data: dict[str, dict], role_uri: str) -> Optional[list[str]]:

168 """Return the RA sequence following hasNext, or None if the chain is malformed."""

169 members = {ar: info for ar, info in ar_data.items() if info["role_uri"] == role_uri}

170 if not members:

171 return []

172 targets = {t for info in members.values() for t in info["has_next"] if t in members}

173 starts = [ar for ar in members if ar not in targets]

174 if len(starts) != 1:

175 return None

176 order: list[str] = []

177 seen: set[str] = set()

178 current: Optional[str] = starts[0]

179 while current is not None:

180 if current in seen:

181 return None

182 seen.add(current)

183 order.append(members[current]["ra"])

184 nexts = [t for t in members[current]["has_next"] if t in members]

185 if len(nexts) > 1:

186 return None

187 current = nexts[0] if nexts else None

188 if len(seen) != len(members):

189 return None

190 return order

191

192

193def _check_provenance(uri: str, cache: EntityCache) -> Optional[str]:

194 """Return an error subtype, or None if the latest snapshot exists and is valid."""

195 prov_file = find_prov_file(cache.data_file(uri))

196 if prov_file is None:

197 return "provenance_missing"

198 snapshots: dict[int, dict] = {}

199 for graph in load_json_from_file(prov_file):

200 for snapshot in graph["@graph"]:

201 if uri in _ids(snapshot, SPECIALIZATION_OF):

202 number = int(snapshot["@id"].rsplit("/se/", 1)[1])

203 snapshots[number] = snapshot

204 if not snapshots:

205 return "provenance_missing"

206 if INVALIDATED in snapshots[max(snapshots)]:

207 return "provenance_invalidated"

208 return None

209

210

211def check_curated_row(

212 row: dict, row_num: int, cache: EntityCache, base_iri: str

213) -> RowResult:

214 result = RowResult()

215

216 id_group = _extract_entity_groups(row["id"], "id", base_iri)[0]

217 br_uri = id_group["omid_uri"]

218 if not br_uri:

219 result.warnings.append({"type": "row_without_omid", "row": row_num})

220 return result

221

222 ra_by_role: dict[str, list[str]] = {}

223 entity_uris = {br_uri}

224 id_groups_by_uri = {br_uri: id_group}

225 for col in AGENT_COLUMNS:

226 for group in _extract_entity_groups(row[col], col, base_iri):

227 if group["omid_uri"]:

228 ra_by_role.setdefault(col, []).append(group["omid_uri"])

229 entity_uris.add(group["omid_uri"])

230 for group in _extract_entity_groups(row["venue"], "venue", base_iri):

231 if group["omid_uri"]:

232 entity_uris.add(group["omid_uri"])

233 id_groups_by_uri[group["omid_uri"]] = group

234

235 # A. data presence

236 present: dict[str, Optional[dict]] = {}

237 for uri in entity_uris:

238 result.counts["data_graph.checked"] += 1

239 entity = cache.get(uri)

240 present[uri] = entity

241 if entity is None:

242 result.counts["data_graph.failed"] += 1

243 result.errors.append(

244 {

245 "type": "data_graph_missing",

246 "omid": uri,

247 "file": cache.data_file(uri),

248 "row": row_num,

249 }

250 )

251

252 # B. identifier linkage (br + venue)

253 for uri, group in id_groups_by_uri.items():

254 entity = present.get(uri)

255 if entity is None:

256 continue

257 linked = _linked_identifiers(entity, cache)

258 for schema, value in group["recognized"]:

259 normalized = normalize_id(f"{schema}:{value}")

260 if not normalized:

261 continue

262 result.counts["identifier.checked"] += 1

263 if normalized not in linked:

264 result.counts["identifier.failed"] += 1

265 result.errors.append(

266 {

267 "type": "identifier_not_linked",

268 "omid": uri,

269 "identifier": normalized,

270 "linked": sorted(linked),

271 "row": row_num,

272 }

273 )

274

275 br = present[br_uri]

276 if br is not None:

277 _check_metadata(row, row_num, br, result)

278 _check_agents_and_chain(row_num, br, ra_by_role, cache, result)

279

280 # D. provenance presence

281 for uri in entity_uris:

282 if present.get(uri) is None:

283 continue

284 result.counts["provenance.checked"] += 1

285 problem = _check_provenance(uri, cache)

286 if problem == "provenance_missing":

287 result.counts["provenance.failed"] += 1

288 result.errors.append({"type": problem, "omid": uri, "row": row_num})

289 elif problem == "provenance_invalidated":

290 # entity referenced by the run but its latest snapshot is invalidated

291 # (typically a pre-existing merge/deletion leftover, not a fault of this run)

292 result.counts["provenance.invalidated"] += 1

293 result.warnings.append({"type": problem, "omid": uri, "row": row_num})

294

295 return result

296

297

298def _check_metadata(row: dict, row_num: int, br: dict, result: RowResult) -> None:

299 result.counts["metadata.checked"] += 1

300 failed = False

301

302 title = row["title"].strip()

303 rdf_title = _values(br, TITLE)

304 if title and (not rdf_title or rdf_title[0].casefold() != title.casefold()):

305 failed = True

306 result.errors.append(

307 {

308 "type": "metadata_mismatch",

309 "subtype": "title",

310 "omid": br["@id"],

311 "csv": title,

312 "rdf": rdf_title,

313 "row": row_num,

314 }

315 )

316

317 pub_date = row["pub_date"].strip()

318 rdf_date = _values(br, PUB_DATE)

319 if pub_date and (not rdf_date or rdf_date[0] != pub_date):

320 failed = True

321 result.errors.append(

322 {

323 "type": "metadata_mismatch",

324 "subtype": "pub_date",

325 "omid": br["@id"],

326 "csv": pub_date,

327 "rdf": rdf_date,

328 "row": row_num,

329 }

330 )

331

332 type_str = row["type"].strip().lower()

333 rdf_type = _entity_type_string(br)

334 if type_str and _canonical_type(type_str) != _canonical_type(rdf_type):

335 failed = True

336 result.errors.append(

337 {

338 "type": "metadata_mismatch",

339 "subtype": "type",

340 "omid": br["@id"],

341 "csv": type_str,

342 "rdf": rdf_type,

343 "row": row_num,

344 }

345 )

346

347 if failed:

348 result.counts["metadata.failed"] += 1

349

350

351def _load_ar_group(br: dict, cache: EntityCache) -> dict[str, dict]:

352 ar_data: dict[str, dict] = {}

353 for ar_uri in _ids(br, IS_DOC_CONTEXT_FOR):

354 ar = cache.get(ar_uri)

355 if ar is None:

356 continue

357 ar_data[ar_uri] = {

358 "role_uri": (_ids(ar, WITH_ROLE) or [""])[0],

359 "ra": (_ids(ar, IS_HELD_BY) or [None])[0],

360 "has_next": _ids(ar, HAS_NEXT),

361 }

362 return ar_data

363

364

365def _check_agents_and_chain(

366 row_num: int,

367 br: dict,

368 ra_by_role: dict[str, list[str]],

369 cache: EntityCache,

370 result: RowResult,

371) -> None:

372 ar_data = _load_ar_group(br, cache)

373

374 # E. structural anomalies (the regression check for the duplicate-RA fix)

375 result.counts["hasnext.checked"] += 1

376 role_groups: dict[str, dict[str, dict]] = {}

377 for ar_uri, info in ar_data.items():

378 role = info["role_uri"].rsplit("/", 1)[-1] if info["role_uri"] else "unknown"

379 role_groups.setdefault(role, {})[ar_uri] = {

380 "ra": info["ra"],

381 "has_next": info["has_next"],

382 }

383 anomalies = []

384 for role, group in role_groups.items():

385 anomalies.extend(find_anomalies(br["@id"], role, group))

386 if anomalies:

387 result.counts["hasnext.failed"] += 1

388 for anomaly in anomalies:

389 result.errors.append(

390 {

391 "type": "hasnext_anomaly",

392 "anomaly_type": anomaly["anomaly_type"],

393 "omid": br["@id"],

394 "details": anomaly["details"],

395 "row": row_num,

396 }

397 )

398

399 # Orphan agent roles (withRole but no isHeldBy) are a pre-existing structural

400 # leftover, usually from an old merge/deletion. They are excluded from the agent

401 # comparison below and reported separately as warnings.

402 for ar_uri, info in ar_data.items():

403 if info["ra"] is None:

404 result.counts["agents.orphan"] += 1

405 result.warnings.append(

406 {

407 "type": "ar_without_agent",

408 "omid": br["@id"],

409 "ar": ar_uri,

410 "role": info["role_uri"].rsplit("/", 1)[-1] or "unknown",

411 "row": row_num,

412 }

413 )

414

415 # C. agents set + order vs curated CSV (real agents only)

416 result.counts["agents.checked"] += 1

417 agents_failed = False

418 for col, expected in ra_by_role.items():

419 role_uri = f"http://purl.org/spar/pro/{col}"

420 rdf_ras = [

421 info["ra"]

422 for info in ar_data.values()

423 if info["role_uri"] == role_uri and info["ra"] is not None

424 ]

425 if set(expected) != set(rdf_ras):

426 agents_failed = True

427 result.errors.append(

428 {

429 "type": "agents_mismatch",

430 "subtype": col,

431 "omid": br["@id"],

432 "csv": expected,

433 "rdf": rdf_ras,

434 "row": row_num,

435 }

436 )

437 continue

438 order = _ar_chain_order(ar_data, role_uri)

439 if order is not None:

440 order = [ra for ra in order if ra is not None]

441 if order != expected:

442 agents_failed = True

443 result.errors.append(

444 {

445 "type": "agent_order_mismatch",

446 "subtype": col,

447 "omid": br["@id"],

448 "csv": expected,

449 "rdf": order,

450 "row": row_num,

451 }

452 )

453 if agents_failed:

454 result.counts["agents.failed"] += 1

455

456

457def check_input_row(row: dict, row_num: int, base_iri: str) -> RowResult:

458 """Verify each input row is represented in the curated output (no dropped rows).

459

460 Title/metadata fidelity is not compared here: the input text is raw while the RDF

461 holds the fully cleaned form (capitalisation, HTML stripping, ...), so a textual

462 diff is pure noise. Fidelity is covered by check C against the curated CSV.

463 """

464 assert _index is not None

465 result = RowResult()

466 result.counts["input.checked"] += 1

467

468 id_group = _extract_entity_groups(row["id"], "id", base_iri)[0]

469 recognized = [normalize_id(f"{s}:{v}") for s, v in id_group["recognized"]]

470 recognized = [r for r in recognized if r]

471 if not recognized:

472 result.counts["input.unverifiable"] += 1

473 return result

474

475 omids = {_index[r] for r in recognized if r in _index}

476 if not omids:

477 result.counts["input.failed"] += 1

478 result.errors.append(

479 {

480 "type": "input_row_dropped",

481 "identifiers": recognized,

482 "row": row_num,

483 }

484 )

485 elif len(omids) > 1:

486 result.warnings.append(

487 {

488 "type": "input_multiple_omids",

489 "identifiers": recognized,

490 "omids": sorted(omids),

491 "row": row_num,

492 }

493 )

494 return result

495

496

497def _init_worker(rdf_dir: str, dir_split: int, items_per_file: int) -> None:

498 global _config

499 _config = (rdf_dir, dir_split, items_per_file)

500

501

502def _run_curated(args: tuple) -> RowResult:

503 row, row_num, base_iri = args

504 assert _config is not None

505 cache = EntityCache(*_config)

506 try:

507 return check_curated_row(row, row_num, cache, base_iri)

508 except Exception: # noqa: BLE001 - record the row and keep scanning

509 result = RowResult()

510 result.counts["row_exception"] += 1

511 result.errors.append(

512 {

513 "type": "row_check_exception",

514 "row": row_num,

515 "traceback": traceback.format_exc(),

516 }

517 )

518 return result

519

520

521def _text_lines(f):

522 for raw in f:

523 yield raw.decode("utf-8", errors="replace")

524

525

526def _rows(csv_path: str) -> Iterator[tuple[dict, int]]:

527 with open(csv_path, "rb") as f:

528 reader = csv.DictReader(line.replace("\0", "") for line in _text_lines(f))

529 for i, row in enumerate(reader):

530 yield row, i + 1

531

532

533def _iter_curated(

534 csv_path: str, base_iri: str, index: Optional[dict[str, str]]

535) -> Iterator[tuple[dict, int]]:

536 """Stream curated rows, building the id->omid index in the same pass when needed."""

537 for row, row_num in _rows(csv_path):

538 if index is not None:

539 group = _extract_entity_groups(row["id"], "id", base_iri)[0]

540 if group["omid_uri"]:

541 for schema, value in group["recognized"]:

542 normalized = normalize_id(f"{schema}:{value}")

543 if normalized:

544 index[normalized] = group["omid_uri"]

545 yield row, row_num

546

547

548def _count_progress() -> Progress:

549 """Progress that shows a rising count, with no total (rows are not pre-counted)."""

550 return Progress(

551 SpinnerColumn(),

552 TextColumn("[progress.description]{task.description}"),

553 TextColumn("[cyan]{task.completed}[/cyan]"),

554 TimeElapsedColumn(),

555 console=console,

556 )

557

558

559def _drive(

560 executor, submit, items, description: str, workers: int

561) -> tuple[Counter, list[dict], list[dict]]:

562 """Run ``submit`` over a stream of ``items`` with a bounded in-flight window.

563

564 Results are folded incrementally so neither the futures nor the per-row results

565 accumulate in memory.

566 """

567 counts: Counter = Counter()

568 errors: list[dict] = []

569 warnings: list[dict] = []

570 max_in_flight = workers * 4

571 in_flight: set = set()

572

573 with _count_progress() as progress:

574 task = progress.add_task(description)

575

576 def fold(futures) -> None:

577 for future in futures:

578 result = future.result()

579 counts.update(result.counts)

580 errors.extend(result.errors)

581 warnings.extend(result.warnings)

582 progress.advance(task)

583

584 for item in items:

585 in_flight.add(submit(executor, item))

586 if len(in_flight) >= max_in_flight:

587 done, in_flight = wait(in_flight, return_when=FIRST_COMPLETED)

588 fold(done)

589 fold(as_completed(in_flight))

590

591 return counts, errors, warnings

592

593

594def _merge(

595 into: tuple[Counter, list[dict], list[dict]],

596 other: tuple[Counter, list[dict], list[dict]],

597) -> None:

598 into[0].update(other[0])

599 into[1].extend(other[1])

600 into[2].extend(other[2])

601

602

603def main() -> None:

604 parser = argparse.ArgumentParser(

605 description="Check MetaProcess RDF files (data + provenance) on disk",

606 formatter_class=RichHelpFormatter,

607 )

608 parser.add_argument("-c", "--config", required=True, help="Meta config YAML path")

609 parser.add_argument(

610 "--csv", required=True, help="Curated output CSV produced by the run"

611 )

612 parser.add_argument(

613 "--input-csv", help="Original input CSV for the input cross-check"

614 )

615 parser.add_argument("-o", "--output", required=True, help="Output JSON report path")

616 parser.add_argument(

617 "--workers", type=int, default=8, help="Parallel workers (default: 8)"

618 )

619 args = parser.parse_args()

620

621 with open(args.config, encoding="utf-8") as f:

622 settings = yaml.safe_load(f)

623 rdf_dir = os.path.join(settings["output_rdf_dir"], "rdf")

624 base_iri = settings["base_iri"]

625 config = (rdf_dir, settings["dir_split_number"], settings["items_per_file"])

626

627 index: Optional[dict[str, str]] = {} if args.input_csv else None

628

629 console.print(f"Checking every row of {os.path.basename(args.csv)}...")

630 with ProcessPoolExecutor(

631 max_workers=args.workers,

632 initializer=_init_worker,

633 initargs=config,

634 mp_context=multiprocessing.get_context("forkserver"),

635 ) as pool:

636 aggregated = _drive(

637 pool,

638 lambda ex, item: ex.submit(_run_curated, (item[0], item[1], base_iri)),

639 _iter_curated(args.csv, base_iri, index),

640 "Checking curated rows",

641 args.workers,

642 )

643

644 if args.input_csv:

645 assert index is not None

646 global _index

647 _index = index

648 console.print(

649 f"Cross-checking every row of {os.path.basename(args.input_csv)}..."

650 )

651 with ThreadPoolExecutor(max_workers=args.workers) as pool:

652 _merge(

653 aggregated,

654 _drive(

655 pool,

656 lambda ex, item: ex.submit(

657 check_input_row, item[0], item[1], base_iri

658 ),

659 _rows(args.input_csv),

660 "Checking input rows",

661 args.workers,

662 ),

663 )

664

665 counts, errors, warnings = aggregated

666 status = "PASS" if not errors else "FAIL"

667 summary = {

668 check: {

669 "checked": counts[f"{check}.checked"],

670 "failed": counts[f"{check}.failed"],

671 }

672 for check in CHECKS

673 }

674 summary["provenance"]["invalidated"] = counts["provenance.invalidated"]

675 summary["input"] = {

676 "checked": counts["input.checked"],

677 "failed": counts["input.failed"],

678 "unverifiable": counts["input.unverifiable"],

679 }

680

681 report = {

682 "status": status,

683 "timestamp": datetime.now().isoformat(),

684 "config_path": os.path.abspath(args.config),

685 "csv": os.path.abspath(args.csv),

686 "input_csv": os.path.abspath(args.input_csv) if args.input_csv else None,

687 "summary": summary,

688 "errors_total": len(errors),

689 "warnings_total": len(warnings),

690 "errors": errors,

691 "warnings": warnings,

692 }

693 os.makedirs(os.path.dirname(os.path.abspath(args.output)) or ".", exist_ok=True)

694 with open(args.output, "wb") as f:

695 f.write(orjson.dumps(report, option=orjson.OPT_INDENT_2))

696

697 console.print(f"\nStatus: [bold]{status}[/bold]")

698 for check, stats in summary.items():

699 console.print(

700 f" {check:11} checked={stats['checked']} failed={stats.get('failed', 0)}"

701 )

702 console.print(f"Errors: {len(errors)}, Warnings: {len(warnings)}")

703 console.print(f"Report saved to {args.output}")

704 if errors:

705 sys.exit(1)

706

707

708if __name__ == "__main__":

709 main()

Coverage for oc_meta / run / meta / check_rdf_files.py: 68%

370 statements