File size: 1,842 Bytes
5bfc214 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | """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)))
|