DankMaterialShell/scripts/format-staged.py

#!/usr/bin/env python3
"""Format staged .qml files using qmlls (the Qt QML language server) and
staged .js files using the qmlformat CLI (qmlls never answers a formatting
request for a JavaScript document; .mjs is not supported by qmlformat).

Per .qml file:
  1. Speak LSP over stdio to qmlls: initialize -> didOpen -> formatting,
     apply returned edits, save, `git add`.
  2. Run qmllint on the formatted file and warn about unused imports
     (informational only — never modifies files).

Per .js file: run `qmlformat <file>`, write its stdout back, `git add`.

Refuses to run if any staged file also has unstaged changes, since `git add`
would silently absorb those into the commit.
"""

import json
import os
import queue
import shutil
import subprocess
import sys
import threading
from pathlib import Path


TAB_SIZE = 4
REQUEST_TIMEOUT = 30
QMLLS_CANDIDATES = ["qmlls6", "qmlls"]
QMLLINT_CANDIDATES = ["/usr/lib/qt6/bin/qmllint", "qmllint6", "qmllint"]
QMLFORMAT_CANDIDATES = ["/usr/lib/qt6/bin/qmlformat", "qmlformat6", "qmlformat"]
FORMATTED_SUFFIXES = (".qml", ".js")


def git(*args, cwd=None):
    return subprocess.run(
        ["git", *args],
        cwd=cwd,
        capture_output=True,
        text=True,
        check=True,
    ).stdout


def repo_root():
    return Path(git("rev-parse", "--show-toplevel").strip())


def staged_files(root):
    out = git("diff", "--cached", "--name-only", "--diff-filter=ACMR", cwd=root)
    return [root / line for line in out.splitlines() if line.endswith(FORMATTED_SUFFIXES)]


def has_unstaged_changes(root, file):
    rel = str(file.relative_to(root))
    return git("diff", "--name-only", "--", rel, cwd=root).strip() != ""


def find_qmlls():
    for name in QMLLS_CANDIDATES:
        path = shutil.which(name)
        if path:
            return path
    return None


def find_qmlformat():
    for candidate in QMLFORMAT_CANDIDATES:
        path = candidate if "/" in candidate and Path(candidate).is_file() else shutil.which(candidate)
        if path:
            return path
    return None


def format_js(qmlformat, file):
    """Return qmlformat's output for a .js file, or None when it rejects the file."""
    result = subprocess.run([qmlformat, str(file)], capture_output=True, text=True)
    if result.returncode != 0:
        return None
    return result.stdout


def find_qmllint():
    for candidate in QMLLINT_CANDIDATES:
        path = candidate if "/" in candidate and Path(candidate).is_file() else shutil.which(candidate)
        if not path:
            continue
        try:
            result = subprocess.run([path, "--help"], capture_output=True, text=True, timeout=5)
        except (subprocess.TimeoutExpired, OSError):
            continue
        if "--json" in result.stdout:
            return path
    return None


def lint_unused_imports(qmllint, file):
    """Return a list of (line, message, suspect) for unused-import warnings.

    `suspect` is True when the same line also has an import-resolution failure,
    which often means the warning is a false positive (qmllint couldn't find
    the module, so its 'unused' verdict is unreliable).
    """
    result = subprocess.run(
        [qmllint, "--unused-imports", "warning", "--json", "-", str(file)],
        capture_output=True, text=True,
    )
    try:
        data = json.loads(result.stdout)
    except json.JSONDecodeError:
        return []

    files = data.get("files", [])
    if not files:
        return []
    warnings = files[0].get("warnings", [])

    failed_lines = {w["line"] for w in warnings if w.get("id") == "import" and "line" in w}
    findings = []
    for w in warnings:
        if w.get("id") != "unused-imports" or "line" not in w:
            continue
        line = w["line"]
        findings.append((line, w.get("message", "Unused import"), line in failed_lines))
    findings.sort(key=lambda x: x[0])
    return findings


