"""Run a vendor command-line tool's own installer, and say how it went.
Qt-free installer helpers. :func:`run_install` blocks, so GUI callers should run it off
the GUI thread; nothing here touches a widget.
This module defines no install commands. Each tool's ``install_methods`` in
:mod:`spacr.qt.ai.providers` supplies the rows used by :func:`plan_install`, which picks
the first row whose prerequisite programs are on ``PATH``. The tool's ``install_hint``
is built from those same rows; this module does not wire them to a screen or Install
button.
What a run reports is a kind (:data:`INSTALLED`, :data:`NO_NETWORK`, ...)
rather than a sentence, so the screen can say something the user can act on,
in the user's language, for each way an install goes wrong.
"""
from __future__ import annotations
import os
import re
import shlex
import shutil
import signal
import subprocess
import sys
import tempfile
import threading
import time
from collections import deque
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, List, Optional, Sequence, Union
from .providers import InstallMethod, platform_family
#: The CLI is installed and was found afterwards.
INSTALLED = "installed"
#: The user pressed Cancel and the installer was stopped.
CANCELLED = "cancelled"
#: The installer ran past :data:`INSTALL_TIMEOUT_S` and was stopped.
TIMED_OUT = "timed out"
#: The download could not reach its server.
NO_NETWORK = "no network"
#: The server answered and refused the download.
REFUSED = "refused"
#: The installer was not allowed to write where it installs.
PERMISSION = "permission"
#: The installer exited 0 and the CLI still cannot be found.
NOT_FOUND = "not found"
#: The installer could not be started at all.
NOT_STARTED = "not started"
#: No row of the tool's install methods can run on this computer.
NOT_AUTOMATIC = "not automatic"
#: The installer failed for a reason none of the other kinds names.
FAILED = "failed"
#: Seconds an installer may run before it is stopped as hung.
INSTALL_TIMEOUT_S = 20 * 60
#: Seconds a stopped installer is given to exit before it is killed.
STOP_GRACE_S = 3
#: Seconds between two checks for Cancel or the time limit.
WATCH_INTERVAL_S = 0.2
#: How many of the installer's last output lines are kept for the verdict.
KEPT_LINES = 40
#: How many of those lines a failure message quotes.
QUOTED_LINES = 2
#: The longest quotation of an installer's output, in characters.
QUOTED_CHARS = 300
#: What a missing prerequisite is called when a message names it.
REQUIREMENT_NAMES = {
"npm": "npm (Node.js)",
"brew": "Homebrew",
"conda": "conda",
"winget": "winget",
"curl": "curl",
"bash": "bash",
}
#: Output that means the installer could not reach the network.
NETWORK_TEXT = (
"could not resolve host", "enotfound", "eai_again", "etimedout",
"econnrefused", "econnreset", "enetunreach", "network is unreachable",
"temporary failure in name resolution", "name or service not known",
"failed to connect", "connection timed out", "connection refused",
"connection failed", "could not connect", "getaddrinfo",
)
#: Output that means a server answered and refused the download.
REFUSED_TEXT = (
"the requested url returned error", "403 forbidden", "404 not found",
"401 unauthorized", "e403", "e404", "http error 4", "http error 5",
)
#: Output that means the installer could not write where it installs.
PERMISSION_TEXT = (
"eacces", "eperm", "permission denied", "access is denied",
"operation not permitted", "not writable",
)
#: curl's exit statuses for a network that could not be reached.
CURL_NETWORK_EXITS = (5, 6, 7, 28, 35, 56)
#: curl's exit status for an HTTP error answer under ``--fail``.
CURL_REFUSED_EXIT = 22
_ESCAPES = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]"
r"|\x1b\][^\x07]*\x07"
r"|[\x00-\x08\x0b-\x1f\x7f]")
_LINE_BREAKS = re.compile(rb"[\r\n]")
_spawn = subprocess.Popen
_query = subprocess.run
_signal_group = getattr(os, "killpg", None)
@dataclass(frozen=True)
[docs]
class InstallPlan:
"""What pressing Install would run for one tool on this computer.
:ivar tool: the tool, a :class:`spacr.qt.ai.providers.CommandLineTool`.
:ivar method: the row that would run, or ``None`` when none of the
tool's rows can run here.
:ivar platform: the ``sys.platform`` value the plan was made for.
"""
tool: Any
method: Optional[InstallMethod]
platform: str
@property
[docs]
def automatic(self) -> bool:
"""Whether a row was found that spaCR can run."""
return self.method is not None
@property
[docs]
def command(self) -> str:
"""The command to show: the row that would run, else the hint."""
if self.method is not None:
return self.method.shown
return str(getattr(self.tool, "install_hint", "") or "")
@property
[docs]
def needs(self) -> str:
"""The programs that would make an install possible, for a message.
:returns: the first program each row needs, named for a reader and
joined with " or ", such as ``"npm (Node.js) or Homebrew"``.
"""
names: List[str] = []
for method in getattr(self.tool, "install_methods", ()) or ():
for need in method.needs[:1]:
name = REQUIREMENT_NAMES.get(need, need)
if name not in names:
names.append(name)
return " or ".join(names)
@dataclass(frozen=True)
[docs]
class InstallOutcome:
"""How one run of an installer ended.
:ivar kind: one of the module's kinds, such as :data:`INSTALLED`.
:ivar exit_status: the recorded installer exit status, or ``None`` when no status is
retained, including cancellation and timeout outcomes.
:ivar tail: the installer's last output, shortened for a failure message, or the
startup exception text when the installer could not be started.
:ivar location: where the CLI was found afterwards, or ``""``.
"""
kind: str
exit_status: Optional[int] = None
tail: str = ""
location: str = ""
@property
[docs]
def ok(self) -> bool:
"""Whether ``kind`` is :data:`INSTALLED`; this property performs no new executable
lookup."""
return self.kind == INSTALLED
[docs]
def plan_install(tool: Any, platform: Optional[str] = None) -> InstallPlan:
"""Choose the row of ``tool``'s install methods that can run here.
:param tool: a :class:`spacr.qt.ai.providers.CommandLineTool`, or any
object with the same attributes; one with no ``install_methods``
gets a plan with no method.
:param platform: a ``sys.platform`` value; the running one by default.
:returns: the plan, whose ``method`` is the first row whose ``needs``
are all on ``PATH``, or ``None`` when there is no such row.
"""
platform = platform or sys.platform
for method in getattr(tool, "install_methods", ()) or ():
if all(shutil.which(need) for need in method.needs):
return InstallPlan(tool, method, platform)
return InstallPlan(tool, None, platform)
[docs]
def launch_command(method: InstallMethod,
platform: str) -> Union[List[str], str]:
"""The command line that starts ``method``'s installer.
:param method: the row to run.
:param platform: a ``sys.platform`` value.
:returns: for ``"cmd"`` on Windows, the shown line itself as one string,
which Windows hands to ``cmd`` unchanged; for ``"shell"``, ``bash -o
pipefail -c`` and the command, so a failed download in ``curl ... |
bash`` is a failed install rather than an empty script run
successfully; otherwise the command split into a program, looked up
on ``PATH``, and its arguments (:func:`split_command`).
"""
if method.runner == "cmd" and platform_family(platform) == "win32":
return method.shown
if method.runner == "shell":
return [shutil.which("bash") or "bash", "-o", "pipefail", "-c",
method.command]
argv = split_command(method.command, platform)
argv[0] = shutil.which(argv[0]) or argv[0]
return argv
[docs]
def split_command(command: str, platform: str) -> List[str]:
"""Split a command as typed into a program and its arguments.
:param command: the command, with any path that holds a space quoted
the way :func:`spacr.qt.ai.providers.quote_path` quotes it.
:param platform: a ``sys.platform`` value.
:returns: the words. On Windows a backslash is part of a path, not an
escape, and a double-quoted word loses its quotes.
"""
if platform_family(platform) != "win32":
return shlex.split(command)
words = shlex.split(command, posix=False)
return [word[1:-1] if len(word) > 1 and word[0] == word[-1] == '"'
else word for word in words]
def _start_options(platform: str) -> dict:
"""Keyword arguments that put the installer in a group of its own.
:param platform: a ``sys.platform`` value.
:returns: a new session on POSIX, so Cancel can stop the whole
``curl | bash`` pipeline and not only its first process; a new
process group with no console window on Windows.
"""
if platform_family(platform) == "win32":
flags = (getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x200)
| getattr(subprocess, "CREATE_NO_WINDOW", 0x8000000))
return {"creationflags": flags}
return {"start_new_session": True}
[docs]
def clean_line(raw: bytes) -> str:
"""One line of installer output, decoded and without terminal codes.
:param raw: the bytes between two line breaks.
:returns: the text with colour codes and control characters removed and
the ends stripped.
"""
text = raw.decode("utf-8", "replace")
return _ESCAPES.sub("", text).strip()
[docs]
def stop_process(proc: Any, platform: str) -> None:
"""Request termination of an installer process or its process group.
On POSIX, signal the installer's group even when the installer itself has exited,
because a child may still hold its output open. A running installer gets a
termination request and, after :data:`STOP_GRACE_S`, a kill attempt if its wait
times out. On Windows, a running installer is targeted with ``taskkill /T /F``
without waiting for that command; an already exited installer is left alone. These
attempts do not confirm that every descendant has stopped.
:param proc: the installer, running or exited.
:param platform: a ``sys.platform`` value.
"""
exited = proc.poll() is not None
if platform_family(platform) == "win32":
if exited:
return
try:
_spawn(["taskkill", "/T", "/F", "/PID", str(proc.pid)],
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
except OSError:
proc.kill()
return
if exited:
try:
_signal_group(proc.pid, signal.SIGTERM)
except (OSError, TypeError):
pass
return
try:
_signal_group(proc.pid, signal.SIGTERM)
except (OSError, TypeError):
proc.terminate()
try:
proc.wait(timeout=STOP_GRACE_S)
except subprocess.TimeoutExpired:
try:
_signal_group(proc.pid, signal.SIGKILL)
except (OSError, TypeError):
proc.kill()
def _kill_group(proc: Any, platform: str) -> None:
"""Kill what is left of an installer's group after it was asked to stop.
:param proc: the installer, running or exited.
:param platform: a ``sys.platform`` value; nothing is done on Windows,
where :func:`stop_process` already ended the whole tree by force.
"""
if platform_family(platform) == "win32":
return
try:
_signal_group(proc.pid, signal.SIGKILL)
except (OSError, TypeError):
pass
def _watch(proc: Any, stop: Optional[threading.Event], deadline: float,
finished: threading.Event, fired: List[str], platform: str) -> None:
"""Stop ``proc`` when Cancel is pressed or the time limit passes.
Runs on a thread of its own, because the thread running the install is
blocked reading the installer's output and cannot notice either.
:param proc: the running installer.
:param stop: set by Cancel, or ``None``.
:param deadline: the ``time.monotonic()`` value past which it is hung.
:param finished: set once the installer's output has closed and it has
exited, to end this watch. When it is still not set
:data:`STOP_GRACE_S` after the stop, whatever is left of the
installer's group is killed.
:param fired: receives :data:`CANCELLED` or :data:`TIMED_OUT` when this
watch is what stopped the installer.
:param platform: a ``sys.platform`` value.
"""
while not finished.wait(WATCH_INTERVAL_S):
if stop is not None and stop.is_set():
fired.append(CANCELLED)
elif time.monotonic() >= deadline:
fired.append(TIMED_OUT)
else:
continue
stop_process(proc, platform)
if not finished.wait(STOP_GRACE_S):
_kill_group(proc, platform)
return
[docs]
def classify(exit_status: int, lines: Sequence[str],
method: InstallMethod) -> str:
"""Classify an installer failure from its output markers and exit status.
:param exit_status: its exit status.
:param lines: its last output lines.
:param method: the row that ran, whose command says whether a curl exit
status can be read as curl's.
:returns: :data:`NO_NETWORK`, :data:`REFUSED`, :data:`PERMISSION` or
:data:`FAILED`.
"""
text = "\n".join(lines).lower()
if any(marker in text for marker in NETWORK_TEXT):
return NO_NETWORK
if any(marker in text for marker in REFUSED_TEXT):
return REFUSED
if any(marker in text for marker in PERMISSION_TEXT):
return PERMISSION
if method.command.startswith("curl "):
if exit_status in CURL_NETWORK_EXITS:
return NO_NETWORK
if exit_status == CURL_REFUSED_EXIT:
return REFUSED
return FAILED
[docs]
def quote_tail(lines: Sequence[str]) -> str:
"""The installer's last lines, short enough for a message.
:param lines: its output lines, oldest first.
:returns: the last :data:`QUOTED_LINES` joined by a space and cut to
:data:`QUOTED_CHARS`.
"""
text = " ".join(list(lines)[-QUOTED_LINES:])
if len(text) > QUOTED_CHARS:
text = text[:QUOTED_CHARS - 1].rstrip() + "…"
return text
[docs]
def run_install(plan: InstallPlan,
on_output: Optional[Callable[[str], None]] = None,
stop: Optional[threading.Event] = None, *,
timeout_s: float = INSTALL_TIMEOUT_S) -> InstallOutcome:
"""Run ``plan``'s installer to the end and say how it went.
Blocks while consuming the installer's output and waiting for its process to exit,
so run it off the GUI thread. Standard input supplies end-of-file. The installer
runs in a scratch folder, with cleanup attempted afterwards, because the Windows
command downloads ``install.cmd`` into its working folder. Cancellation and timeout
handling request termination; neither guarantees a fixed upper bound on this call's
duration.
:param plan: from :func:`plan_install`.
:param on_output: called with each non-blank line of output, on the
calling thread, as it arrives.
:param stop: set it to request cancellation of the installer.
:param timeout_s: seconds before the watcher requests termination after the
installer starts; not a bound on the total call duration.
:returns: the recorded outcome; :data:`INSTALLED` requires an exit status of 0 and a
CLI found afterwards, with ``location`` saying where. If cancellation or timeout
was recorded, its kind is returned unless both success checks pass. Other
non-zero exits are classified from their output and status; an exit of 0 with no
CLI found is not a successful install.
"""
if plan.method is None:
return InstallOutcome(NOT_AUTOMATIC)
command = launch_command(plan.method, plan.platform)
workdir = tempfile.mkdtemp(prefix="spacr-install-")
kept: deque = deque(maxlen=KEPT_LINES)
fired: List[str] = []
try:
try:
proc = _spawn(command, stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
cwd=workdir, env=os.environ.copy(),
**_start_options(plan.platform))
except (OSError, ValueError) as exc:
return InstallOutcome(NOT_STARTED, tail=str(exc))
finished = threading.Event()
watcher = threading.Thread(
target=_watch, name="spacr-install-watch", daemon=True,
args=(proc, stop, time.monotonic() + timeout_s, finished, fired,
plan.platform))
watcher.start()
pending = b""
try:
while True:
chunk = proc.stdout.read1(4096)
if not chunk:
break
pending += chunk
*complete, pending = _LINE_BREAKS.split(pending)
for raw in complete:
_keep(raw, kept, on_output)
_keep(pending, kept, on_output)
exit_status = proc.wait()
finally:
finished.set()
watcher.join(timeout=STOP_GRACE_S + 1)
proc.stdout.close()
finally:
shutil.rmtree(workdir, ignore_errors=True)
tail = quote_tail(kept)
if exit_status != 0:
if fired:
return InstallOutcome(fired[0], None, tail)
return InstallOutcome(classify(exit_status, kept, plan.method),
exit_status, tail)
location = locate(plan.tool, plan.platform)
if fired and not location:
return InstallOutcome(fired[0], None, tail)
return InstallOutcome(INSTALLED if location else NOT_FOUND, 0, tail,
location)
def _keep(raw: bytes, kept: deque,
on_output: Optional[Callable[[str], None]]) -> None:
"""Record one raw output line and pass it on, unless it is blank.
:param raw: the bytes of the line.
:param kept: the recent lines, which this appends to.
:param on_output: the caller's line callback, or ``None``.
"""
line = clean_line(raw)
if not line:
return
kept.append(line)
if on_output is not None:
on_output(line)
[docs]
def likely_folders(platform: Optional[str] = None) -> List[str]:
"""Folders installers put a CLI in that may not be on ``PATH``.
A program started from a desktop icon often has a shorter ``PATH`` than a
terminal, and Claude's own installer puts ``claude`` in
``~/.local/bin``, which is exactly the folder such a ``PATH`` leaves
out.
:param platform: a ``sys.platform`` value; the running one by default.
:returns: candidate folders, most likely first, without duplicates.
"""
platform = platform or sys.platform
home = Path.home()
prefixes = [Path(sys.prefix)]
if os.environ.get("CONDA_PREFIX"):
prefixes.append(Path(os.environ["CONDA_PREFIX"]))
if platform_family(platform) == "win32":
local = Path(os.environ.get("LOCALAPPDATA", home / "AppData/Local"))
roaming = Path(os.environ.get("APPDATA", home / "AppData/Roaming"))
programs = Path(os.environ.get("ProgramFiles", "C:/Program Files"))
folders = [home / ".local" / "bin", local / "Microsoft/WinGet/Links",
programs / "GitHub CLI", roaming / "npm"]
for prefix in prefixes:
folders += [prefix / "Scripts", prefix / "Library" / "bin"]
else:
folders = [home / ".local" / "bin", home / ".claude" / "local",
Path("/opt/homebrew/bin"), Path("/usr/local/bin"),
Path("/home/linuxbrew/.linuxbrew/bin"),
home / ".linuxbrew" / "bin"]
folders += [prefix / "bin" for prefix in prefixes]
npm = _npm_folder(platform)
if npm:
folders.append(Path(npm))
seen: List[str] = []
for folder in folders:
text = str(folder)
if text not in seen:
seen.append(text)
return seen
def _npm_folder(platform: str) -> str:
"""Where ``npm install -g`` puts programs, or ``""``.
:param platform: a ``sys.platform`` value.
:returns: the global prefix's ``bin`` folder on POSIX and the prefix
itself on Windows, as ``npm prefix -g`` reports it; ``""`` when
there is no npm or it does not answer.
"""
npm = shutil.which("npm")
if not npm:
return ""
try:
answer = _query([npm, "prefix", "-g"], stdin=subprocess.DEVNULL,
capture_output=True, text=True, timeout=20)
except (OSError, subprocess.SubprocessError):
return ""
prefix = (answer.stdout or "").strip()
if answer.returncode != 0 or not prefix:
return ""
if platform_family(platform) == "win32":
return prefix
return str(Path(prefix) / "bin")
[docs]
def locate(tool: Any, platform: Optional[str] = None) -> str:
"""Find ``tool``'s CLI on ``PATH`` or in a folder installers use.
:param tool: a :class:`spacr.qt.ai.providers.CommandLineTool`.
:param platform: a ``sys.platform`` value; the running one by default.
:returns: the executable path found in the searched locations, or ``""`` when none
was found there.
"""
name = str(getattr(tool, "cli_name", "") or "")
if not name:
return ""
found = shutil.which(name)
if found:
return found
for folder in likely_folders(platform):
found = shutil.which(name, path=folder)
if found:
return found
return ""
[docs]
def put_on_path(location: str) -> bool:
"""Make the folder holding ``location`` part of this process's ``PATH``.
So a CLI installed into a folder the desktop's ``PATH`` leaves out works
in spaCR at once, without a restart. Changes this process's environment,
so call it on the GUI thread.
:param location: the path of an executable, from :func:`locate`.
:returns: ``True`` when ``PATH`` was changed.
"""
if not location:
return False
folder = os.path.dirname(os.path.abspath(location))
current = os.environ.get("PATH", "")
parts = [os.path.normcase(os.path.abspath(p))
for p in current.split(os.pathsep) if p]
if os.path.normcase(folder) in parts:
return False
os.environ["PATH"] = folder + (os.pathsep + current if current else "")
return True