| """Unified sample schema (spec §3.2) and its validator (spec §3.2, tests/test_corpus_schema.py).""" |
| from __future__ import annotations |
|
|
| REQUIRED_COLUMNS = [ |
| "sample_id", "source", "project", "cve", "cwe", "commit_id", |
| "func_vuln", "func_patch", "diff", "patch_lines_vuln", |
| "commit_date", "target", "pair_id", |
| ] |
|
|
|
|
| def validate_corpus(rows: list[dict]) -> dict: |
| """Returns a dict of violation lists (empty lists = that check passed). |
| Does not raise — callers (tests, the build script) decide what's fatal.""" |
| violations: dict[str, list] = { |
| "null_func_vuln": [], |
| "missing_patch_lines_for_target1": [], |
| "bad_pair_integrity": [], |
| "unparseable_commit_date": [], |
| "duplicate_sample_id": [], |
| } |
|
|
| seen_ids: set[str] = set() |
| for r in rows: |
| if not r.get("func_vuln"): |
| violations["null_func_vuln"].append(r["sample_id"]) |
| if r["target"] == 1 and not r.get("patch_lines_vuln"): |
| violations["missing_patch_lines_for_target1"].append(r["sample_id"]) |
| if r["sample_id"] in seen_ids: |
| violations["duplicate_sample_id"].append(r["sample_id"]) |
| seen_ids.add(r["sample_id"]) |
| cd = r.get("commit_date") |
| if cd is not None and not _looks_like_date(cd): |
| violations["unparseable_commit_date"].append((r["sample_id"], cd)) |
|
|
| by_pair: dict[str, list[dict]] = {} |
| for r in rows: |
| by_pair.setdefault(r["pair_id"], []).append(r) |
| for pair_id, pair_rows in by_pair.items(): |
| targets = sorted(r["target"] for r in pair_rows) |
| if len(pair_rows) != 2 or targets != [0, 1]: |
| violations["bad_pair_integrity"].append(pair_id) |
|
|
| return violations |
|
|
|
|
| def _looks_like_date(value: str) -> bool: |
| import re |
|
|
| return bool(re.match(r"^\d{4}-\d{2}-\d{2}", str(value))) |
|
|
