#!/usr/bin/env python3 """Validate semantic classifications and render a self-contained HTML review.""" from __future__ import annotations import argparse import hashlib import json import os import re import sys import tempfile from pathlib import Path from typing import Any SCHEMA_VERSION = 1 COLLECTOR_NAME = "semantic-diff-review/collect_changes.py" HUNK_ID = re.compile(r"^H-[0-9A-F]{12,64}(?:-[0-9A-F]{8})?(?:-[0-9]+)?$") RISK_LEVELS = {"low", "medium", "high"} CLASSIFICATION_KEYS = {"schema_version", "groups"} GROUP_KEYS = { "title", "purpose", "risk", "review_points", "suggested_commit_message", "hunk_ids", } RISK_KEYS = {"level", "rationale"} class RenderError(RuntimeError): """Raised when evidence or semantic classification is invalid.""" def load_json(path: Path) -> Any: try: with path.open("r", encoding="utf-8") as handle: return json.load(handle) except FileNotFoundError as exc: raise RenderError(f"File not found: {path}") from exc except json.JSONDecodeError as exc: raise RenderError(f"Invalid JSON in {path}: {exc}") from exc def require_dict(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict): raise RenderError(f"{label} must be an object") return value def require_exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None: actual = set(value) missing = sorted(expected - actual) unknown = sorted(actual - expected) if missing or unknown: details = [] if missing: details.append(f"missing {', '.join(missing)}") if unknown: details.append(f"unknown {', '.join(unknown)}") raise RenderError(f"{label} has invalid fields: {'; '.join(details)}") def require_string(value: Any, label: str, *, allow_empty: bool = False) -> str: if not isinstance(value, str): raise RenderError(f"{label} must be a string") if not allow_empty and not value.strip(): raise RenderError(f"{label} must not be empty") return value def canonical_evidence(records: list[dict[str, Any]]) -> str: return json.dumps(records, ensure_ascii=True, sort_keys=True, separators=(",", ":")) def validate_changes(document: Any) -> tuple[dict[str, Any], list[dict[str, Any]]]: root = require_dict(document, "changes") if root.get("schema_version") != SCHEMA_VERSION: raise RenderError("Unsupported changes schema_version") if root.get("generator") != COLLECTOR_NAME: raise RenderError("changes.json was not produced by the bundled collector") repository = require_dict(root.get("repository"), "changes.repository") require_string(repository.get("root"), "changes.repository.root") head = repository.get("head") if head is not None: require_string(head, "changes.repository.head") target_value = repository.get("target") if target_value is None: target = {"kind": "working-tree"} repository = {**repository, "target": target} else: target = require_dict(target_value, "changes.repository.target") kind = require_string(target.get("kind"), "changes.repository.target.kind") if kind == "working-tree": require_exact_keys(target, {"kind"}, "changes.repository.target") elif kind == "commit": require_exact_keys( target, {"kind", "revision", "commit", "base"}, "changes.repository.target", ) require_string(target["revision"], "changes.repository.target.revision") require_string(target["commit"], "changes.repository.target.commit") require_string(target["base"], "changes.repository.target.base") else: raise RenderError( "changes.repository.target.kind must be working-tree or commit" ) records = root.get("hunks") if not isinstance(records, list): raise RenderError("changes.hunks must be an array") seen: set[str] = set() validated: list[dict[str, Any]] = [] required_fields = { "id", "scope", "status", "old_path", "new_path", "kind", "header", "additions", "deletions", "patch_sha256", "patch", } for index, raw_record in enumerate(records): label = f"changes.hunks[{index}]" record = require_dict(raw_record, label) require_exact_keys(record, required_fields, label) hunk_id = require_string(record["id"], f"{label}.id") if not HUNK_ID.fullmatch(hunk_id): raise RenderError(f"{label}.id is not a valid collector hunk ID") if hunk_id in seen: raise RenderError(f"Duplicate collected hunk ID: {hunk_id}") seen.add(hunk_id) scope = require_string(record["scope"], f"{label}.scope") if scope not in {"staged", "unstaged", "untracked", "commit"}: raise RenderError( f"{label}.scope must be staged, unstaged, untracked, or commit" ) require_string(record["status"], f"{label}.status") require_string(record["old_path"], f"{label}.old_path") require_string(record["new_path"], f"{label}.new_path") kind = require_string(record["kind"], f"{label}.kind") if kind not in {"text", "binary-or-metadata", "empty"}: raise RenderError(f"{label}.kind is invalid") require_string(record["header"], f"{label}.header", allow_empty=True) for stat in ("additions", "deletions"): if not isinstance(record[stat], int) or record[stat] < 0: raise RenderError(f"{label}.{stat} must be a non-negative integer") patch = require_string(record["patch"], f"{label}.patch", allow_empty=True) expected_hash = require_string( record["patch_sha256"], f"{label}.patch_sha256" ) actual_hash = hashlib.sha256( patch.encode("utf-8", "surrogateescape") ).hexdigest() if actual_hash != expected_hash: raise RenderError( f"Collected patch integrity check failed for {hunk_id}; re-run collection" ) validated.append(record) digest = require_string(root.get("evidence_sha256"), "changes.evidence_sha256") actual_digest = hashlib.sha256(canonical_evidence(validated).encode("ascii")).hexdigest() if digest != actual_digest: raise RenderError("Collected evidence integrity check failed; re-run collection") return repository, validated def validate_classification( document: Any, hunks: list[dict[str, Any]] ) -> list[dict[str, Any]]: root = require_dict(document, "classification") require_exact_keys(root, CLASSIFICATION_KEYS, "classification") if root["schema_version"] != SCHEMA_VERSION: raise RenderError("Unsupported classification schema_version") groups = root["groups"] if not isinstance(groups, list): raise RenderError("classification.groups must be an array") known_ids = {hunk["id"] for hunk in hunks} assigned: list[str] = [] validated: list[dict[str, Any]] = [] for index, raw_group in enumerate(groups): label = f"classification.groups[{index}]" group = require_dict(raw_group, label) require_exact_keys(group, GROUP_KEYS, label) title = require_string(group["title"], f"{label}.title") purpose = require_string(group["purpose"], f"{label}.purpose") risk = require_dict(group["risk"], f"{label}.risk") require_exact_keys(risk, RISK_KEYS, f"{label}.risk") level = require_string(risk["level"], f"{label}.risk.level").lower() if level not in RISK_LEVELS: raise RenderError(f"{label}.risk.level must be low, medium, or high") rationale = require_string(risk["rationale"], f"{label}.risk.rationale") points = group["review_points"] if not isinstance(points, list) or not points: raise RenderError(f"{label}.review_points must be a non-empty array") review_points = [ require_string(point, f"{label}.review_points[{point_index}]") for point_index, point in enumerate(points) ] message = require_string( group["suggested_commit_message"], f"{label}.suggested_commit_message" ) hunk_ids = group["hunk_ids"] if not isinstance(hunk_ids, list) or not hunk_ids: raise RenderError(f"{label}.hunk_ids must be a non-empty array") normalized_ids = [ require_string(hunk_id, f"{label}.hunk_ids[{hunk_index}]") for hunk_index, hunk_id in enumerate(hunk_ids) ] unknown = sorted(set(normalized_ids) - known_ids) if unknown: raise RenderError(f"{label} references unknown hunk IDs: {', '.join(unknown)}") assigned.extend(normalized_ids) validated.append( { "id": f"group-{index + 1}", "title": title, "purpose": purpose, "risk": {"level": level, "rationale": rationale}, "review_points": review_points, "suggested_commit_message": message, "hunk_ids": normalized_ids, } ) if not known_ids and groups: raise RenderError("classification.groups must be empty when there are no hunks") duplicates = sorted({item for item in assigned if assigned.count(item) > 1}) if duplicates: raise RenderError(f"Hunk IDs assigned more than once: {', '.join(duplicates)}") missing = sorted(known_ids - set(assigned)) if missing: raise RenderError(f"Unclassified hunk IDs: {', '.join(missing)}") return validated def build_payload( repository: dict[str, Any], hunks: list[dict[str, Any]], groups: list[dict[str, Any]], ) -> dict[str, Any]: by_id = {hunk["id"]: hunk for hunk in hunks} rendered_groups = [] for group in groups: group_hunks = [by_id[hunk_id] for hunk_id in group["hunk_ids"]] paths = sorted( { hunk["new_path"] if hunk["new_path"] != "/dev/null" else hunk["old_path"] for hunk in group_hunks } ) rendered_groups.append( { **group, "hunks": group_hunks, "stats": { "additions": sum(hunk["additions"] for hunk in group_hunks), "deletions": sum(hunk["deletions"] for hunk in group_hunks), "files": len(paths), "hunks": len(group_hunks), }, } ) root = repository["root"] return { "repository": { "name": Path(root).name or root, "root": root, "head": repository.get("head"), "target": repository["target"], }, "totals": { "groups": len(rendered_groups), "hunks": len(hunks), "additions": sum(hunk["additions"] for hunk in hunks), "deletions": sum(hunk["deletions"] for hunk in hunks), }, "groups": rendered_groups, } def safe_json_for_html(payload: dict[str, Any]) -> str: encoded = json.dumps(payload, ensure_ascii=True, separators=(",", ":")) return encoded.replace("<", "\\u003c").replace(">", "\\u003e").replace("&", "\\u0026") HTML_TEMPLATE = r''' Semantic Diff Review
Δ
Semantic Diff Review
0 groups 0 hunks +0 −0 Git-derived patches
''' def render_html(payload: dict[str, Any]) -> str: return HTML_TEMPLATE.replace("__REVIEW_DATA__", safe_json_for_html(payload)) def atomic_write(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False, ) as handle: temp_path = Path(handle.name) handle.write(content) handle.flush() os.fsync(handle.fileno()) os.replace(temp_path, path) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--changes", default=".semantic-review/changes.json", help="Collector JSON input", ) parser.add_argument( "--classification", default=".semantic-review/classification.json", help="Semantic classification JSON input", ) parser.add_argument( "--output", default=".semantic-review/review.html", help="Self-contained HTML output", ) return parser.parse_args() def main() -> int: args = parse_args() changes_path = Path(args.changes).expanduser().resolve() classification_path = Path(args.classification).expanduser().resolve() output_path = Path(args.output).expanduser().resolve() try: repository, hunks = validate_changes(load_json(changes_path)) groups = validate_classification(load_json(classification_path), hunks) atomic_write(output_path, render_html(build_payload(repository, hunks, groups))) except (OSError, RenderError) as exc: print(f"error: {exc}", file=sys.stderr) return 1 print(f"Rendered {len(groups)} groups and {len(hunks)} hunks -> {output_path}") return 0 if __name__ == "__main__": raise SystemExit(main())