Skills › Scientific Agent Skills › geniml › corpus_auditor.py
corpus_auditor.py
304 lines · skills/geniml/scripts/corpus_auditor.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit a local interval manifest without exposing sample metadata values."""
4from __future__ import annotations
6import argparse
7import sys
8from collections import Counter, defaultdict
9from pathlib import Path
11from _common import (
12 HARD_MAX_BYTES,
13 HARD_MAX_FILES,
14 SafetyError,
15 add_path_mode_argument,
16 delimiter_from_path,
17 display_path,
18 fail_json,
19 int_type,
20 local_path,
21 print_json,
22 read_delimited_manifest,
23 sha256_file,
24)
27TOOL = "interval-corpus-auditor"
30def build_parser() -> argparse.ArgumentParser:
31 parser = argparse.ArgumentParser(
32 description=(
33 "Audit a local CSV/TSV manifest for paths, assemblies, duplicate "
34 "content, and patient/donor split leakage. Metadata values are not printed."
35 )
36 )
37 parser.add_argument("--manifest", required=True, help="Local CSV/TSV manifest.")
38 parser.add_argument(
39 "--delimiter",
40 choices=("auto", "csv", "tsv"),
41 default="auto",
42 )
43 parser.add_argument(
44 "--path-column",
45 default="path",
46 help="Column containing BED paths (default: path).",
47 )
48 parser.add_argument(
49 "--assembly-column",
50 default="assembly",
51 help="Assembly column (default: assembly).",
52 )
53 parser.add_argument(
54 "--expected-assembly",
55 help="Optional expected assembly/accession.",
56 )
57 parser.add_argument(
58 "--group-column",
59 action="append",
60 default=[],
61 help="Patient/donor grouping column; repeat as needed.",
62 )
63 parser.add_argument(
64 "--split-column",
65 help="Train/validation/test column used for leakage checks.",
66 )
67 parser.add_argument(
68 "--checksums",
69 action="store_true",
70 help="Hash each bounded BED to detect duplicate content.",
71 )
72 parser.add_argument(
73 "--max-files",
74 type=int_type(minimum=1, maximum=HARD_MAX_FILES, label="max-files"),
75 default=10_000,
76 )
77 parser.add_argument(
78 "--max-manifest-bytes",
79 type=int_type(
80 minimum=1,
81 maximum=HARD_MAX_BYTES,
82 label="max-manifest-bytes",
83 ),
84 default=64 * 1024**2,
85 )
86 parser.add_argument(
87 "--max-file-bytes",
88 type=int_type(minimum=1, maximum=HARD_MAX_BYTES, label="max-file-bytes"),
89 default=2 * 1024**3,
90 )
91 parser.add_argument(
92 "--max-total-bytes",
93 type=int_type(minimum=1, maximum=HARD_MAX_BYTES, label="max-total-bytes"),
94 default=4 * 1024**3,
95 )
96 parser.add_argument(
97 "--max-output-files",
98 type=int_type(minimum=0, maximum=1_000, label="max-output-files"),
99 default=100,
100 help="Maximum bounded per-file summaries (default: 100).",
101 )
102 add_path_mode_argument(parser)
103 return parser
106def _manifest_local_path(raw: str, manifest: Path) -> Path:
107 candidate = Path(raw)
108 if not candidate.is_absolute():
109 candidate = manifest.parent / candidate
110 return local_path(str(candidate), kind="file")
113def audit(args: argparse.Namespace) -> tuple[dict, int]:
114 manifest = local_path(args.manifest, kind="file")
115 delimiter = delimiter_from_path(manifest, args.delimiter)
116 fields, rows = read_delimited_manifest(
117 manifest,
118 delimiter_name=delimiter,
119 max_bytes=args.max_manifest_bytes,
120 max_rows=args.max_files,
121 )
122 if args.path_column not in fields:
123 raise SafetyError(f"path column is missing: {args.path_column}")
124 if args.split_column and args.split_column not in fields:
125 raise SafetyError(f"split column is missing: {args.split_column}")
126 if len(args.group_column) > 20:
127 raise SafetyError("at most 20 group columns may be checked")
128 missing_groups = [column for column in args.group_column if column not in fields]
129 if missing_groups:
130 raise SafetyError(f"group columns are missing: {', '.join(missing_groups)}")
131 if args.expected_assembly and not args.expected_assembly.strip():
132 raise SafetyError("expected assembly must not be blank")
134 errors: Counter[str] = Counter()
135 warnings: Counter[str] = Counter()
136 path_counts: Counter[Path] = Counter()
137 assembly_counts: Counter[str] = Counter()
138 group_splits: dict[str, dict[str, set[str]]] = {
139 column: defaultdict(set) for column in args.group_column
140 }
141 missing_group_values: Counter[str] = Counter()
142 split_missing = 0
143 total_bytes = 0
144 file_summaries: list[dict] = []
145 checksum_counts: Counter[str] = Counter()
147 if args.assembly_column not in fields:
148 warnings["assembly_column_missing"] += 1
150 for row_index, row in enumerate(rows, start=1):
151 raw_path = row.get(args.path_column, "")
152 if not raw_path:
153 errors["blank_path"] += 1
154 continue
155 try:
156 bed_path = _manifest_local_path(raw_path, manifest)
157 except (OSError, SafetyError):
158 errors["invalid_or_missing_local_path"] += 1
159 continue
161 path_counts[bed_path] += 1
162 size = bed_path.lstat().st_size
163 if size > args.max_file_bytes:
164 errors["file_exceeds_byte_limit"] += 1
165 continue
166 total_bytes += size
167 if total_bytes > args.max_total_bytes:
168 raise SafetyError("corpus exceeds max-total-bytes")
169 lowered_name = bed_path.name.lower()
170 if not (lowered_name.endswith(".bed") or lowered_name.endswith(".bed.gz")):
171 warnings["nonstandard_bed_extension"] += 1
173 assembly = row.get(args.assembly_column, "") if args.assembly_column in fields else ""
174 if assembly:
175 if len(assembly) > 200:
176 errors["assembly_value_too_long"] += 1
177 else:
178 assembly_counts[assembly] += 1
179 if args.expected_assembly and assembly != args.expected_assembly:
180 errors["unexpected_assembly"] += 1
181 elif args.assembly_column in fields:
182 errors["blank_assembly"] += 1
184 split = row.get(args.split_column, "") if args.split_column else ""
185 if args.split_column and not split:
186 split_missing += 1
187 for group_column in args.group_column:
188 group_value = row.get(group_column, "")
189 if not group_value:
190 missing_group_values[group_column] += 1
191 elif args.split_column and split:
192 group_splits[group_column][group_value].add(split)
194 checksum: str | None = None
195 if args.checksums:
196 checksum, _ = sha256_file(bed_path, max_bytes=args.max_file_bytes)
197 checksum_counts[checksum] += 1
198 if len(file_summaries) < args.max_output_files:
199 summary = {
200 "path": display_path(bed_path, row_index, args.path_mode),
201 "size_bytes": size,
202 }
203 if checksum:
204 summary["sha256"] = checksum
205 file_summaries.append(summary)
207 duplicate_path_rows = sum(count - 1 for count in path_counts.values() if count > 1)
208 if duplicate_path_rows:
209 warnings["duplicate_path_rows"] = duplicate_path_rows
210 if len(assembly_counts) > 1:
211 errors["mixed_assemblies"] = len(assembly_counts)
212 if not assembly_counts:
213 warnings["assembly_not_recorded"] += 1
214 if split_missing:
215 errors["blank_split"] = split_missing
216 for column, count in missing_group_values.items():
217 if count:
218 errors[f"blank_group:{column}"] = count
220 leakage: dict[str, dict[str, int]] = {}
221 for column, groups in group_splits.items():
222 crossing = sum(1 for splits in groups.values() if len(splits) > 1)
223 leakage[column] = {
224 "distinct_groups": len(groups),
225 "groups_crossing_splits": crossing,
226 }
227 if crossing:
228 errors[f"group_leakage:{column}"] = crossing
230 duplicate_content_groups = 0
231 duplicate_content_files = 0
232 if args.checksums:
233 duplicate_content_groups = sum(1 for count in checksum_counts.values() if count > 1)
234 duplicate_content_files = sum(
235 count for count in checksum_counts.values() if count > 1
236 )
237 if duplicate_content_groups:
238 warnings["duplicate_content_groups"] = duplicate_content_groups
240 manifest_digest, manifest_size = sha256_file(
241 manifest,
242 max_bytes=args.max_manifest_bytes,
243 )
244 report = {
245 "ok": not errors,
246 "tool": TOOL,
247 "contract": {
248 "network_used": False,
249 "metadata_values_emitted": False,
250 "symlinks_allowed": False,
251 },
252 "manifest": {
253 "path": display_path(manifest, 0, args.path_mode),
254 "sha256": manifest_digest,
255 "size_bytes": manifest_size,
256 "delimiter": delimiter,
257 "column_count": len(fields),
258 "column_names_emitted": False,
259 },
260 "summary": {
261 "manifest_rows": len(rows),
262 "valid_local_file_rows": sum(path_counts.values()),
263 "unique_local_files": len(path_counts),
264 "total_input_bytes": total_bytes,
265 "assembly_count": len(assembly_counts),
266 "assemblies": (
267 sorted(assembly_counts)[:20]
268 if args.path_mode == "full"
269 else "<redacted>"
270 ),
271 "duplicate_path_rows": duplicate_path_rows,
272 "duplicate_content_groups": duplicate_content_groups,
273 "duplicate_content_files": duplicate_content_files,
274 "file_summaries_omitted": max(
275 0,
276 sum(path_counts.values()) - len(file_summaries),
277 ),
278 },
279 "leakage_checks": leakage,
280 "errors": dict(sorted(errors.items())),
281 "warnings": dict(sorted(warnings.items())),
282 "files": file_summaries,
283 "recommendations": [
284 "keep every patient/donor/biological replicate in exactly one split",
285 "build universes and fit preprocessing on training groups only",
286 "validate each BED against one checksummed chromosome-sizes file",
287 "store full metadata only in the protected project manifest",
288 ],
289 }
290 return report, 0 if not errors else 2
293def main() -> int:
294 args = build_parser().parse_args()
295 try:
296 report, exit_code = audit(args)
297 print_json(report)
298 return exit_code
299 except (OSError, SafetyError, UnicodeError) as exc:
300 return fail_json(TOOL, exc)
303if __name__ == "__main__":
304 sys.exit(main())