Ali Zarinkolah
ac3d545467
Why: - a diagnostic quietly suppressed (noqa, type: ignore, Any, getattr, broad except, config downgrades) is functionally worse than one left failing. Changes: - hook now scans added lines per file/config edit for suppression patterns and reports them, and remembers prior diagnostics so one vanishing in the same edit that added a suppression marker is flagged rather than treated as fixed.
381 lines
13 KiB
Python
381 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Run Python quality tools for Claude Code Write/Edit hooks.
|
|
|
|
The hook is intentionally non-blocking: it formats/fixes what Ruff can fix safely,
|
|
then reports remaining Ruff/ty diagnostics back to Claude as additional context.
|
|
|
|
It also audits each edit for *suppression*: inline ignore comments, typing weakened
|
|
to `Any`, dynamic-attribute escapes, and config-level rule downgrades. The point is
|
|
that silencing a diagnostic can never be quieter than fixing it. Diagnostics are
|
|
remembered between runs, so a diagnostic that disappears in the same edit that
|
|
introduced a suppression marker is reported loudly rather than passing as "clean".
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PYTHON_SUFFIXES = {".py", ".pyi"}
|
|
CONFIG_NAMES = {"pyproject.toml", "ty.toml", ".ty.toml", "ruff.toml", ".ruff.toml", "setup.cfg"}
|
|
SKIP_PARTS = {".git", ".venv", "__pycache__"}
|
|
STATE_RELATIVE = Path(".claude") / "hooks" / ".quality_state.json"
|
|
|
|
# Markers that make a diagnostic go away without necessarily fixing what it found.
|
|
SUPPRESSION_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
|
|
("type-checker suppression", re.compile(r"#\s*(?:ty|type|pyright|mypy)\s*:\s*ignore")),
|
|
("lint suppression", re.compile(r"#\s*noqa\b")),
|
|
("coverage suppression", re.compile(r"#\s*pragma:\s*no\s*cover")),
|
|
("typing weakened to Any", re.compile(r"(?:->|:)\s*(?:typing\.)?Any\b|\bcast\s*\(")),
|
|
("dynamic attribute escape", re.compile(r"\b(?:get|set|has)attr\s*\(")),
|
|
("broad exception swallow", re.compile(r"\bexcept\s+(?:BaseException|Exception)\b")),
|
|
("narrowing assert", re.compile(r"\bassert\s+.+\bis\s+not\s+None\b")),
|
|
)
|
|
|
|
# Config edits that disable checks repo-wide. These files are not Python, so the
|
|
# normal per-file check never runs on them and the effect is otherwise invisible.
|
|
CONFIG_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
|
|
("rule downgraded", re.compile(r'^\s*[\w-]+\s*=\s*"(?:ignore|warn)"')),
|
|
("rule/override block", re.compile(r"^\s*\[\[?tool\.(?:ty|ruff)[\w.]*\]\]?")),
|
|
("lint ignore list", re.compile(r"^\s*(?:ignore|extend-ignore|per-file-ignores)\s*=")),
|
|
("paths excluded from checks", re.compile(r"^\s*(?:exclude|extend-exclude)\s*=")),
|
|
)
|
|
|
|
HUNK_RE = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,\d+)? @@")
|
|
TY_DIAGNOSTIC_RE = re.compile(r"^(?:error|warning)\[([a-z0-9-]+)\]")
|
|
RUFF_DIAGNOSTIC_RE = re.compile(r"^.+?:\d+:\d+:\s+([A-Z]+\d+)\b")
|
|
|
|
|
|
def main() -> int:
|
|
payload = _read_payload()
|
|
file_path = _extract_file_path(payload)
|
|
if file_path is None:
|
|
return 0
|
|
|
|
repo_root = _repo_root()
|
|
path = _resolve_path(file_path, repo_root)
|
|
|
|
if not path.is_file() or not SKIP_PARTS.isdisjoint(path.parts):
|
|
return 0
|
|
|
|
is_python = path.suffix in PYTHON_SUFFIXES
|
|
is_config = not is_python and path.name in CONFIG_NAMES
|
|
if not (is_python or is_config):
|
|
return 0
|
|
|
|
relative_path = _display_path(path, repo_root)
|
|
|
|
if is_config:
|
|
findings = _scan_added_lines(_added_lines(path, repo_root), CONFIG_PATTERNS)
|
|
message = _build_config_message(relative_path=relative_path, findings=findings)
|
|
_emit(message)
|
|
return 0
|
|
|
|
before = _sha256(path)
|
|
|
|
fix = _run(["uv", "run", "ruff", "check", "--fix", str(path)], repo_root)
|
|
fmt = _run(["uv", "run", "ruff", "format", str(path)], repo_root)
|
|
lint = _run(["uv", "run", "ruff", "check", str(path)], repo_root)
|
|
|
|
# `--error-on-warning` makes ty warnings visible to the hook while this
|
|
# script still exits 0, so warnings are reported but do not stop Claude.
|
|
types = _run(["uv", "run", "ty", "check", "--error-on-warning", str(path)], repo_root)
|
|
|
|
after = _sha256(path)
|
|
changed = before != after
|
|
|
|
current_codes = _diagnostic_codes(lint, types)
|
|
previous_codes = _read_previous_codes(repo_root, relative_path)
|
|
_write_current_codes(repo_root, relative_path, current_codes)
|
|
|
|
findings = _scan_added_lines(_added_lines(path, repo_root), SUPPRESSION_PATTERNS)
|
|
|
|
message = _build_message(
|
|
relative_path=relative_path,
|
|
changed=changed,
|
|
results=(fix, fmt, lint, types),
|
|
findings=findings,
|
|
resolved_codes=sorted(previous_codes - current_codes),
|
|
)
|
|
_emit(message)
|
|
|
|
return 0
|
|
|
|
|
|
def _emit(message: str) -> None:
|
|
if not message:
|
|
return
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"suppressOutput": True,
|
|
"hookSpecificOutput": {
|
|
"hookEventName": "PostToolUse",
|
|
"additionalContext": message,
|
|
},
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def _read_payload() -> dict[str, object]:
|
|
try:
|
|
raw = sys.stdin.read()
|
|
return json.loads(raw) if raw.strip() else {}
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
|
|
def _extract_file_path(payload: dict[str, object]) -> str | None:
|
|
response = payload.get("tool_response")
|
|
if isinstance(response, dict):
|
|
for key in ("filePath", "file_path"):
|
|
value = response.get(key)
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
|
|
tool_input = payload.get("tool_input")
|
|
if isinstance(tool_input, dict):
|
|
value = tool_input.get("file_path")
|
|
if isinstance(value, str) and value:
|
|
return value
|
|
|
|
return None
|
|
|
|
|
|
def _repo_root() -> Path:
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--show-toplevel"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if result.returncode == 0:
|
|
return Path(result.stdout.strip()).resolve()
|
|
return Path.cwd().resolve()
|
|
|
|
|
|
def _resolve_path(file_path: str, repo_root: Path) -> Path:
|
|
path = Path(file_path).expanduser()
|
|
if not path.is_absolute():
|
|
path = repo_root / path
|
|
return path.resolve()
|
|
|
|
|
|
def _display_path(path: Path, repo_root: Path) -> str:
|
|
try:
|
|
return str(path.relative_to(repo_root))
|
|
except ValueError:
|
|
return str(path)
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _run(command: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
command,
|
|
cwd=cwd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def _combined_output(result: subprocess.CompletedProcess[str]) -> str:
|
|
return "\n".join(part.strip() for part in (result.stdout, result.stderr) if part.strip())
|
|
|
|
|
|
def _command_label(result: subprocess.CompletedProcess[str]) -> str:
|
|
return " ".join(result.args) if isinstance(result.args, list) else str(result.args)
|
|
|
|
|
|
def _diagnostic_codes(*results: subprocess.CompletedProcess[str]) -> set[str]:
|
|
"""Extract stable rule identifiers (ty rule names, Ruff codes) from tool output."""
|
|
codes: set[str] = set()
|
|
for result in results:
|
|
for line in _combined_output(result).splitlines():
|
|
stripped = line.strip()
|
|
ty_match = TY_DIAGNOSTIC_RE.match(stripped)
|
|
if ty_match:
|
|
codes.add(ty_match.group(1))
|
|
continue
|
|
ruff_match = RUFF_DIAGNOSTIC_RE.match(stripped)
|
|
if ruff_match:
|
|
codes.add(ruff_match.group(1))
|
|
return codes
|
|
|
|
|
|
def _state_path(repo_root: Path) -> Path:
|
|
return repo_root / STATE_RELATIVE
|
|
|
|
|
|
def _read_previous_codes(repo_root: Path, relative_path: str) -> set[str]:
|
|
try:
|
|
state = json.loads(_state_path(repo_root).read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return set()
|
|
entry = state.get(relative_path) if isinstance(state, dict) else None
|
|
if not isinstance(entry, list):
|
|
return set()
|
|
return {code for code in entry if isinstance(code, str)}
|
|
|
|
|
|
def _write_current_codes(repo_root: Path, relative_path: str, codes: set[str]) -> None:
|
|
state_path = _state_path(repo_root)
|
|
try:
|
|
state = json.loads(state_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
state = {}
|
|
if not isinstance(state, dict):
|
|
state = {}
|
|
|
|
if codes:
|
|
state[relative_path] = sorted(codes)
|
|
else:
|
|
state.pop(relative_path, None)
|
|
|
|
try:
|
|
state_path.parent.mkdir(parents=True, exist_ok=True)
|
|
state_path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _added_lines(path: Path, repo_root: Path) -> list[tuple[int, str]]:
|
|
"""Return (line number, text) for lines this working tree adds over HEAD."""
|
|
relative_path = _display_path(path, repo_root)
|
|
|
|
tracked = _run(["git", "ls-files", "--error-unmatch", "--", relative_path], repo_root)
|
|
if tracked.returncode != 0:
|
|
try:
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return []
|
|
return list(enumerate(text.splitlines(), start=1))
|
|
|
|
diff = _run(["git", "diff", "-U0", "HEAD", "--", relative_path], repo_root)
|
|
if diff.returncode != 0:
|
|
return []
|
|
return _parse_added_lines(diff.stdout)
|
|
|
|
|
|
def _parse_added_lines(diff_text: str) -> list[tuple[int, str]]:
|
|
added: list[tuple[int, str]] = []
|
|
lineno = 0
|
|
for line in diff_text.splitlines():
|
|
hunk = HUNK_RE.match(line)
|
|
if hunk:
|
|
lineno = int(hunk.group(1))
|
|
continue
|
|
if line.startswith(("+++", "---")):
|
|
continue
|
|
if line.startswith("+"):
|
|
added.append((lineno, line[1:]))
|
|
lineno += 1
|
|
elif not line.startswith("-"):
|
|
lineno += 1
|
|
return added
|
|
|
|
|
|
def _scan_added_lines(
|
|
added: list[tuple[int, str]],
|
|
patterns: tuple[tuple[str, re.Pattern[str]], ...],
|
|
) -> list[str]:
|
|
findings: list[str] = []
|
|
seen: set[tuple[int, str]] = set()
|
|
for lineno, text in added:
|
|
for category, pattern in patterns:
|
|
if pattern.search(text) and (lineno, category) not in seen:
|
|
seen.add((lineno, category))
|
|
findings.append(f"L{lineno} [{category}]: {text.strip()[:160]}")
|
|
return findings
|
|
|
|
|
|
def _build_message(
|
|
*,
|
|
relative_path: str,
|
|
changed: bool,
|
|
results: tuple[subprocess.CompletedProcess[str], ...],
|
|
findings: list[str],
|
|
resolved_codes: list[str],
|
|
) -> str:
|
|
lines: list[str] = []
|
|
|
|
if changed:
|
|
lines.append(
|
|
f"Python quality hook updated `{relative_path}` with Ruff safe fixes/formatting."
|
|
)
|
|
lines.append("Read the file before making another manual edit to avoid stale text.")
|
|
|
|
for result in results:
|
|
if result.returncode == 0:
|
|
continue
|
|
|
|
output = _combined_output(result) or f"Command exited with status {result.returncode}."
|
|
lines.extend(
|
|
[
|
|
"",
|
|
f"Remaining diagnostics from `{_command_label(result)}`:",
|
|
"```text",
|
|
output[-6000:],
|
|
"```",
|
|
]
|
|
)
|
|
|
|
if findings:
|
|
lines.extend(
|
|
[
|
|
"",
|
|
f"Suppression audit — `{relative_path}` adds these lines over HEAD:",
|
|
"```text",
|
|
"\n".join(findings[-60:]),
|
|
"```",
|
|
"A silenced diagnostic is not a fixed diagnostic. For each entry: either the",
|
|
"code as written requires it, or it exists to make a checker pass — and the",
|
|
"second case hides a bug rather than resolving it. If a checker reported a real",
|
|
"problem, fix the problem. If you believe the checker is wrong, say so to the",
|
|
"user instead of suppressing it yourself.",
|
|
]
|
|
)
|
|
|
|
if resolved_codes:
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"!! These diagnostics vanished in the same edit that added the markers",
|
|
f" above: {', '.join(resolved_codes)}",
|
|
"If any pointed at a genuine logic error, that error is still present and",
|
|
"is now merely unreported. Do NOT call this file clean. Tell the user what",
|
|
"you changed and why each diagnostic no longer applies.",
|
|
]
|
|
)
|
|
|
|
return "\n".join(lines).strip()
|
|
|
|
|
|
def _build_config_message(*, relative_path: str, findings: list[str]) -> str:
|
|
if not findings:
|
|
return ""
|
|
return "\n".join(
|
|
[
|
|
f"Quality-config audit — `{relative_path}` adds these lines over HEAD:",
|
|
"```text",
|
|
"\n".join(findings[-60:]),
|
|
"```",
|
|
"These lines can disable or downgrade checks across the whole repository, and no",
|
|
"per-file check runs on this file. Loosening checker configuration is the user's",
|
|
"decision, not yours: if you changed it to make diagnostics go away, revert it and",
|
|
"report the diagnostics instead.",
|
|
]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|