class LspClient:
    def __init__(self, command):
        self.proc = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
        )
        self._next_id = 1
        self._write_lock = threading.Lock()
        self._responses = queue.Queue()
        # Drain stdout continuously: qmlls floods publishDiagnostics (~100KB
        # per file) and blocks once the 64KB pipe fills, deadlocking against
        # our own blocking write of a large didOpen.
        threading.Thread(target=self._pump, daemon=True).start()

    def _send(self, msg):
        body = json.dumps(msg).encode("utf-8")
        header = f"Content-Length: {len(body)}\r\n\r\n".encode("ascii")
        with self._write_lock:
            self.proc.stdin.write(header + body)
            self.proc.stdin.flush()

    def _read(self):
        headers = {}
        while True:
            line = self.proc.stdout.readline()
            if not line:
                raise RuntimeError("qmlls closed unexpectedly")
            line = line.decode("ascii").rstrip("\r\n")
            if line == "":
                break
            key, _, value = line.partition(":")
            headers[key.strip().lower()] = value.strip()
        length = int(headers["content-length"])
        body = b""
        while len(body) < length:
            chunk = self.proc.stdout.read(length - len(body))
            if not chunk:
                raise RuntimeError("qmlls closed mid-message")
            body += chunk
        return json.loads(body)

    def _pump(self):
        while True:
            try:
                msg = self._read()
            except (RuntimeError, OSError, ValueError):
                self._responses.put(None)
                return
            if "id" in msg and "method" in msg:
                # Server-to-client request — reply with null so it doesn't stall.
                try:
                    self._send({"jsonrpc": "2.0", "id": msg["id"], "result": None})
                except OSError:
                    pass
                continue
            if "id" in msg and ("result" in msg or "error" in msg):
                self._responses.put(msg)

    def request(self, method, params):
        req_id = self._next_id
        self._next_id += 1
        self._send({"jsonrpc": "2.0", "id": req_id, "method": method, "params": params})
        while True:
            try:
                msg = self._responses.get(timeout=REQUEST_TIMEOUT)
            except queue.Empty:
                raise RuntimeError(f"LSP {method} timed out after {REQUEST_TIMEOUT}s")
            if msg is None:
                raise RuntimeError("qmlls closed unexpectedly")
            if msg.get("id") != req_id:
                continue
            if "error" in msg:
                raise RuntimeError(f"LSP {method} error: {msg['error']}")
            return msg.get("result")

    def notify(self, method, params):
        self._send({"jsonrpc": "2.0", "method": method, "params": params})

    def shutdown(self):
        try:
            self.request("shutdown", None)
            self.notify("exit", None)
        except Exception:
            pass
        try:
            self.proc.wait(timeout=2)
        except subprocess.TimeoutExpired:
            self.proc.kill()


def apply_edits(text, edits):
    """Apply LSP TextEdits (non-overlapping) to text, end-first."""
    if not edits:
        return text

    lines = text.splitlines(keepends=True)
    line_starts = [0]
    for line in lines:
        line_starts.append(line_starts[-1] + len(line))

    def offset(pos):
        line = pos["line"]
        if line >= len(line_starts):
            return len(text)
        return min(line_starts[line] + pos["character"], len(text))

    sorted_edits = sorted(
        edits,
        key=lambda e: (e["range"]["start"]["line"], e["range"]["start"]["character"]),
        reverse=True,
    )
    for edit in sorted_edits:
        start = offset(edit["range"]["start"])
        end = offset(edit["range"]["end"])
        text = text[:start] + edit["newText"] + text[end:]
    return text


def start_client(qmlls, root):
    client = LspClient([qmlls])
    client.request("initialize", {
        "processId": os.getpid(),
        "rootUri": root.as_uri(),
        "workspaceFolders": [{"uri": root.as_uri(), "name": root.name}],
        "capabilities": {
            "textDocument": {
                "formatting": {"dynamicRegistration": False},
                "synchronization": {"dynamicRegistration": False},
            },
        },
    })
    client.notify("initialized", {})
    return client


def format_js_files(qmlformat, root, files):
    changed = 0
    skipped = 0
    for file in files:
        rel = file.relative_to(root)
        print(f"  {rel} ... ", end="", flush=True)
        original = file.read_text()
        new_text = format_js(qmlformat, file)
        if new_text is None:
            skipped += 1
            print("skipped (qmlformat rejected the file)")
            continue
        if new_text == original:
            print("unchanged")
            continue
        file.write_text(new_text)
        git("add", "--", str(rel), cwd=root)
        changed += 1
        print("formatted & staged")
    return changed, skipped


