Coverage for oc_meta / run / infodir / check.py: 85%
203 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
5from __future__ import annotations
7import argparse
8import os
9import tempfile
10from dataclasses import dataclass, field
11from datetime import datetime, timezone
12from itertools import islice
14from rich_argparse import RichHelpFormatter
16from oc_meta.lib.console import console
17from oc_meta.run.infodir._common import (
18 COUNTERS_PER_CHUNK,
19 ENTITY_TYPES,
20 CounterKey,
21 DataScanResult,
22 SourceDataError,
23 SparseCounterStore,
24 scan_data,
25 scan_provenance,
26 write_json,
27)
29DEFAULT_WORKERS = 4
30DEFAULT_MAX_EXAMPLES = 100
33@dataclass
34class ResultCategory:
35 max_examples: int
36 total: int = 0
37 examples: list[dict[str, object]] = field(default_factory=list)
39 def add(self, example: dict[str, object]) -> None:
40 self.total += 1
41 if len(self.examples) < self.max_examples:
42 self.examples.append(example)
44 def report(self) -> dict[str, object]:
45 return {
46 "total": self.total,
47 "examples": self.examples,
48 "truncated": self.total > len(self.examples),
49 }
52@dataclass(frozen=True)
53class CounterFiles:
54 entity: dict[CounterKey, str]
55 provenance: dict[CounterKey, str]
58def check_info_dir(
59 root_path: str,
60 info_dir: str,
61 output_path: str,
62 workers: int = DEFAULT_WORKERS,
63 max_examples: int = DEFAULT_MAX_EXAMPLES,
64 temp_dir: str | None = None,
65) -> tuple[dict[str, object], int]:
66 root_path = os.path.abspath(root_path)
67 info_dir = os.path.abspath(info_dir)
68 if max_examples < 0:
69 raise ValueError("max_examples must be non-negative")
71 scratch_parent = (
72 os.path.abspath(temp_dir) if temp_dir is not None else os.path.dirname(info_dir)
73 )
74 provenance_scan = None
75 data_scan = None
76 try:
77 with tempfile.TemporaryDirectory(
78 prefix=".check-info-dir.", dir=scratch_parent
79 ) as scratch_dir:
80 store = SparseCounterStore(scratch_dir)
81 try:
82 provenance_scan = scan_provenance(root_path, store, workers)
83 data_scan = scan_data(
84 root_path,
85 store,
86 workers,
87 max_examples,
88 )
89 if provenance_scan.zip_files == 0 and data_scan.zip_files == 0:
90 raise SourceDataError(f"No RDF ZIP files found in {root_path}")
91 report = _compare_info_dir(
92 root_path,
93 info_dir,
94 store,
95 provenance_scan.zip_files,
96 provenance_scan.entities,
97 data_scan,
98 max_examples,
99 )
100 finally:
101 store.close()
102 except SourceDataError as error:
103 report = _scan_failed_report(
104 root_path,
105 info_dir,
106 str(error),
107 provenance_scan.zip_files if provenance_scan is not None else 0,
108 provenance_scan.entities if provenance_scan is not None else 0,
109 data_scan.zip_files if data_scan is not None else 0,
110 data_scan.entities if data_scan is not None else 0,
111 )
112 write_json(output_path, report)
113 console.print(f"Info directory check failed: {error}")
114 console.print(f"Report saved to {os.path.abspath(output_path)}")
115 return report, 2
117 write_json(output_path, report)
118 status = report["status"]
119 exit_code = 0 if status == "aligned" else 1
120 console.print(f"Info directory status: {status}")
121 console.print(f"Report saved to {os.path.abspath(output_path)}")
122 return report, exit_code
125def _compare_info_dir(
126 root_path: str,
127 info_dir: str,
128 store: SparseCounterStore,
129 provenance_zip_files: int,
130 provenance_entities: int,
131 data_scan: DataScanResult,
132 max_examples: int,
133) -> dict[str, object]:
134 entity_mismatches = ResultCategory(max_examples)
135 provenance_mismatches = ResultCategory(max_examples)
136 file_errors = ResultCategory(max_examples)
137 counter_files = _counter_files(info_dir, file_errors)
139 expected_entity: dict[CounterKey, int] = {}
140 for key in store.keys() | set(data_scan.maxima):
141 expected_entity[key] = max(
142 store.maximum(key),
143 data_scan.maxima[key] if key in data_scan.maxima else 0,
144 )
146 for key in sorted(set(expected_entity) | set(counter_files.entity)):
147 expected = expected_entity[key] if key in expected_entity else 0
148 path = counter_files.entity[key] if key in counter_files.entity else None
149 if path is None:
150 entity_mismatches.add(_entity_mismatch(key, expected, None, "missing"))
151 continue
152 actual = _read_entity_counter(path, file_errors)
153 if actual is None:
154 continue
155 if key not in expected_entity:
156 file_errors.add(
157 {
158 "path": path,
159 "error": "unexpected_entity_counter_file",
160 }
161 )
162 if actual != expected:
163 relation = "too_low" if actual < expected else "too_high"
164 entity_mismatches.add(_entity_mismatch(key, expected, actual, relation))
166 for key in sorted(store.keys() | set(counter_files.provenance)):
167 path = (
168 counter_files.provenance[key] if key in counter_files.provenance else None
169 )
170 _compare_provenance_file(
171 store,
172 key,
173 path,
174 provenance_mismatches,
175 file_errors,
176 )
178 live_without_provenance = {
179 "total": data_scan.missing_provenance,
180 "examples": data_scan.missing_examples,
181 "truncated": data_scan.missing_provenance > len(data_scan.missing_examples),
182 }
183 if entity_mismatches.total or provenance_mismatches.total or file_errors.total:
184 status = "mismatched"
185 elif data_scan.missing_provenance:
186 status = "warnings"
187 else:
188 status = "aligned"
190 return {
191 "timestamp": datetime.now(timezone.utc).isoformat(),
192 "status": status,
193 "root_path": root_path,
194 "info_dir": info_dir,
195 "source": {
196 "data_zip_files": data_scan.zip_files,
197 "provenance_zip_files": provenance_zip_files,
198 "data_entities": data_scan.entities,
199 "provenance_entities": provenance_entities,
200 },
201 "entity_counter_mismatches": entity_mismatches.report(),
202 "provenance_counter_mismatches": provenance_mismatches.report(),
203 "counter_file_errors": file_errors.report(),
204 "live_entities_without_provenance": live_without_provenance,
205 }
208def _counter_files(info_dir: str, file_errors: ResultCategory) -> CounterFiles:
209 entity: dict[CounterKey, str] = {}
210 provenance: dict[CounterKey, str] = {}
211 if not os.path.isdir(info_dir):
212 file_errors.add(
213 {
214 "path": info_dir,
215 "error": "missing_info_directory",
216 }
217 )
218 return CounterFiles(entity=entity, provenance=provenance)
220 with os.scandir(info_dir) as prefix_entries:
221 for prefix_entry in prefix_entries:
222 if not prefix_entry.is_dir():
223 continue
224 prefix = prefix_entry.name
225 with os.scandir(prefix_entry.path) as file_entries:
226 for file_entry in file_entries:
227 if not file_entry.is_file() or not file_entry.name.endswith(".txt"):
228 continue
229 parsed = _counter_filename(file_entry.name)
230 if parsed is None:
231 file_errors.add(
232 {
233 "path": file_entry.path,
234 "error": "unexpected_counter_file",
235 }
236 )
237 continue
238 kind, short_name = parsed
239 key = (prefix, short_name)
240 if kind == "entity":
241 entity[key] = file_entry.path
242 else:
243 provenance[key] = file_entry.path
244 return CounterFiles(entity=entity, provenance=provenance)
247def _counter_filename(filename: str) -> tuple[str, str] | None:
248 if filename.startswith("info_file_"):
249 kind = "entity"
250 short_name = filename.removeprefix("info_file_").removesuffix(".txt")
251 elif filename.startswith("prov_file_"):
252 kind = "provenance"
253 short_name = filename.removeprefix("prov_file_").removesuffix(".txt")
254 else:
255 return None
256 if short_name not in ENTITY_TYPES:
257 return None
258 return kind, short_name
261def _read_entity_counter(path: str, file_errors: ResultCategory) -> int | None:
262 with open(path, encoding="utf-8") as input_file:
263 lines = input_file.read().splitlines()
264 if len(lines) != 1:
265 file_errors.add(
266 {
267 "path": path,
268 "error": "invalid_entity_counter_line_count",
269 "expected": 1,
270 "actual": len(lines),
271 }
272 )
273 if not lines:
274 return None
275 try:
276 value = int(lines[0])
277 except ValueError:
278 file_errors.add(
279 {
280 "path": path,
281 "error": "invalid_entity_counter_value",
282 "value": lines[0],
283 }
284 )
285 return None
286 if value < 0:
287 file_errors.add(
288 {
289 "path": path,
290 "error": "invalid_entity_counter_value",
291 "value": value,
292 }
293 )
294 return None
295 return value
298def _compare_provenance_file(
299 store: SparseCounterStore,
300 key: CounterKey,
301 path: str | None,
302 mismatches: ResultCategory,
303 file_errors: ResultCategory,
304) -> None:
305 expected_lines = store.maximum(key)
306 if path is None:
307 if expected_lines == 0:
308 return
309 file_errors.add(
310 {
311 "prefix": key[0],
312 "short_name": key[1],
313 "error": "missing_provenance_counter_file",
314 }
315 )
316 for start, expected in store.iter_chunks(key):
317 for offset, value in enumerate(expected):
318 if value:
319 mismatches.add(
320 _provenance_mismatch(
321 key,
322 start + offset,
323 value,
324 0,
325 "missing",
326 )
327 )
328 return
330 if expected_lines == 0:
331 file_errors.add(
332 {
333 "path": path,
334 "error": "unexpected_provenance_counter_file",
335 }
336 )
338 actual_lines = 0
339 with open(path, encoding="utf-8") as input_file:
340 while True:
341 lines = list(islice(input_file, COUNTERS_PER_CHUNK))
342 if not lines:
343 break
344 start = actual_lines + 1
345 expected = store.read_span(key, start, start + len(lines) - 1)
346 for offset, line in enumerate(lines):
347 resource_number = start + offset
348 actual = _parse_provenance_counter(
349 line,
350 path,
351 resource_number,
352 file_errors,
353 )
354 if actual is None:
355 continue
356 expected_value = expected[offset]
357 if actual != expected_value:
358 mismatches.add(
359 _provenance_mismatch(
360 key,
361 resource_number,
362 expected_value,
363 actual,
364 _counter_relation(expected_value, actual),
365 )
366 )
367 actual_lines += len(lines)
369 if actual_lines < expected_lines:
370 for start, expected in store.iter_chunks(key, start=actual_lines + 1):
371 for offset, value in enumerate(expected):
372 if value:
373 mismatches.add(
374 _provenance_mismatch(
375 key,
376 start + offset,
377 value,
378 0,
379 "missing",
380 )
381 )
382 if actual_lines != expected_lines:
383 file_errors.add(
384 {
385 "path": path,
386 "error": "invalid_provenance_counter_line_count",
387 "expected": expected_lines,
388 "actual": actual_lines,
389 }
390 )
393def _parse_provenance_counter(
394 line: str,
395 path: str,
396 resource_number: int,
397 file_errors: ResultCategory,
398) -> int | None:
399 stripped = line.strip()
400 if not stripped:
401 return 0
402 try:
403 value = int(stripped)
404 except ValueError:
405 file_errors.add(
406 {
407 "path": path,
408 "line": resource_number,
409 "error": "invalid_provenance_counter_value",
410 "value": stripped,
411 }
412 )
413 return None
414 if value < 0:
415 file_errors.add(
416 {
417 "path": path,
418 "line": resource_number,
419 "error": "invalid_provenance_counter_value",
420 "value": value,
421 }
422 )
423 return None
424 return value
427def _entity_mismatch(
428 key: CounterKey,
429 expected: int,
430 actual: int | None,
431 relation: str,
432) -> dict[str, object]:
433 return {
434 "prefix": key[0],
435 "short_name": key[1],
436 "expected": expected,
437 "actual": actual,
438 "relation": relation,
439 }
442def _provenance_mismatch(
443 key: CounterKey,
444 resource_number: int,
445 expected: int,
446 actual: int,
447 relation: str,
448) -> dict[str, object]:
449 return {
450 "prefix": key[0],
451 "short_name": key[1],
452 "resource_number": resource_number,
453 "expected": expected,
454 "actual": actual,
455 "relation": relation,
456 }
459def _counter_relation(expected: int, actual: int) -> str:
460 if expected == 0:
461 return "unexpected"
462 if actual == 0:
463 return "missing"
464 return "too_low" if actual < expected else "too_high"
467def _scan_failed_report(
468 root_path: str,
469 info_dir: str,
470 error: str,
471 provenance_zip_files: int,
472 provenance_entities: int,
473 data_zip_files: int,
474 data_entities: int,
475) -> dict[str, object]:
476 return {
477 "timestamp": datetime.now(timezone.utc).isoformat(),
478 "status": "scan_failed",
479 "root_path": root_path,
480 "info_dir": info_dir,
481 "source": {
482 "data_zip_files": data_zip_files,
483 "provenance_zip_files": provenance_zip_files,
484 "data_entities": data_entities,
485 "provenance_entities": provenance_entities,
486 "error": error,
487 },
488 }
491def main() -> int: # pragma: no cover
492 parser = argparse.ArgumentParser(
493 description="Verify filesystem counters against RDF data and provenance.",
494 formatter_class=RichHelpFormatter,
495 )
496 parser.add_argument("directory", type=str, help="Path to the RDF directory to scan")
497 parser.add_argument("info_dir", type=str, help="Counter directory to verify")
498 parser.add_argument(
499 "-o",
500 "--output",
501 type=str,
502 default="check_info_dir_report.json",
503 help="Output JSON report path (default: check_info_dir_report.json)",
504 )
505 parser.add_argument(
506 "--workers",
507 type=int,
508 default=DEFAULT_WORKERS,
509 help=f"Worker processes (default: {DEFAULT_WORKERS})",
510 )
511 parser.add_argument(
512 "--max-examples",
513 type=int,
514 default=DEFAULT_MAX_EXAMPLES,
515 help=f"Examples retained per category (default: {DEFAULT_MAX_EXAMPLES})",
516 )
517 parser.add_argument(
518 "--temp-dir",
519 help="Temporary storage directory (default: parent of info_dir)",
520 )
521 args = parser.parse_args()
522 _, exit_code = check_info_dir(
523 args.directory,
524 args.info_dir,
525 args.output,
526 workers=args.workers,
527 max_examples=args.max_examples,
528 temp_dir=args.temp_dir,
529 )
530 return exit_code
533if __name__ == "__main__": # pragma: no cover
534 raise SystemExit(main())