← Files EW AI FlowchartARCHIVED FILE

scripts/validate_knowledge_json.py

4.12 KB · Nov 5, 2026 · 00:44 UTC

↓ Download file

See the change to this file →

#!/usr/bin/env python3
"""Validate EW technical knowledge contracts without executing source content."""
from __future__ import annotations
import argparse
import json
import re
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SCHEMA = ROOT / "docs/contract-knowledge/technical-knowledge.schema.json"
ID_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._:-]*$")

def validate(doc: object) -> list[str]:
    errors = []
    if not isinstance(doc, dict):
        return ["$: expected object"]
    schema = json.loads(SCHEMA.read_text(encoding="utf-8"))
    # Minimal deterministic structural checks; not a full JSON Schema engine.
    for key in schema["required"]:
        if key not in doc:
            errors.append(f"$.{key}: required")
    if doc.get("schemaVersion") != "ew-knowledge-0.1":
        errors.append("$.schemaVersion: unsupported version")
    if doc.get("contractType") != "technical_knowledge":
        errors.append("$.contractType: must be technical_knowledge")
    for key in ("knowledgeId", "title"):
        if not isinstance(doc.get(key), str) or not doc[key].strip():
            errors.append(f"$.{key}: nonempty string required")
    if isinstance(doc.get("knowledgeId"), str) and not ID_RE.fullmatch(doc["knowledgeId"]):
        errors.append("$.knowledgeId: invalid stable ID")
    source = doc.get("source")
    if not isinstance(source, dict) or not all(isinstance(source.get(k), str) and source[k].strip() for k in ("documentId", "title")):
        errors.append("$.source: documentId and title required")
    review = doc.get("review")
    if not isinstance(review, dict) or review.get("status") not in {"draft","pending_review","approved","revoked","superseded"} or not isinstance(review.get("version"), str) or not review["version"]:
        errors.append("$.review: status and version required")
    claims = doc.get("claims")
    if not isinstance(claims, list) or not claims:
        errors.append("$.claims: at least one claim required")
        return errors
    seen = set()
    for i, claim in enumerate(claims):
        p = f"$.claims[{i}]"
        if not isinstance(claim, dict):
            errors.append(f"{p}: object required")
            continue
        cid = claim.get("claimId")
        if not isinstance(cid, str) or not cid.strip() or cid in seen:
            errors.append(f"{p}.claimId: missing or duplicate")
        seen.add(cid)
        if not isinstance(claim.get("statement"), str) or not claim["statement"].strip():
            errors.append(f"{p}.statement: required")
        status = claim.get("verificationStatus")
        if status not in {"unverified","reviewed","rejected"}:
            errors.append(f"{p}.verificationStatus: invalid")
        anchors = claim.get("sourceAnchors")
        if not isinstance(anchors, list):
            errors.append(f"{p}.sourceAnchors: array required")
            anchors = []
        if status == "reviewed" and not anchors:
            errors.append(f"{p}.sourceAnchors: reviewed claim needs source evidence")
        for j, anchor in enumerate(anchors):
            if not isinstance(anchor, dict) or not isinstance(anchor.get("locator"), str) or not anchor["locator"].strip():
                errors.append(f"{p}.sourceAnchors[{j}]: source locator required")
    if isinstance(review, dict) and review.get("status") == "approved":
        if not review.get("reviewerId") or not review.get("approvedAt"):
            errors.append("$.review: approved contract needs reviewerId and approvedAt")
        if any(not isinstance(c, dict) or c.get("verificationStatus") != "reviewed" for c in claims):
            errors.append("$.claims: approved contract cannot contain unreviewed claims")
    return errors

def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("path", type=Path)
    args = ap.parse_args()
    try:
        doc = json.loads(args.path.read_text(encoding="utf-8"))
    except (ValueError, OSError) as exc:
        print(f"ERROR: {exc}")
        return 2
    errors = validate(doc)
    for e in errors:
        print("ERROR:", e)
    print(f"Validation: {len(errors)} error(s)")
    return 1 if errors else 0

if __name__ == "__main__":
    raise SystemExit(main())

SHA-256: 485edf99bf0f55f1dab9a42b0364a778bb453559a26101e21339aefc89fb2567