Source code for simvx.ide.lint.runner

"""Linting and formatting subprocess runner.

The commands are configurable (``IDESection.lint_command`` /
``format_command``); the shipped defaults are ``ruff check`` and ``ruff format``.
"""

import json
import logging
import shlex
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from ..config import Config
    from ..state import State

from ..state import Diagnostic

log = logging.getLogger(__name__)

#: Source of the diagnostics this module raises about the formatter itself, as
#: opposed to the ones it parses out of the linter's report.
_FORMATTER_SOURCE = "format"


def _parse_refusal(data: bytes, path: str) -> tuple[int, int, str] | None:
    """``(line, column, reason)`` if *data* is not source Python will run.

    Zero-indexed, matching :class:`~simvx.ide.state.Diagnostic`. Returns None
    when the bytes compile, which is the case a formatter is expected to
    produce. The check is :func:`compile` rather than a parse of our own,
    because the file has to satisfy the interpreter and nothing narrower: a
    formatter is free to emit any syntax this Python accepts.
    """
    try:
        text = data.decode("utf-8")
    except UnicodeDecodeError as exc:
        return (0, 0, f"result is not UTF-8 ({exc.reason} at byte {exc.start})")

    try:
        compile(text, path, "exec")
    except SyntaxError as exc:
        return (max((exc.lineno or 1) - 1, 0), max((exc.offset or 1) - 1, 0), exc.msg or "syntax error")
    except ValueError as exc:
        # A null byte, which compile() rejects without a position.
        return (0, 0, str(exc))
    return None




class LintRunner:
    """Runs the configured lint and format commands and feeds diagnostics into State."""

    def __init__(self, state: State, config: Config):
        self._state = state
        self._config = config
        self._env = config.get_env(state.project_root) if state.project_root else None

    # -- Public API ------------------------------------------------------------



    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




    def format_file(self, path: str) -> bool:
        """Run the configured formatter on *path*. Returns True when it stands.

        The formatter rewrites the file in place, so its result is re-read
        before the rewrite is allowed to stand. Nothing a working formatter
        emits should fail to compile, which is exactly why the check is worth
        keeping: a formatter that is broken, misconfigured or handed a file it
        mis-tokenises writes over the user's source with something that will not
        run. When the result no longer compiles, the file goes back to what it
        was and the refusal is reported as an error diagnostic, which is what
        puts it in the output panel.
        """
        cmd = self._config.format_command
        cmd_list = shlex.split(cmd) + [path]
        env = self._env or self._config.get_env(self._state.project_root)
        source = Path(path)
        try:
            before = source.read_bytes()
        except OSError as exc:
            log.warning("Cannot read %s before formatting: %s", path, exc)
            return False
        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("Format command not found: %s", cmd)
            self._report_formatter_fault(
                path,
                0,
                0,
                f"format-on-save is on and {cmd_list[0]!r} is not installed, so nothing was formatted. "
                "Install it, set a different format_command, or turn format_on_save off.",
            )
            return False
        except subprocess.TimeoutExpired:
            log.warning("Format timed out for %s", path)
            return False

        if result.returncode not in (0, 1):
            log.warning("Format failed for %s: %s", path, result.stderr.strip())
            return False

        try:
            after = source.read_bytes()
        except OSError as exc:
            log.warning("Cannot read %s after formatting: %s", path, exc)
            return False
        if after == before:
            return True
        refusal = _parse_refusal(after, path)
        if refusal is None:
            return True

        source.write_bytes(before)
        line, column, reason = refusal
        log.warning("Format refused for %s: %s", path, reason)
        self._report_formatter_fault(
            path,
            line,
            column,
            f"{cmd!r} produced source that does not compile ({reason}); "
            "the file was left as it was. Set a different format_command, or turn format_on_save off.",
        )
        return False


    def _report_formatter_fault(self, path: str, line: int, column: int, message: str) -> None:
        """Put a formatter fault where the user will see it, not only in the log.

        Format-on-save runs without being asked for, so a fault in it has no
        other surface: the diagnostics channel is what reaches the output panel.
        """
        self._state.set_diagnostics(
            path,
            [
                Diagnostic(
                    path=path,
                    line=line,
                    col_start=column,
                    col_end=column + 1,
                    severity=1,
                    message=message,
                    source=_FORMATTER_SOURCE,
                )
            ],
        )



    def lint_on_save(self, path: str):
        """Called after a file is saved -- runs lint and updates diagnostics."""
        self.lint_file(path)




    def format_on_save(self, path: str):
        """Format *path* on save, but only when the ``format_on_save`` setting is enabled."""
        if self._config.format_on_save:
            self.format_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