diff --git a/ovos_core/skill_installer.py b/ovos_core/skill_installer.py index de847d258329..3f4f5304fbd3 100644 --- a/ovos_core/skill_installer.py +++ b/ovos_core/skill_installer.py @@ -1,15 +1,17 @@ import enum +import shutil import sys from importlib import reload from os.path import exists from subprocess import Popen, PIPE from typing import Optional +import requests from combo_lock import NamedLock -from ovos_config.config import Configuration import ovos_plugin_manager from ovos_bus_client import Message +from ovos_config.config import Configuration from ovos_utils.log import LOG @@ -22,8 +24,9 @@ class InstallError(str, enum.Enum): class SkillsStore: # default constraints to use if none are given - DEFAULT_CONSTRAINTS = '/etc/mycroft/constraints.txt' # TODO XDG paths, keep backwards compat for now with msm/osm + DEFAULT_CONSTRAINTS = 'https://raw.githubusercontent.com/OpenVoiceOS/ovos-releases/refs/heads/main/constraints-stable.txt' PIP_LOCK = NamedLock("ovos_pip.lock") + UV = shutil.which("uv") # use 'uv pip' if available, speeds things up a lot and is the default in raspOVOS def __init__(self, bus, config=None): self.config = config or Configuration().get("skills", {}).get("installer", {}) @@ -44,6 +47,27 @@ def play_success_sound(self): snd = self.config.get("sounds", {}).get("pip_success", "snd/acknowledge.mp3") self.bus.emit(Message("mycroft.audio.play_sound", {"uri": snd})) + @staticmethod + def validate_constrainsts(constraints: str): + if constraints.startswith('http'): + LOG.debug(f"Constraints url: {constraints}") + try: + response = requests.head(constraints) + if response.status_code != 200: + LOG.error(f'Remote constraints file not accessible: {response.status_code}') + return False + return True + except Exception as e: + LOG.error(f'Error accessing remote constraints: {str(e)}') + return False + + # Use constraints to limit the installed versions + if not exists(constraints): + LOG.error('Couldn\'t find the constraints file') + return False + + return True + def pip_install(self, packages: list, constraints: Optional[str] = None, print_logs: bool = True): @@ -51,15 +75,18 @@ def pip_install(self, packages: list, LOG.error("no package list provided to install") self.play_error_sound() return False - # Use constraints to limit the installed versions - if constraints and not exists(constraints): - LOG.error('Couldn\'t find the constraints file') + + # can be set in mycroft.conf to change to testing/alpha channels + constraints = constraints or self.config.get("constraints", SkillsStore.DEFAULT_CONSTRAINTS) + + if not self.validate_constrainsts(constraints): self.play_error_sound() return False - elif exists(SkillsStore.DEFAULT_CONSTRAINTS): - constraints = SkillsStore.DEFAULT_CONSTRAINTS - pip_args = [sys.executable, '-m', 'pip', 'install'] + if self.UV is not None: + pip_args = [self.UV, 'pip', 'install'] + else: + pip_args = [sys.executable, '-m', 'pip', 'install'] if constraints: pip_args += ['-c', constraints] if self.config.get("break_system_packages", False): @@ -101,30 +128,37 @@ def pip_uninstall(self, packages: list, self.play_error_sound() return False - # Use constraints to limit package removal - if constraints and not exists(constraints): - LOG.error('Couldn\'t find the constraints file') + # can be set in mycroft.conf to change to testing/alpha channels + constraints = constraints or self.config.get("constraints", SkillsStore.DEFAULT_CONSTRAINTS) + + if not self.validate_constrainsts(constraints): self.play_error_sound() return False - elif exists(SkillsStore.DEFAULT_CONSTRAINTS): - constraints = SkillsStore.DEFAULT_CONSTRAINTS - if constraints: + # get protected packages that can't be uninstalled + # by default cant uninstall any official ovos package via this bus api + if constraints.startswith("http"): + cpkgs = requests.get(constraints).text.split("\n") + elif exists(constraints): with open(constraints) as f: - # remove version pinning and normalize _ to - (pip accepts both) - cpkgs = [p.split("~")[0].split("<")[0].split(">")[0].split("=")[0].replace("_", "-") - for p in f.read().split("\n") if p.strip()] + cpkgs = f.read().split("\n") else: cpkgs = ["ovos-core", "ovos-utils", "ovos-plugin-manager", "ovos-config", "ovos-bus-client", "ovos-workshop"] - # normalize _ to - (pip accepts both) - if any(p.replace("_", "-") in cpkgs for p in packages): + # remove version pinning and normalize _ to - (pip accepts both) + cpkgs = [p.split("~")[0].split("<")[0].split(">")[0].split("=")[0].replace("_", "-") + for p in cpkgs] + + if any(p in cpkgs for p in packages): LOG.error(f'tried to uninstall a protected package: {cpkgs}') self.play_error_sound() return False - pip_args = [sys.executable, '-m', 'pip', 'uninstall', '-y'] + if self.UV is not None: + pip_args = [self.UV, 'pip', 'uninstall'] + else: + pip_args = [sys.executable, '-m', 'pip', 'uninstall', '-y'] if self.config.get("break_system_packages", False): pip_args += ["--break-system-packages"] @@ -152,7 +186,8 @@ def pip_uninstall(self, packages: list, self.play_success_sound() return True - def validate_skill(self, url): + @staticmethod + def validate_skill(url): if not url.startswith("https://github.com/"): return False # TODO - check if setup.py