def main():
    root = repo_root()
    files = staged_files(root)
    if not files:
        print("No staged .qml or .js files.")
        return 0

    dirty = [f for f in files if has_unstaged_changes(root, f)]
    if dirty:
        print("Refusing to format: staged files have unstaged changes:", file=sys.stderr)
        for f in dirty:
            print(f"  {f.relative_to(root)}", file=sys.stderr)
        print("\nStash or stage those changes first.", file=sys.stderr)
        return 1

    js_files = [f for f in files if f.suffix == ".js"]
    files = [f for f in files if f.suffix == ".qml"]

    changed = 0
    skipped = 0
    if js_files:
        qmlformat = find_qmlformat()
        if not qmlformat:
            print(f"qmlformat not found (tried: {', '.join(QMLFORMAT_CANDIDATES)})", file=sys.stderr)
            return 1
        changed, skipped = format_js_files(qmlformat, root, js_files)

    if not files:
        print(f"\n{changed} of {len(js_files)} file(s) changed.")
        if skipped:
            print(f"{skipped} file(s) skipped (could not be formatted; see above).")
        return 0

    qmlls = find_qmlls()
    if not qmlls:
        print(f"qmlls not found (tried: {', '.join(QMLLS_CANDIDATES)})", file=sys.stderr)
        return 1

    qmllint = find_qmllint()
    if not qmllint:
        print("warning: qmllint with --json not found; skipping unused-import checks", file=sys.stderr)

    client = start_client(qmlls, root)
    unused_by_file = {}
    try:
        for file in files:
            rel = file.relative_to(root)
            print(f"  {rel} ... ", end="", flush=True)

            original = file.read_text()
            uri = file.as_uri()

            client.notify("textDocument/didOpen", {
                "textDocument": {
                    "uri": uri,
                    "languageId": "qml",
                    "version": 1,
                    "text": original,
                },
            })

            try:
                edits = client.request("textDocument/formatting", {
                    "textDocument": {"uri": uri},
                    "options": {"tabSize": TAB_SIZE, "insertSpaces": True},
                })
            except (RuntimeError, OSError) as exc:
                # qmlls (qmlformat's DOM) chokes on some valid files: it refuses
                # them ("Cannot format invalid documents!") or outright crashes
                # (e.g. SIGABRT on a function declaration inside a property
                # binding). Don't let one file's formatter bug abort the commit.
                skipped += 1
                if client.proc.poll() is not None:
                    print("skipped (qmlls crashed on this file; restarting it)")
                    client = start_client(qmlls, root)
                    continue
                if "timed out" in str(exc):
                    print("skipped (qmlls timed out on this file; restarting it)")
                    client.proc.kill()
                    client = start_client(qmlls, root)
                    continue
                client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})
                if "invalid document" in str(exc).lower():
                    print("skipped (qmlls rejected as invalid)")
                else:
                    print(f"skipped ({exc})")
                continue

            client.notify("textDocument/didClose", {"textDocument": {"uri": uri}})

            new_text = apply_edits(original, edits or [])
            if new_text == original:
                print("unchanged")
                continue

            file.write_text(new_text)
            git("add", "--", str(rel), cwd=root)
            changed += 1
            print("formatted & staged")

        if qmllint:
            for file in files:
                findings = lint_unused_imports(qmllint, file)
                if findings:
                    unused_by_file[file] = findings

        print(f"\n{changed} of {len(files) + len(js_files)} file(s) changed.")
        if skipped:
            print(f"{skipped} file(s) skipped (could not be formatted; see above).")

        if unused_by_file:
            print("\nUnused import warnings (informational, not auto-removed):")
            for file, findings in unused_by_file.items():
                rel = file.relative_to(root)
                for line, message, suspect in findings:
                    suffix = "  [suspect: import didn't resolve]" if suspect else ""
                    print(f"  {rel}:{line}  {message}{suffix}")

        return 0
    finally:
        client.shutdown()


if __name__ == "__main__":
    sys.exit(main())