chatbot_v3/.claude/hooks/python_quality.py

#!/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.
"""

from __future__ import annotations

import hashlib
import json
import subprocess
import sys
from pathlib import Path

PYTHON_SUFFIXES = {".py", ".pyi"}
SKIP_PARTS = {".git", ".venv", "__pycache__"}


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 _should_check(path):
        return 0

    relative_path = _display_path(path, repo_root)
    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

    message = _build_message(
        relative_path=relative_path,
        changed=changed,
        fix=fix,
        fmt=fmt,
        lint=lint,
        types=types,
    )
    if message:
        print(
            json.dumps(
                {
                    "suppressOutput": True,
                    "hookSpecificOutput": {
                        "hookEventName": "PostToolUse",
                        "additionalContext": message,
                    },
                }
            )
        )

    return 0


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 _should_check(path: Path) -> bool:
    return path.is_file() and path.suffix in PYTHON_SUFFIXES and SKIP_PARTS.isdisjoint(path.parts)


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 _build_message(
    *,
    relative_path: str,
    changed: bool,
    fix: subprocess.CompletedProcess[str],
    fmt: subprocess.CompletedProcess[str],
    lint: subprocess.CompletedProcess[str],
    types: subprocess.CompletedProcess[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 (fix, fmt, lint, types):
        if result.returncode == 0:
            continue

        output = _combined_output(result)
        if not output:
            output = f"Command exited with status {result.returncode}."

        lines.extend(
            [
                "",
                f"Remaining diagnostics from `{_command_label(result)}`:",
                "```text",
                output[-6000:],
                "```",
            ]
        )

    return "\n".join(lines).strip()


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