"""Linting and formatting subprocess runner using ruff."""
from __future__ import annotations
import json
import logging
import shlex
import subprocess
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..config import IDEConfig
from ..state import IDEState
from ..state import Diagnostic
log = logging.getLogger(__name__)
[docs]
class LintRunner:
"""Runs ruff check / ruff format and feeds diagnostics into IDEState."""
def __init__(self, state: IDEState, config: IDEConfig):
self._state = state
self._config = config
self._env = config.get_env(state.project_root) if state.project_root else None
# -- Public API ------------------------------------------------------------
[docs]
def lint_file(self, path: str) -> list[Diagnostic]:
"""Run ruff check on *path*, parse JSON output, update state diagnostics."""
cmd = self._config.lint_command
cmd_list = shlex.split(cmd) + [path]
env = self._env or self._config.get_env(self._state.project_root)
try:
result = subprocess.run(
cmd_list,
capture_output=True,
text=True,
timeout=15,
cwd=self._state.project_root or None,
env=env,
)
except FileNotFoundError:
log.warning("Lint command not found: %s", cmd)
return []
except subprocess.TimeoutExpired:
log.warning("Lint timed out for %s", path)
return []
diagnostics = self._parse_ruff_json(result.stdout)
self._state.set_diagnostics(path, diagnostics)
return diagnostics
[docs]
def lint_on_save(self, path: str):
"""Called after a file is saved -- runs lint and updates diagnostics."""
self.lint_file(path)
# -- Parsing ---------------------------------------------------------------
def _parse_ruff_json(self, output: str) -> list[Diagnostic]:
"""Parse ruff check --output-format=json output into Diagnostic objects."""
if not output or not output.strip():
return []
try:
entries = json.loads(output)
except json.JSONDecodeError:
log.debug("Failed to parse ruff JSON output")
return []
diagnostics: list[Diagnostic] = []
for entry in entries:
code = entry.get("code", "")
message = entry.get("message", "")
filename = entry.get("filename", "")
loc = entry.get("location", {})
end_loc = entry.get("end_location", {})
line = loc.get("row", 1) - 1
col_start = loc.get("column", 1) - 1
col_end = end_loc.get("column", col_start + 2) - 1
# Ruff codes: E/W = style, F = pyflakes errors, B = bugbear
# Map to severity: F-codes and some E-codes are errors, rest warnings
severity = 2 # warning by default
if code.startswith("F") or code.startswith("E9"):
severity = 1 # error
diagnostics.append(
Diagnostic(
path=filename,
line=line,
col_start=col_start,
col_end=col_end,
severity=severity,
message=message,
source="ruff",
code=code,
)
)
return diagnostics