Skills › Scientific Agent Skills › hypogenic › audit_dataset.py
audit_dataset.py
412 lines · skills/hypogenic/scripts/audit_dataset.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit pinned HypoGeniC dataset files, schemas, splits, and duplicates."""
4from __future__ import annotations
6import argparse
7import hashlib
8from collections import Counter, defaultdict
9from collections.abc import Mapping, Sequence
10from pathlib import Path
11from typing import Any
13if __package__:
14 from ._common import (
15 MAX_CONFIG_BYTES,
16 MAX_JSON_BYTES,
17 MAX_ROWS,
18 CliError,
19 bounded_int,
20 canonical_json_bytes,
21 checked_input_file,
22 checked_root,
23 count_values,
24 emit_error,
25 emit_json,
26 load_json_document,
27 resolve_manifest_path,
28 sha256_file,
29 validate_dataset_manifest,
30 )
31else:
32 from _common import ( # type: ignore
33 MAX_CONFIG_BYTES,
34 MAX_JSON_BYTES,
35 MAX_ROWS,
36 CliError,
37 bounded_int,
38 canonical_json_bytes,
39 checked_input_file,
40 checked_root,
41 count_values,
42 emit_error,
43 emit_json,
44 load_json_document,
45 resolve_manifest_path,
46 sha256_file,
47 validate_dataset_manifest,
48 )
51def build_parser() -> argparse.ArgumentParser:
52 parser = argparse.ArgumentParser(
53 description=(
54 "Audit immutable local dataset files without network access and without "
55 "rendering or following instructions in dataset text."
56 )
57 )
58 parser.add_argument("--manifest", required=True, help="Strict local manifest JSON")
59 parser.add_argument(
60 "--manifest-root",
61 default=".",
62 help="Existing boundary containing the manifest",
63 )
64 parser.add_argument(
65 "--data-root",
66 default=None,
67 help=(
68 "Existing pinned dataset directory. If omitted, use manifest.root "
69 "beneath --manifest-root."
70 ),
71 )
72 parser.add_argument(
73 "--max-file-bytes",
74 type=int,
75 default=MAX_JSON_BYTES,
76 help=f"Per-split byte cap (default {MAX_JSON_BYTES})",
77 )
78 parser.add_argument(
79 "--max-rows",
80 type=int,
81 default=100_000,
82 help="Per-split row cap (default 100000)",
83 )
84 parser.add_argument(
85 "--max-evidence-groups",
86 type=int,
87 default=20,
88 help="Maximum hashed duplicate groups reported per category",
89 )
90 return parser
93def _data_root_from_manifest(
94 manifest_root: str,
95 relative_root: str,
96) -> Path:
97 base = checked_root(manifest_root)
98 candidate = base / relative_root
99 current = base
100 for part in Path(relative_root).parts:
101 current /= part
102 if current.is_symlink():
103 raise CliError(f"manifest root contains a symlink: {part!r}")
104 try:
105 resolved = candidate.resolve(strict=True)
106 resolved.relative_to(base)
107 except (OSError, ValueError) as exc:
108 raise CliError("manifest root is not an existing directory within its boundary") from exc
109 return checked_root(resolved)
112def _field_names(row: Mapping[str, Any], *, context: str) -> tuple[str, ...]:
113 if not row or len(row) > 128:
114 raise CliError(f"{context} must contain 1 to 128 fields")
115 fields: list[str] = []
116 for key in row:
117 if not isinstance(key, str) or not key or len(key) > 128:
118 raise CliError(f"{context} contains an invalid field name")
119 fields.append(key)
120 return tuple(fields)
123def _rows_from_document(
124 document: Any,
125 *,
126 max_rows: int,
127 split_name: str,
128) -> tuple[list[dict[str, Any]], str]:
129 """Normalize upstream column-oriented JSON or a strict list of row objects."""
131 if isinstance(document, dict):
132 fields = _field_names(document, context=f"{split_name} columns")
133 columns: dict[str, list[Any]] = {}
134 lengths: set[int] = set()
135 for field in fields:
136 value = document[field]
137 if not isinstance(value, list):
138 raise CliError(f"{split_name}.{field} must be a list")
139 columns[field] = value
140 lengths.add(len(value))
141 if len(lengths) != 1:
142 raise CliError(f"{split_name} columns do not have equal lengths")
143 row_count = next(iter(lengths))
144 if row_count == 0 or row_count > max_rows:
145 raise CliError(f"{split_name} must contain 1 to {max_rows} rows")
146 rows = [
147 {field: columns[field][index] for field in fields}
148 for index in range(row_count)
149 ]
150 return rows, "column_oriented"
152 if isinstance(document, list):
153 if not document or len(document) > max_rows:
154 raise CliError(f"{split_name} must contain 1 to {max_rows} rows")
155 if not all(isinstance(row, dict) for row in document):
156 raise CliError(f"{split_name} row-oriented data must contain only objects")
157 first_fields = _field_names(document[0], context=f"{split_name}[0]")
158 field_set = set(first_fields)
159 rows = []
160 for index, row in enumerate(document):
161 if set(row) != field_set:
162 raise CliError(f"{split_name}[{index}] has a different schema")
163 rows.append(dict(row))
164 return rows, "row_oriented"
166 raise CliError(
167 f"{split_name} JSON root must be a column object or a list of row objects"
168 )
171def _digest(value: Any) -> str:
172 return hashlib.sha256(canonical_json_bytes(value)).hexdigest()
175def _bounded_groups(
176 groups: Mapping[str, list[tuple[str, int]]],
177 *,
178 maximum: int,
179 cross_split_only: bool,
180) -> list[dict[str, Any]]:
181 result: list[dict[str, Any]] = []
182 if maximum == 0:
183 return result
184 for digest in sorted(groups):
185 occurrences = groups[digest]
186 split_names = sorted({split for split, _ in occurrences})
187 if cross_split_only and len(split_names) < 2:
188 continue
189 if not cross_split_only and len(occurrences) < 2:
190 continue
191 result.append(
192 {
193 "sha256": digest,
194 "splits": split_names,
195 "occurrence_count": len(occurrences),
196 "occurrences": [
197 {"split": split, "row_index": index}
198 for split, index in occurrences[:20]
199 ],
200 "occurrences_truncated": len(occurrences) > 20,
201 }
202 )
203 if len(result) >= maximum:
204 break
205 return result
208def audit(
209 manifest: dict[str, Any],
210 *,
211 data_root: Path,
212 max_file_bytes: int,
213 max_rows: int,
214 max_evidence_groups: int,
215 manifest_sha256: str,
216) -> dict[str, Any]:
217 label_field = manifest["label_field"]
218 identity_fields = manifest["identity_fields"]
219 exact_groups: dict[str, list[tuple[str, int]]] = defaultdict(list)
220 identity_groups: dict[str, list[tuple[str, int]]] = defaultdict(list)
221 split_reports: list[dict[str, Any]] = []
222 schemas: dict[str, tuple[str, ...]] = {}
223 checksum_failures: list[str] = []
224 within_split_duplicate_extras = 0
226 for split in manifest["splits"]:
227 name = split["name"]
228 path = resolve_manifest_path(
229 split["path"],
230 data_root=data_root,
231 suffixes={".json"},
232 max_bytes=max_file_bytes,
233 )
234 actual_sha256 = sha256_file(path, max_bytes=max_file_bytes)
235 checksum_matches = actual_sha256 == split["sha256"]
236 if not checksum_matches:
237 checksum_failures.append(name)
238 document = load_json_document(
239 split["path"],
240 root=data_root,
241 max_bytes=max_file_bytes,
242 )
243 rows, layout = _rows_from_document(
244 document,
245 max_rows=max_rows,
246 split_name=name,
247 )
248 schema = tuple(rows[0])
249 schemas[name] = schema
250 missing = [
251 field for field in [label_field, *identity_fields] if field not in schema
252 ]
253 if missing:
254 raise CliError(f"{name} is missing manifest fields: {', '.join(missing)}")
256 labels: list[str] = []
257 split_exact = Counter()
258 for index, row in enumerate(rows):
259 label = row[label_field]
260 if not isinstance(label, str) or not label or len(label) > 256:
261 raise CliError(f"{name}[{index}].{label_field} must be a bounded string")
262 labels.append(label)
263 exact = _digest(row)
264 identity = _digest({field: row[field] for field in identity_fields})
265 exact_groups[exact].append((name, index))
266 identity_groups[identity].append((name, index))
267 split_exact[exact] += 1
268 duplicates = sum(count - 1 for count in split_exact.values() if count > 1)
269 within_split_duplicate_extras += duplicates
270 split_reports.append(
271 {
272 "name": name,
273 "path": split["path"],
274 "layout": layout,
275 "row_count": len(rows),
276 "field_count": len(schema),
277 "fields": list(schema),
278 "label_counts_redacted": count_values(labels),
279 "expected_sha256": split["sha256"],
280 "actual_sha256": actual_sha256,
281 "checksum_matches": checksum_matches,
282 "within_split_exact_duplicate_extra_rows": duplicates,
283 "raw_text_included": False,
284 }
285 )
287 schema_sets = {tuple(sorted(schema)) for schema in schemas.values()}
288 schemas_match = len(schema_sets) == 1
289 cross_exact_all = {
290 digest: occurrences
291 for digest, occurrences in exact_groups.items()
292 if len({split for split, _ in occurrences}) > 1
293 }
294 cross_identity_all = {
295 digest: occurrences
296 for digest, occurrences in identity_groups.items()
297 if len({split for split, _ in occurrences}) > 1
298 }
299 ok = (
300 not checksum_failures
301 and schemas_match
302 and not cross_exact_all
303 and not cross_identity_all
304 )
305 warnings: list[str] = []
306 if within_split_duplicate_extras:
307 warnings.append("within-split exact duplicates were found")
308 if not schemas_match:
309 warnings.append("split schemas differ")
310 return {
311 "ok": ok,
312 "audit_kind": "hypogenic_dataset_local_audit",
313 "schema_version": manifest["schema_version"],
314 "manifest_sha256": manifest_sha256,
315 "source": manifest["source"],
316 "data_root_redacted": True,
317 "label_field": label_field,
318 "identity_fields": identity_fields,
319 "splits": split_reports,
320 "checksums": {
321 "all_match": not checksum_failures,
322 "failed_splits": checksum_failures,
323 },
324 "schemas_match_across_splits": schemas_match,
325 "duplicates": {
326 "within_split_exact_duplicate_extra_rows": within_split_duplicate_extras,
327 "cross_split_exact_group_count": len(cross_exact_all),
328 "cross_split_identity_group_count": len(cross_identity_all),
329 "cross_split_exact_evidence": _bounded_groups(
330 cross_exact_all,
331 maximum=max_evidence_groups,
332 cross_split_only=True,
333 ),
334 "cross_split_identity_evidence": _bounded_groups(
335 cross_identity_all,
336 maximum=max_evidence_groups,
337 cross_split_only=True,
338 ),
339 "evidence_is_hashes_and_indices_only": True,
340 },
341 "split_policy": {
342 "preserved": ok and not cross_identity_all,
343 "train_use": "generation_and_update",
344 "validation_use": "selection_only",
345 "test_use": "locked_until_final_evaluation",
346 },
347 "warnings": warnings,
348 "dataset_text_interpreted_as_instructions": False,
349 "raw_dataset_text_included": False,
350 "network_access": False,
351 "model_called": False,
352 }
355def main(argv: Sequence[str] | None = None) -> int:
356 args = build_parser().parse_args(argv)
357 try:
358 max_file_bytes = bounded_int(
359 args.max_file_bytes,
360 name="max_file_bytes",
361 minimum=1,
362 maximum=MAX_JSON_BYTES,
363 )
364 max_rows = bounded_int(
365 args.max_rows,
366 name="max_rows",
367 minimum=1,
368 maximum=MAX_ROWS,
369 )
370 max_evidence = bounded_int(
371 args.max_evidence_groups,
372 name="max_evidence_groups",
373 minimum=0,
374 maximum=1000,
375 )
376 manifest_path = checked_input_file(
377 args.manifest,
378 root=args.manifest_root,
379 suffixes={".json"},
380 max_bytes=MAX_CONFIG_BYTES,
381 )
382 manifest = validate_dataset_manifest(
383 load_json_document(
384 args.manifest,
385 root=args.manifest_root,
386 max_bytes=MAX_CONFIG_BYTES,
387 )
388 )
389 data_root = (
390 checked_root(args.data_root)
391 if args.data_root is not None
392 else _data_root_from_manifest(args.manifest_root, manifest["root"])
393 )
394 report = audit(
395 manifest,
396 data_root=data_root,
397 max_file_bytes=max_file_bytes,
398 max_rows=max_rows,
399 max_evidence_groups=max_evidence,
400 manifest_sha256=sha256_file(
401 manifest_path,
402 max_bytes=MAX_CONFIG_BYTES,
403 ),
404 )
405 emit_json(report)
406 return 0 if report["ok"] else 3
407 except (CliError, OSError) as error:
408 return emit_error(error)
411if __name__ == "__main__":
412 raise SystemExit(main())