951 lines
34 KiB
Python
Executable File
951 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Semantic Impact Report Generator v4 — Three-Layer Architecture
|
|
|
|
Layer 1 — Cargo.toml Dependency Graph (auto)
|
|
Scans workspace Cargo.toml files to build crate dependency DAG.
|
|
Assigns blast radius automatically: leaf crates = LOW, hub crates = HIGH.
|
|
|
|
Layer 2 — git diff API Analysis
|
|
Parses git diff output for public API changes (+pub fn, -pub struct, etc.).
|
|
Classifies each change as ADD (safe), REMOVE (breaking), CHANGE (potentially breaking).
|
|
|
|
Layer 3 — cargo-semver-checks Integration (optional, --semver)
|
|
Runs cargo semver-checks on changed crates for definitive semver analysis.
|
|
|
|
Usage:
|
|
# Pre-commit (staged changes)
|
|
python3 logs/generate_semantic_report.py --mode staged
|
|
|
|
# Post-commit (last commit)
|
|
python3 logs/generate_semantic_report.py --mode last
|
|
|
|
# With semver check (slow)
|
|
python3 logs/generate_semantic_report.py --mode staged --semver
|
|
|
|
# Skip Layer 2 API analysis
|
|
python3 logs/generate_semantic_report.py --mode staged --no-api
|
|
|
|
Exit codes:
|
|
0 = safe (LOW/MEDIUM impact)
|
|
1 = HIGH risk found — review required
|
|
|
|
Dependencies: Python 3.8+, tomllib (stdlib 3.11+) or tomli (fallback)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# CONFIG
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
WORKSPACE_CARGO = REPO_ROOT / "Cargo.toml"
|
|
MANIFEST_PATH = REPO_ROOT / "project_manifest.json"
|
|
LOG_DIR = REPO_ROOT / "logs"
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# TOML parser — compatible with Python 3.8+
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
try:
|
|
import tomllib
|
|
except ImportError:
|
|
import tomli as tomllib
|
|
|
|
|
|
# ============================================================================
|
|
# LAYER 1: Cargo.toml Dependency Graph
|
|
# ============================================================================
|
|
|
|
def parse_cargo_toml(path: Path) -> dict:
|
|
"""Parse a Cargo.toml file and return as dict."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
return tomllib.load(f)
|
|
except Exception as e:
|
|
print(f" [WARN] Failed to parse {path}: {e}", file=sys.stderr)
|
|
return {}
|
|
|
|
|
|
def build_dependency_graph() -> dict:
|
|
"""
|
|
Build the full workspace dependency graph from Cargo.toml files.
|
|
|
|
Returns:
|
|
{
|
|
"crates": { "hcie-protocol": { "path": "hcie-protocol",
|
|
"deps": [],
|
|
"dependents": ["hcie-color", "hcie-blend", ...] } },
|
|
"all_files": [ "hcie-protocol/src/lib.rs", ... ],
|
|
"file_to_crate": { "hcie-protocol/src/lib.rs": "hcie-protocol", ... },
|
|
"blast_radius": { "hcie-protocol": "HIGH", "hcie-color": "LOW", ... },
|
|
"errors": []
|
|
}
|
|
"""
|
|
workspace = parse_cargo_toml(WORKSPACE_CARGO)
|
|
members = workspace.get("workspace", {}).get("members", [])
|
|
errors = []
|
|
|
|
# 1. Discover all crate directories
|
|
crate_paths = {}
|
|
for member in members:
|
|
member_path = REPO_ROOT / member
|
|
cargo_path = member_path / "Cargo.toml"
|
|
if cargo_path.exists():
|
|
data = parse_cargo_toml(cargo_path)
|
|
name = data.get("package", {}).get("name", Path(member).name)
|
|
crate_paths[name] = str(member_path.relative_to(REPO_ROOT))
|
|
else:
|
|
errors.append(f"Member path missing Cargo.toml: {member}")
|
|
|
|
if not crate_paths:
|
|
return {"crates": {}, "all_files": [], "file_to_crate": {},
|
|
"blast_radius": {}, "errors": ["No crates found in workspace"]}
|
|
|
|
# 2. Parse each crate's dependencies
|
|
crates = {}
|
|
for name, rel_path in crate_paths.items():
|
|
cargo_path = REPO_ROOT / rel_path / "Cargo.toml"
|
|
data = parse_cargo_toml(cargo_path)
|
|
deps = data.get("dependencies", {})
|
|
|
|
internal_deps = []
|
|
for dep_name, dep_info in deps.items():
|
|
if isinstance(dep_info, dict) and "path" in dep_info:
|
|
internal_deps.append(dep_name)
|
|
|
|
crates[name] = {
|
|
"path": rel_path,
|
|
"deps": internal_deps,
|
|
"dependents": [],
|
|
}
|
|
|
|
# 3. Build reverse dependency list (who depends on whom)
|
|
for name, info in crates.items():
|
|
for dep_name in info["deps"]:
|
|
if dep_name in crates:
|
|
crates[dep_name]["dependents"].append(name)
|
|
|
|
# 4. Discover all source files — .rs per crate
|
|
all_files = []
|
|
file_to_crate = {}
|
|
for name, info in crates.items():
|
|
src_dir = REPO_ROOT / info["path"] / "src"
|
|
if src_dir.exists():
|
|
for rs_file in sorted(src_dir.rglob("*.rs")):
|
|
rel = rs_file.relative_to(REPO_ROOT).as_posix()
|
|
all_files.append(rel)
|
|
file_to_crate[rel] = name
|
|
|
|
# 5. Auto-assign blast radius based on dependency hub weight
|
|
blast_radius = {}
|
|
for name, info in crates.items():
|
|
num_dependents = len(info["dependents"])
|
|
if num_dependents >= 5:
|
|
blast_radius[name] = "HIGH"
|
|
elif num_dependents >= 2:
|
|
blast_radius[name] = "MEDIUM"
|
|
elif num_dependents == 0 and any(name in c["deps"] for c in crates.values()):
|
|
blast_radius[name] = "LOW"
|
|
else:
|
|
blast_radius[name] = "LOW"
|
|
|
|
# 6. Validate existing manifest entries for file-level info
|
|
manifest_annotations = {}
|
|
if MANIFEST_PATH.exists():
|
|
try:
|
|
with open(MANIFEST_PATH) as f:
|
|
manifest = json.load(f)
|
|
for entry in manifest.get("files", []):
|
|
fp = entry["file_path"]
|
|
manifest_annotations[fp] = {
|
|
"description": entry.get("description", ""),
|
|
"blast_radius": entry.get("blast_radius", "LOW"),
|
|
"semantic_dependencies": entry.get("semantic_dependencies", []),
|
|
}
|
|
except Exception as e:
|
|
errors.append(f"Failed to read manifest: {e}")
|
|
|
|
return {
|
|
"crates": crates,
|
|
"all_files": all_files,
|
|
"file_to_crate": file_to_crate,
|
|
"blast_radius": blast_radius,
|
|
"manifest_annotations": manifest_annotations,
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
def get_file_blast_radius(file_path: str, graph: dict) -> str:
|
|
"""Determine blast radius for a file based on crate-level + manifest annotations."""
|
|
ann = graph.get("manifest_annotations", {}).get(file_path, {})
|
|
if ann and ann.get("blast_radius", "LOW") != "LOW":
|
|
return ann["blast_radius"]
|
|
|
|
crate = graph.get("file_to_crate", {}).get(file_path)
|
|
if crate:
|
|
return graph.get("blast_radius", {}).get(crate, "LOW")
|
|
return "LOW"
|
|
|
|
|
|
def get_file_crate(file_path: str, graph: dict) -> str:
|
|
"""Return crate name for a given file path."""
|
|
return graph.get("file_to_crate", {}).get(file_path, "")
|
|
|
|
|
|
# ============================================================================
|
|
# LAYER 2: git diff API Analysis
|
|
# ============================================================================
|
|
|
|
PUB_ADD_PATTERNS = [
|
|
re.compile(r'^\+pub\s+(fn|struct|enum|trait|type|const|static|mod|use|macro_rules!)\s+(\w+)'),
|
|
re.compile(r'^\+pub\s+(unsafe\s+)?fn\s+(\w+)'),
|
|
re.compile(r'^\+pub\s+(unsafe\s+)?trait\s+(\w+)'),
|
|
]
|
|
|
|
PUB_REMOVE_PATTERNS = [
|
|
re.compile(r'^\-pub\s+(fn|struct|enum|trait|type|const|static|mod|use|macro_rules!)\s+(\w+)'),
|
|
re.compile(r'^\-pub\s+(unsafe\s+)?fn\s+(\w+)'),
|
|
re.compile(r'^\-pub\s+(unsafe\s+)?trait\s+(\w+)'),
|
|
]
|
|
|
|
HUNK_CONTEXT_PATTERN = re.compile(
|
|
r'@@\s+-(\d+),?\d*\s+\+(\d+),?\d*\s+@@\s*(.*)'
|
|
)
|
|
PUB_CONTEXT_PATTERNS = [
|
|
re.compile(r'pub\s+(fn|struct|enum|trait|type|const|mod)\s+(\w+)'),
|
|
re.compile(r'pub\s+(unsafe\s+)?(fn|trait)\s+(\w+)'),
|
|
]
|
|
|
|
ENUM_VARIANT_LINE = re.compile(r'^\s*(\w+)\s*(?:\(.*\))?\s*[,{]?\s*$')
|
|
PUB_FIELD_LINE = re.compile(r'^\s+pub\s+(\w+)\s*:')
|
|
|
|
PRIV_PATTERNS = [
|
|
re.compile(r'^[+-]\s+(?!pub\b)'),
|
|
]
|
|
|
|
IGNORE_PATTERNS = [
|
|
re.compile(r'^[+-]\s*$'),
|
|
re.compile(r'^[+-]\s*(//|#\[)'),
|
|
re.compile(r'^[+-]\s*\}'),
|
|
re.compile(r'^[+-]\s*use\s+'),
|
|
]
|
|
|
|
|
|
class APIAnalysis:
|
|
def __init__(self):
|
|
self.added = []
|
|
self.removed = []
|
|
self.changed = []
|
|
self.field_changes = []
|
|
self.variant_additions = []
|
|
self.priv_changes = 0
|
|
self.total_hunks = 0
|
|
|
|
@property
|
|
def has_breaking_changes(self) -> bool:
|
|
return bool(self.removed or self.changed)
|
|
|
|
@property
|
|
def has_safe_changes(self) -> bool:
|
|
return bool(self.added or self.variant_additions or self.field_changes)
|
|
|
|
@property
|
|
def summary(self) -> str:
|
|
parts = []
|
|
if self.added:
|
|
parts.append(f"+{len(self.added)} pub item")
|
|
if self.variant_additions:
|
|
parts.append(f"+{len(self.variant_additions)} enum var")
|
|
if self.field_changes:
|
|
parts.append(f"+{len(self.field_changes)} struct field")
|
|
if self.removed:
|
|
parts.append(f"-{len(self.removed)} pub (BREAKING)")
|
|
if self.changed:
|
|
parts.append(f"~{len(self.changed)} sig change")
|
|
if self.priv_changes:
|
|
parts.append(f"{self.priv_changes} internal")
|
|
return ", ".join(parts) if parts else "no API changes"
|
|
|
|
|
|
def analyze_git_diff(file_path: str, mode: str = "staged") -> APIAnalysis:
|
|
analysis = APIAnalysis()
|
|
|
|
if not (file_path.endswith(".rs") or file_path.endswith(".svelte") or file_path.endswith(".ts")):
|
|
return analysis
|
|
|
|
try:
|
|
if mode == "last":
|
|
cmd = ["git", "diff", "HEAD~1", "HEAD", "--", file_path]
|
|
elif mode == "staged":
|
|
cmd = ["git", "diff", "--cached", "--", file_path]
|
|
else:
|
|
cmd = ["git", "diff", "--", file_path]
|
|
result = subprocess.run(
|
|
cmd, capture_output=True, text=True, cwd=str(REPO_ROOT),
|
|
)
|
|
diff_text = result.stdout
|
|
except Exception:
|
|
return analysis
|
|
|
|
if not diff_text.strip():
|
|
return analysis
|
|
|
|
current_hunk = []
|
|
in_hunk = False
|
|
for line in diff_text.split("\n"):
|
|
if line.startswith("@@"):
|
|
if current_hunk:
|
|
_classify_hunk(current_hunk, analysis)
|
|
current_hunk = [line]
|
|
in_hunk = True
|
|
elif in_hunk:
|
|
current_hunk.append(line)
|
|
|
|
if current_hunk:
|
|
_classify_hunk(current_hunk, analysis)
|
|
|
|
return analysis
|
|
|
|
|
|
def _classify_hunk(hunk_lines: list, analysis: APIAnalysis):
|
|
analysis.total_hunks += 1
|
|
|
|
header = hunk_lines[0]
|
|
context = ""
|
|
hunk_m = HUNK_CONTEXT_PATTERN.search(header)
|
|
if hunk_m:
|
|
context = hunk_m.group(3).strip()
|
|
|
|
inside_pub_enum = False
|
|
inside_pub_struct = False
|
|
pub_context_name = ""
|
|
for pat in PUB_CONTEXT_PATTERNS:
|
|
cm = pat.search(context)
|
|
if cm:
|
|
groups = cm.groups()
|
|
kind = groups[0] if groups[0] else groups[2]
|
|
name = groups[-1]
|
|
if kind == "enum":
|
|
inside_pub_enum = True
|
|
pub_context_name = name
|
|
elif kind == "struct":
|
|
inside_pub_struct = True
|
|
pub_context_name = name
|
|
break
|
|
|
|
lines = [l for l in hunk_lines if not any(p.match(l) for p in IGNORE_PATTERNS)]
|
|
|
|
added = [l[1:] for l in lines if l.startswith("+")]
|
|
removed = [l[1:] for l in lines if l.startswith("-")]
|
|
|
|
for line in removed:
|
|
for pat in PUB_REMOVE_PATTERNS:
|
|
m = pat.search(line)
|
|
if m:
|
|
kind = m.group(1) if m.lastindex >= 1 else "item"
|
|
name = m.group(m.lastindex)
|
|
analysis.removed.append(f"{kind} {name}")
|
|
break
|
|
|
|
for line in added:
|
|
for pat in PUB_ADD_PATTERNS:
|
|
m = pat.search(line)
|
|
if m:
|
|
kind = m.group(1) if m.lastindex >= 1 else "item"
|
|
name = m.group(m.lastindex)
|
|
analysis.added.append(f"{kind} {name}")
|
|
break
|
|
|
|
if inside_pub_enum:
|
|
for line in added:
|
|
stripped = line.strip().rstrip(",")
|
|
vm = ENUM_VARIANT_LINE.match(stripped)
|
|
if vm and not stripped.startswith("///") and not stripped.startswith("#["):
|
|
variant = vm.group(1)
|
|
if variant not in ("{", "}", "pub", "enum"):
|
|
analysis.variant_additions.append(
|
|
f"{pub_context_name}::{variant}"
|
|
)
|
|
|
|
if inside_pub_struct:
|
|
for line in added:
|
|
fm = PUB_FIELD_LINE.match(line)
|
|
if fm:
|
|
analysis.field_changes.append(
|
|
f"{pub_context_name}.{fm.group(1)}"
|
|
)
|
|
|
|
for line in lines:
|
|
if any(p.match(line) for p in PRIV_PATTERNS) and not any(p.match(line) for p in IGNORE_PATTERNS):
|
|
analysis.priv_changes += 1
|
|
|
|
added_fns = set()
|
|
removed_fns = set()
|
|
FN_SIG_PATTERN = re.compile(r'^[+-]\s+pub\s+(?:unsafe\s+)?fn\s+(\w+)\s*\(')
|
|
for line in added:
|
|
m = FN_SIG_PATTERN.search(line)
|
|
if m:
|
|
added_fns.add(m.group(1))
|
|
for line in removed:
|
|
m = FN_SIG_PATTERN.search(line)
|
|
if m:
|
|
removed_fns.add(m.group(1))
|
|
|
|
for fn_name in removed_fns & added_fns:
|
|
analysis.changed.append(fn_name)
|
|
if fn_name in analysis.removed:
|
|
analysis.removed = [x for x in analysis.removed if fn_name not in x]
|
|
|
|
|
|
# ============================================================================
|
|
# LAYER 3: cargo-semver-checks Integration
|
|
# ============================================================================
|
|
|
|
def _is_lib_crate(crate_rel: str) -> bool:
|
|
lib_rs = REPO_ROOT / crate_rel / "src" / "lib.rs"
|
|
return lib_rs.exists()
|
|
|
|
|
|
def run_semver_checks(changed_crates: list, graph: dict, mode: str = "last") -> dict:
|
|
results = {}
|
|
|
|
try:
|
|
subprocess.run(["cargo", "semver-checks", "--version"],
|
|
capture_output=True, text=True, check=True)
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
for c in changed_crates:
|
|
results[c] = {"violations": [], "error": "cargo-semver-checks not installed. Run: cargo install cargo-semver-checks"}
|
|
return results
|
|
|
|
nightly = subprocess.run(
|
|
["rustup", "run", "nightly", "rustc", "--version"],
|
|
capture_output=True, text=True,
|
|
)
|
|
if nightly.returncode != 0:
|
|
for c in changed_crates:
|
|
results[c] = {"violations": [], "error": "Nightly Rust required for rustdoc JSON. Run: rustup toolchain install nightly"}
|
|
return results
|
|
|
|
if mode == "last":
|
|
baseline_ref = "HEAD~1"
|
|
else:
|
|
baseline_ref = "HEAD"
|
|
|
|
for crate_name in changed_crates:
|
|
crate_info = graph.get("crates", {}).get(crate_name)
|
|
if not crate_info:
|
|
continue
|
|
crate_rel = crate_info["path"]
|
|
if not _is_lib_crate(crate_rel):
|
|
results[crate_name] = {"violations": [], "error": "skipped (binary crate, no lib.rs)"}
|
|
continue
|
|
results[crate_name] = _check_single_crate(
|
|
crate_name, crate_rel, baseline_ref,
|
|
)
|
|
|
|
return results
|
|
|
|
|
|
def _gen_rustdoc(crate_name: str, crate_path: str):
|
|
subprocess.run(
|
|
["cargo", "+nightly", "rustdoc", "--lib",
|
|
"--manifest-path", os.path.join(crate_path, "Cargo.toml"),
|
|
"--", "-Z", "unstable-options", "--output-format", "json"],
|
|
capture_output=True, text=True, cwd=str(REPO_ROOT),
|
|
timeout=300,
|
|
)
|
|
|
|
|
|
def _check_single_crate(crate_name: str, crate_rel: str,
|
|
baseline_ref: str) -> dict:
|
|
result = {"violations": [], "error": None}
|
|
rustdoc_json_name = crate_name.replace("-", "_")
|
|
crate_path = os.path.join(str(REPO_ROOT), crate_rel)
|
|
|
|
try:
|
|
subprocess.run(
|
|
["git", "checkout", baseline_ref, "--", crate_rel],
|
|
capture_output=True, text=True, cwd=str(REPO_ROOT),
|
|
)
|
|
|
|
_gen_rustdoc(crate_name, crate_path)
|
|
|
|
import shutil
|
|
baseline_json = REPO_ROOT / "target" / "doc" / f"{rustdoc_json_name}.json"
|
|
if not baseline_json.exists():
|
|
baseline_json = REPO_ROOT / "target" / "doc" / crate_name / "index.json"
|
|
|
|
tmp_baseline = Path(tempfile.mktemp(suffix="_baseline.json"))
|
|
if baseline_json.exists():
|
|
shutil.copy(str(baseline_json), str(tmp_baseline))
|
|
|
|
subprocess.run(
|
|
["git", "checkout", "HEAD", "--", crate_rel],
|
|
capture_output=True, text=True, cwd=str(REPO_ROOT),
|
|
)
|
|
|
|
_gen_rustdoc(crate_name, crate_path)
|
|
|
|
current_json = REPO_ROOT / "target" / "doc" / f"{rustdoc_json_name}.json"
|
|
if not current_json.exists():
|
|
current_json = REPO_ROOT / "target" / "doc" / crate_name / "index.json"
|
|
|
|
tmp_current = Path(tempfile.mktemp(suffix="_current.json"))
|
|
if current_json.exists():
|
|
shutil.copy(str(current_json), str(tmp_current))
|
|
|
|
if tmp_baseline.exists() and tmp_current.exists():
|
|
proc = subprocess.run(
|
|
["cargo", "semver-checks",
|
|
"--baseline-rustdoc", str(tmp_baseline),
|
|
"--current-rustdoc", str(tmp_current)],
|
|
capture_output=True, text=True, cwd=str(REPO_ROOT),
|
|
timeout=180,
|
|
)
|
|
output = proc.stdout + proc.stderr
|
|
if proc.returncode != 0:
|
|
result["violations"] = _parse_semver_output(output)
|
|
else:
|
|
parts = []
|
|
if not tmp_baseline.exists():
|
|
parts.append("baseline")
|
|
if not tmp_current.exists():
|
|
parts.append("current")
|
|
result["error"] = f"rustdoc JSON not found: {' + '.join(parts)}"
|
|
|
|
for p in [tmp_baseline, tmp_current]:
|
|
if p.exists():
|
|
p.unlink()
|
|
|
|
except subprocess.TimeoutExpired:
|
|
result["error"] = "cargo-semver-checks timed out (rustdoc compilation too slow)"
|
|
subprocess.run(["git", "checkout", "HEAD", "--", crate_rel],
|
|
capture_output=True, cwd=str(REPO_ROOT))
|
|
except Exception as e:
|
|
result["error"] = str(e).split("\n")[0][:120]
|
|
subprocess.run(["git", "checkout", "HEAD", "--", crate_rel],
|
|
capture_output=True, cwd=str(REPO_ROOT))
|
|
|
|
return result
|
|
|
|
|
|
def _parse_semver_output(output: str) -> list:
|
|
violations = []
|
|
for line in output.split("\n"):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith("[") or line.startswith("Checking") or \
|
|
line.startswith("Comparing") or line.startswith("Detected") or \
|
|
line.startswith("Finished") or line.startswith("Running"):
|
|
continue
|
|
if "error" in line.lower() or "breaking" in line.lower() or \
|
|
"major" in line.lower() or "violation" in line.lower() or \
|
|
"removed" in line.lower() or "changed" in line.lower():
|
|
violations.append(line)
|
|
return violations[:30]
|
|
|
|
|
|
# ============================================================================
|
|
# MAIN REPORT GENERATOR
|
|
# ============================================================================
|
|
|
|
def get_commit_info(mode: str) -> dict:
|
|
result = {
|
|
"commit_id": "",
|
|
"commit_short": "",
|
|
"commit_message": "",
|
|
"previous_commit_id": "",
|
|
}
|
|
|
|
try:
|
|
head = subprocess.check_output(
|
|
["git", "rev-parse", "HEAD"],
|
|
cwd=str(REPO_ROOT), text=True
|
|
).strip()
|
|
result["commit_id"] = head
|
|
result["commit_short"] = head[:7]
|
|
|
|
msg = subprocess.check_output(
|
|
["git", "log", "-1", "--pretty=%s"],
|
|
cwd=str(REPO_ROOT), text=True
|
|
).strip()
|
|
result["commit_message"] = msg
|
|
|
|
if mode == "last":
|
|
try:
|
|
parent = subprocess.check_output(
|
|
["git", "rev-parse", "HEAD~1"],
|
|
cwd=str(REPO_ROOT), text=True
|
|
).strip()
|
|
result["previous_commit_id"] = parent
|
|
except subprocess.CalledProcessError:
|
|
result["previous_commit_id"] = "unknown"
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
def get_changed_files(mode: str) -> list:
|
|
if mode == "staged":
|
|
cmd = ["git", "diff", "--cached", "--name-only"]
|
|
elif mode == "last":
|
|
cmd = ["git", "diff", "HEAD~1", "HEAD", "--name-only"]
|
|
elif mode == "working":
|
|
staged = subprocess.check_output(
|
|
["git", "diff", "--cached", "--name-only"],
|
|
cwd=str(REPO_ROOT), text=True
|
|
).strip().split("\n")
|
|
unstaged = subprocess.check_output(
|
|
["git", "diff", "--name-only"],
|
|
cwd=str(REPO_ROOT), text=True
|
|
).strip().split("\n")
|
|
changed = set(
|
|
s.strip() for s in staged + unstaged if s.strip()
|
|
)
|
|
return sorted(changed)
|
|
else:
|
|
return []
|
|
|
|
try:
|
|
out = subprocess.check_output(cmd, cwd=str(REPO_ROOT), text=True)
|
|
return [l.strip() for l in out.strip().split("\n") if l.strip()]
|
|
except subprocess.CalledProcessError:
|
|
return []
|
|
|
|
|
|
def _reports_dir() -> Path:
|
|
reports = LOG_DIR
|
|
reports.mkdir(parents=True, exist_ok=True)
|
|
return reports
|
|
|
|
|
|
def _save_report(report_text: str, mode: str, has_high_risk: bool, commit_short: str = ""):
|
|
import datetime
|
|
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
risk_tag = "HIGH" if has_high_risk else "LOW"
|
|
suffix = f"_{commit_short}" if commit_short else ""
|
|
filename = f"semantic_impact_{ts}_{risk_tag}_{mode}{suffix}.txt"
|
|
report_path = _reports_dir() / filename
|
|
with open(report_path, "w") as f:
|
|
f.write(report_text)
|
|
return report_path
|
|
|
|
|
|
def generate_report(graph: dict, changed_files: list,
|
|
mode: str, run_api_analysis: bool = True,
|
|
run_semver: bool = False,
|
|
commit_info: dict = None) -> tuple:
|
|
lines = []
|
|
highest_risk = "LOW"
|
|
|
|
lines.append("=== SEMANTIC IMPACT REPORT v4 ===")
|
|
lines.append(f"Mode : {mode}")
|
|
if commit_info and commit_info.get("commit_id"):
|
|
lines.append(f"Commit : {commit_info['commit_short']} — \"{commit_info['commit_message']}\"")
|
|
if mode == "last" and commit_info.get("previous_commit_id"):
|
|
lines.append(f"Range : {commit_info['previous_commit_id'][:7]} → {commit_info['commit_short']}")
|
|
lines.append(f"Files changed : {len(changed_files)}")
|
|
lines.append("")
|
|
|
|
changed_by_crate = defaultdict(list)
|
|
for fp in changed_files:
|
|
crate = get_file_crate(fp, graph)
|
|
changed_by_crate[crate].append(fp)
|
|
|
|
high_crates = []
|
|
med_crates = []
|
|
low_crates = []
|
|
for crate in changed_by_crate:
|
|
br = graph.get("blast_radius", {}).get(crate, "LOW")
|
|
if br == "HIGH":
|
|
high_crates.append(crate)
|
|
elif br == "MEDIUM":
|
|
med_crates.append(crate)
|
|
else:
|
|
low_crates.append(crate)
|
|
|
|
num_high = sum(len(changed_by_crate[c]) for c in high_crates)
|
|
if high_crates:
|
|
highest_risk = "HIGH"
|
|
elif med_crates:
|
|
highest_risk = "MEDIUM"
|
|
|
|
lines.append("[LAYER 1 — CRATE DEPENDENCY IMPACT]")
|
|
lines.append(f"High-radius crates affected : {len(high_crates)} ({num_high} files)")
|
|
lines.append(f"Medium-radius crates : {len(med_crates)}")
|
|
lines.append(f"Low-radius crates : {len(low_crates)}")
|
|
lines.append(f"Overall risk : {highest_risk}")
|
|
lines.append("")
|
|
|
|
if high_crates:
|
|
lines.append("── HIGH-RISK CRATES WITH CHANGES ──")
|
|
for crate in high_crates:
|
|
crate_info = graph.get("crates", {}).get(crate, {})
|
|
files = changed_by_crate[crate]
|
|
lines.append(f" {crate}/ ({len(crate_info.get('dependents', []))} dependents)")
|
|
for fp in files:
|
|
desc = graph.get("manifest_annotations", {}).get(fp, {}).get("description", "")
|
|
if desc:
|
|
lines.append(f" {fp}")
|
|
lines.append(f" → {desc}")
|
|
else:
|
|
lines.append(f" {fp}")
|
|
|
|
deps = crate_info.get("dependents", [])
|
|
if deps:
|
|
lines.append(f" ⤷ Impacts: {', '.join(deps)}")
|
|
lines.append("")
|
|
|
|
if med_crates:
|
|
lines.append("── MEDIUM-RISK CRATES ──")
|
|
for crate in med_crates:
|
|
crate_info = graph.get("crates", {}).get(crate, {})
|
|
files = changed_by_crate[crate]
|
|
lines.append(f" {crate}/ ({len(crate_info.get('dependents', []))} dependents)")
|
|
for fp in files:
|
|
desc = graph.get("manifest_annotations", {}).get(fp, {}).get("description", "")
|
|
if desc:
|
|
lines.append(f" {fp} — {desc}")
|
|
else:
|
|
lines.append(f" {fp}")
|
|
deps = crate_info.get("dependents", [])
|
|
if deps:
|
|
lines.append(f" ⤷ Impacts: {', '.join(deps)}")
|
|
lines.append("")
|
|
|
|
if low_crates:
|
|
lines.append("── LOW-RISK CRATES ──")
|
|
for crate in low_crates:
|
|
for fp in changed_by_crate[crate]:
|
|
lines.append(f" {fp}")
|
|
lines.append("")
|
|
|
|
if run_api_analysis:
|
|
lines.append("[LAYER 2 — API CHANGE ANALYSIS]")
|
|
total_breaking = 0
|
|
total_safe = 0
|
|
|
|
for crate in sorted(changed_by_crate):
|
|
for fp in changed_by_crate[crate]:
|
|
analysis = analyze_git_diff(fp, mode)
|
|
if analysis.total_hunks > 0:
|
|
tag = "BREAKING" if analysis.has_breaking_changes else "safe"
|
|
if analysis.has_breaking_changes:
|
|
total_breaking += 1
|
|
else:
|
|
total_safe += 1
|
|
lines.append(f" {fp}")
|
|
lines.append(f" {tag}: {analysis.summary}")
|
|
if analysis.removed:
|
|
for item in analysis.removed:
|
|
lines.append(f" ❌ REMOVED: {item}")
|
|
if analysis.changed:
|
|
for item in analysis.changed:
|
|
lines.append(f" ⚠️ CHANGED SIG: {item}")
|
|
if analysis.added:
|
|
for item in analysis.added:
|
|
lines.append(f" ✅ ADDED: {item}")
|
|
if analysis.variant_additions:
|
|
for v in analysis.variant_additions:
|
|
lines.append(f" 📦 NEW VARIANT: {v}")
|
|
if analysis.field_changes:
|
|
for f in analysis.field_changes:
|
|
lines.append(f" 🏛️ NEW FIELD: {f}")
|
|
lines.append("")
|
|
|
|
if total_breaking == 0 and total_safe == 0:
|
|
lines.append(" No API-level changes detected (private/internal changes only).")
|
|
lines.append("")
|
|
|
|
if run_semver:
|
|
lines.append("[LAYER 3 — SEMVER CHECK]")
|
|
changed_crate_names = [
|
|
c for c in changed_by_crate
|
|
if c and c in graph.get("crates", {})
|
|
]
|
|
semver_results = run_semver_checks(changed_crate_names, graph, mode)
|
|
for crate, result in semver_results.items():
|
|
if result["error"]:
|
|
lines.append(f" {crate}: ERROR — {result['error']}")
|
|
elif result["violations"]:
|
|
lines.append(f" {crate}: {len(result['violations'])} semver violation(s)")
|
|
for v in result["violations"][:5]:
|
|
lines.append(f" ⚠️ {v}")
|
|
highest_risk = "HIGH"
|
|
else:
|
|
lines.append(f" {crate}: ✅ No semver violations detected")
|
|
lines.append("")
|
|
|
|
unannotated = [
|
|
fp for fp in changed_files
|
|
if fp not in graph.get("manifest_annotations", {})
|
|
and (fp.endswith(".rs") or fp.endswith(".svelte") or fp.endswith(".ts"))
|
|
]
|
|
if unannotated:
|
|
lines.append("[SUGGESTION — Missing Manifest Annotations]")
|
|
lines.append(" These .rs files have no description in project_manifest.json.")
|
|
lines.append(" Consider adding entries for better semantic tracking:")
|
|
for fp in unannotated[:5]:
|
|
crate = get_file_crate(fp, graph)
|
|
br = get_file_blast_radius(fp, graph)
|
|
lines.append(f" {fp} (crate: {crate}, auto-blast: {br})")
|
|
if len(unannotated) > 5:
|
|
lines.append(f" ... and {len(unannotated) - 5} more")
|
|
lines.append("")
|
|
|
|
if graph.get("errors"):
|
|
lines.append("[WARNINGS]")
|
|
for err in graph["errors"]:
|
|
lines.append(f" ⚠️ {err}")
|
|
lines.append("")
|
|
|
|
if highest_risk == "HIGH":
|
|
lines.append("[BLOCKING WARNING]")
|
|
lines.append(" HIGH-risk crate(s) directly modified — review required.")
|
|
if run_api_analysis:
|
|
lines.append(" Check Layer 2 API changes above for breaking changes.")
|
|
lines.append("")
|
|
|
|
lines.append("─" * 60)
|
|
return "\n".join(lines), highest_risk
|
|
|
|
|
|
# ============================================================================
|
|
# CLI
|
|
# ============================================================================
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(
|
|
description="Semantic Impact Report Generator v4",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Pre-commit check (fast — layers 1+2)
|
|
python3 logs/generate_semantic_report.py --mode staged
|
|
|
|
# Post-commit audit
|
|
python3 logs/generate_semantic_report.py --mode last
|
|
|
|
# Full pre-commit with semver check (slow)
|
|
python3 logs/generate_semantic_report.py --mode staged --semver
|
|
|
|
# Quick mode — only dependency graph, no API analysis
|
|
python3 logs/generate_semantic_report.py --mode staged --no-api
|
|
|
|
# Check working tree changes (staged + unstaged)
|
|
python3 logs/generate_semantic_report.py --mode working
|
|
""",
|
|
)
|
|
parser.add_argument(
|
|
"--mode", choices=["staged", "last", "working"],
|
|
default="staged",
|
|
help="Which changes to analyze (default: staged)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-api", action="store_true",
|
|
help="Skip Layer 2 (git diff API analysis)",
|
|
)
|
|
parser.add_argument(
|
|
"--semver", action="store_true",
|
|
help="Enable Layer 3 (cargo-semver-checks, SLOW)",
|
|
)
|
|
parser.add_argument(
|
|
"--save", action="store_true",
|
|
help="Save report to logs/ directory with timestamp",
|
|
)
|
|
parser.add_argument(
|
|
"--json", action="store_true",
|
|
help="Output JSON machine-readable report (for CI/tooling)",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
print(" [Layer 1] Scanning Cargo.toml dependency graph...", file=sys.stderr)
|
|
graph = build_dependency_graph()
|
|
if graph.get("errors"):
|
|
for err in graph["errors"]:
|
|
print(f" [WARN] {err}", file=sys.stderr)
|
|
|
|
num_crates = len(graph.get("crates", {}))
|
|
print(f" → {num_crates} crates, {len(graph['all_files'])} source files", file=sys.stderr)
|
|
|
|
if args.mode == "last":
|
|
print(f" [git] Comparing HEAD~1 → HEAD", file=sys.stderr)
|
|
elif args.mode == "staged":
|
|
print(f" [git] Checking staged changes", file=sys.stderr)
|
|
else:
|
|
print(f" [git] Checking working tree changes", file=sys.stderr)
|
|
|
|
changed_files = get_changed_files(args.mode)
|
|
if not changed_files:
|
|
print(" → No changed files. Nothing to check.", file=sys.stderr)
|
|
sys.exit(0)
|
|
|
|
print(f" → {len(changed_files)} file(s) changed", file=sys.stderr)
|
|
|
|
if not args.no_api:
|
|
print(f" [Layer 2] Analyzing git diff for API changes...", file=sys.stderr)
|
|
else:
|
|
print(f" [Layer 2] Skipped (--no-api)", file=sys.stderr)
|
|
|
|
if args.semver:
|
|
print(f" [Layer 3] Running cargo-semver-checks...", file=sys.stderr)
|
|
else:
|
|
print(f" [Layer 3] Skipped (use --semver to enable)", file=sys.stderr)
|
|
|
|
commit_info = get_commit_info(args.mode)
|
|
|
|
report_text, highest_risk = generate_report(
|
|
graph=graph,
|
|
changed_files=changed_files,
|
|
mode=args.mode,
|
|
run_api_analysis=not args.no_api,
|
|
run_semver=args.semver,
|
|
commit_info=commit_info,
|
|
)
|
|
|
|
if args.json:
|
|
json_out = {
|
|
"version": 4,
|
|
"mode": args.mode,
|
|
"commit_id": commit_info.get("commit_id", ""),
|
|
"commit_short": commit_info.get("commit_short", ""),
|
|
"commit_message": commit_info.get("commit_message", ""),
|
|
"previous_commit_id": commit_info.get("previous_commit_id", ""),
|
|
"files_changed": len(changed_files),
|
|
"highest_risk": highest_risk,
|
|
"high_risk_files": [
|
|
fp for fp in changed_files
|
|
if get_file_blast_radius(fp, graph) == "HIGH"
|
|
],
|
|
"crates_affected": sorted(set(
|
|
get_file_crate(fp, graph) for fp in changed_files
|
|
if get_file_crate(fp, graph)
|
|
)),
|
|
}
|
|
print(json.dumps(json_out, indent=2))
|
|
else:
|
|
print(report_text)
|
|
|
|
if args.save or highest_risk == "HIGH":
|
|
saved_path = _save_report(
|
|
report_text, args.mode, highest_risk == "HIGH",
|
|
commit_short=commit_info.get("commit_short", ""),
|
|
)
|
|
print(f" Report saved: {saved_path}", file=sys.stderr)
|
|
|
|
sys.exit(1 if highest_risk == "HIGH" else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |