Coverage for oc_meta / run / patches / add_missing_provenance.py: 0%
136 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
6from __future__ import annotations
8import argparse
9import multiprocessing
10import os
11import zipfile
12from collections import Counter
13from datetime import datetime, timezone
15import orjson
16import yaml
17from oc_ocdm.counter_handler.filesystem_counter_handler import FilesystemCounterHandler
18from oc_ocdm.graph import GraphSet
19from oc_ocdm.prov import ProvSet
20from oc_ocdm.reader import Reader
21from oc_ocdm.storer import Storer
22from oc_ocdm.support import get_prefix
23from rich_argparse import RichHelpFormatter
25from oc_meta.lib.console import console, create_progress
26from oc_meta.lib.file_manager import collect_zip_files
28PROV_SPECIALIZATION_OF = "http://www.w3.org/ns/prov#specializationOf"
31def _read_json_zip(path: str) -> list:
32 with zipfile.ZipFile(path) as z:
33 json_name = next(n for n in z.namelist() if n.endswith(".json"))
34 return orjson.loads(z.read(json_name))
37def _prov_file(data_zip: str) -> str:
38 return os.path.join(os.path.splitext(data_zip)[0], "prov", "se.zip")
41def _entity_type(uri: str) -> str:
42 return uri.rstrip("/").split("/")[-2]
45def _check_file(data_zip: str) -> tuple[str, int, list[str], bool]:
46 """Return (data_zip, entity_count, missing_uris, prov_file_absent)."""
47 data = _read_json_zip(data_zip)
48 entities = [ent["@id"] for graph in data for ent in graph["@graph"]]
50 prov_path = _prov_file(data_zip)
51 prov_absent = not os.path.exists(prov_path)
52 covered: set[str] = set()
53 if not prov_absent:
54 for graph in _read_json_zip(prov_path):
55 for snapshot in graph["@graph"]:
56 if PROV_SPECIALIZATION_OF in snapshot:
57 for spec in snapshot[PROV_SPECIALIZATION_OF]:
58 covered.add(spec["@id"])
60 missing = [e for e in entities if e not in covered]
61 return data_zip, len(entities), missing, prov_absent
64def _backfill_file(
65 data_file: str,
66 base_iri: str,
67 base_dir: str,
68 info_dir_root: str,
69 dir_split: int,
70 items_per_file: int,
71 zip_output: bool,
72 resp_agent: str,
73 prov_endpoint: str,
74 hotfix_dir: str,
75 upload: bool,
76) -> int:
77 supplier_prefix = get_prefix(_read_json_zip(data_file)[0]["@graph"][0]["@id"])
78 info_dir = os.path.join(info_dir_root, supplier_prefix) + os.sep
79 counter_handler = FilesystemCounterHandler(
80 info_dir=info_dir, supplier_prefix=supplier_prefix
81 )
82 g_set = GraphSet(
83 base_iri,
84 supplier_prefix=supplier_prefix,
85 wanted_label=False,
86 custom_counter_handler=counter_handler,
87 )
88 reader = Reader()
89 graph = reader.load(data_file)
90 if graph is None:
91 raise ValueError(f"Could not load RDF data from {data_file}")
92 reader.import_entities_from_graph(g_set, graph, resp_agent)
94 prov_set = ProvSet(
95 g_set,
96 base_iri,
97 wanted_label=False,
98 supplier_prefix=supplier_prefix,
99 custom_counter_handler=counter_handler,
100 )
101 created = prov_set.generate_provenance()
103 storer = Storer(
104 prov_set,
105 dir_split=dir_split,
106 n_file_item=items_per_file,
107 zip_output=zip_output,
108 )
109 storer.store_all(base_dir, base_iri)
110 if upload:
111 storer.upload_all(prov_endpoint, base_dir=hotfix_dir)
112 return len(created)
115def _find(
116 rdf_dir: str, workers: int, entities_path: str
117) -> tuple[int, Counter, list[str], int]:
118 all_files = collect_zip_files(rdf_dir, only_data=True)
119 console.print(f"Scanning {len(all_files)} data files with {workers} workers...")
121 total_missing = 0
122 by_type: Counter = Counter()
123 affected: set[str] = set()
124 prov_files_absent = 0
126 ctx = multiprocessing.get_context("forkserver")
127 with open(entities_path, "w", encoding="utf-8") as ent_out:
128 with ctx.Pool(workers) as pool:
129 with create_progress() as progress:
130 task = progress.add_task("Checking provenance", total=len(all_files))
131 for data_zip, _count, missing, prov_absent in pool.imap_unordered(
132 _check_file, all_files, chunksize=200
133 ):
134 if prov_absent:
135 prov_files_absent += 1
136 if missing:
137 affected.add(data_zip)
138 total_missing += len(missing)
139 for uri in missing:
140 by_type[_entity_type(uri)] += 1
141 ent_out.write("\n".join(missing) + "\n")
142 progress.update(task, advance=1)
144 return total_missing, by_type, sorted(affected), prov_files_absent
147def main() -> None:
148 parser = argparse.ArgumentParser(
149 description="Backfill the missing se/1 provenance snapshot of data entities",
150 formatter_class=RichHelpFormatter,
151 )
152 parser.add_argument("-c", "--config", required=True, help="Meta config YAML path")
153 parser.add_argument("-o", "--output", required=True, help="Output JSON report path")
154 parser.add_argument(
155 "--workers", type=int, default=4, help="Scan workers (default: 4)"
156 )
157 parser.add_argument(
158 "-r", "--resp-agent", help="Responsible agent URI (default: config resp_agent)"
159 )
160 mode = parser.add_mutually_exclusive_group(required=True)
161 mode.add_argument(
162 "--dry-run", action="store_true", dest="dry_run", help="Report only"
163 )
164 mode.add_argument(
165 "--no-dry-run",
166 action="store_true",
167 dest="no_dry_run",
168 help="Apply the backfill",
169 )
170 args = parser.parse_args()
171 args.dry_run = not args.no_dry_run
173 with open(args.config, encoding="utf-8") as f:
174 settings = yaml.safe_load(f)
176 base_output_dir = settings["base_output_dir"]
177 base_dir = os.path.join(base_output_dir, "rdf") + os.sep
178 info_dir_root = os.path.join(base_output_dir, "info_dir")
179 hotfix_dir = os.path.join(base_output_dir, "to_be_uploaded_hotfix")
180 base_iri = settings["base_iri"]
181 dir_split = settings["dir_split_number"]
182 items_per_file = settings["items_per_file"]
183 zip_output = settings["zip_output_rdf"]
184 rdf_files_only = settings.get("rdf_files_only", False)
185 prov_endpoint = settings["provenance_triplestore_url"]
186 resp_agent = args.resp_agent or settings["resp_agent"]
188 if not os.path.exists(base_dir):
189 parser.error(f"RDF directory not found at {base_dir}")
191 entities_path = os.path.splitext(os.path.abspath(args.output))[0] + ".entities.txt"
192 os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True)
194 total_missing, by_type, affected, prov_files_absent = _find(
195 base_dir, args.workers, entities_path
196 )
198 console.print(
199 f"\n[bold]Found {total_missing} entities without provenance[/bold] "
200 f"in {len(affected)} data files {dict(sorted(by_type.items()))}"
201 )
203 upload = (not args.dry_run) and (not rdf_files_only)
204 created_total = 0
205 fixed_files = 0
206 failures: list[dict] = []
208 if not args.dry_run:
209 console.print(
210 f"\n[bold]Backfilling se/1[/bold] (upload to triplestore: {upload})..."
211 )
212 with create_progress() as progress:
213 task = progress.add_task("Backfilling se/1", total=len(affected))
214 for data_file in affected:
215 try:
216 created_total += _backfill_file(
217 data_file,
218 base_iri,
219 base_dir,
220 info_dir_root,
221 dir_split,
222 items_per_file,
223 zip_output,
224 resp_agent,
225 prov_endpoint,
226 hotfix_dir,
227 upload,
228 )
229 fixed_files += 1
230 except Exception as e: # noqa: BLE001 - record per-file and continue
231 failures.append({"file": data_file, "error": str(e)})
232 console.print(f"[red]Error on {data_file}: {e}[/red]")
233 progress.update(task, advance=1)
235 report = {
236 "config": os.path.abspath(args.config),
237 "rdf_dir": base_dir,
238 "timestamp": datetime.now(timezone.utc).isoformat(),
239 "dry_run": args.dry_run,
240 "resp_agent": resp_agent,
241 "entities_without_provenance": total_missing,
242 "missing_by_type": dict(sorted(by_type.items())),
243 "affected_data_files": len(affected),
244 "prov_files_absent": prov_files_absent,
245 "snapshots_created": created_total,
246 "files_fixed": fixed_files,
247 "failures": failures,
248 "entities_file": entities_path,
249 }
250 with open(args.output, "wb") as f:
251 f.write(orjson.dumps(report, option=orjson.OPT_INDENT_2))
253 if args.dry_run:
254 console.print(
255 f"\n[dim]Dry run: would create {total_missing} se/1 snapshots "
256 f"across {len(affected)} files. Use --no-dry-run to apply.[/dim]"
257 )
258 else:
259 console.print(
260 f"\nCreated {created_total} se/1 snapshots in {fixed_files} files "
261 f"({len(failures)} failures)"
262 )
263 console.print(f"Report saved to {args.output}")
266if __name__ == "__main__":
267 main()