PipxInterface/Pipx.py

# -*- coding: utf-8 -*-

# Copyright (c) 2024 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the pipx GUI logic.
"""

import contextlib
import functools
import json
import os
import pathlib
import sysconfig

from PyQt6.QtCore import QObject, QProcess, pyqtSignal

from eric7 import Preferences
from eric7.EricWidgets import EricMessageBox
from eric7.SystemUtilities import OSUtilities

try:
    from eric7.EricCore.EricProcess import EricProcess
except ImportError:
    # backward compatibility for eric-ide < 24.10
    from .PipxProcess import PipxProcess as EricProcess

from .PipxExecDialog import PipxExecDialog


class Pipx(QObject):
    """
    Class implementing the pipx interface.

    @signal outdatedPackage(package:str, latestVer:str, oudatedDeps:bool) emitted with
    the result of a check for outdated status of a package
    """

    outdatedPackage = pyqtSignal(str, str, bool)

    def __init__(self, parent=None):
        """
        Constructor

        @param parent reference to the user interface object
        @type QObject
        """
        super().__init__(parent)

        self.__ui = parent

        self.__pipxProcesses = []

    ############################################################################
    ## Utility methods
    ############################################################################

    def getPipxVersion(self):
        """
        Public method to get the version of the installed pipx package.

        @return string containing the pipx version number
        @rtype str
        """
        ok, output = self.__runPipxProcess(["--version"])
        if ok:
            return output.strip()
        else:
            return ""

    def getPipxVersionTuple(self):
        """
        Public method to get the version tuple of the installed pipx package.

        @return tuple containing the elements of the pipx version number
        @rtype tuple of (int, int, int)
        """
        from pipx.version import version_tuple  # noqa: I102

        return version_tuple

    def getPipxPaths(self):
        """
        Public method to get the paths used by pipx.

        @return dictionary containing the various pipx paths. The keys are
            'venvsPath', 'appsPath' and 'manPath'.
        @rtype dict[str, Path]
        """
        from pipx.paths import ctx  # noqa: I102

        return {
            "venvsPath": ctx.venvs,
            "appsPath": ctx.bin_dir,
            "manPath": ctx.man_dir,
            "pythonPath": ctx.standalone_python_cachedir,
        }

    def getPipxStrPaths(self):
        """
        Public method to get the paths used by pipx.

        @return dictionary containing the various pipx paths. The keys are
            'venvsPath', 'appsPath' and 'manPath'.
        @rtype dict[str, str]
        """
        from pipx.paths import ctx  # noqa: I102

        return {
            "venvsPath": str(ctx.venvs),
            "appsPath": str(ctx.bin_dir),
            "manPath": str(ctx.man_dir),
            "pythonPath": str(ctx.standalone_python_cachedir),
        }

    def __getPipxExecutable(self):
        """
        Private method to get the path name of the pipx executable.

        @return path name of the pipx executable
        @rtype str
        """
        binDir = sysconfig.get_path("scripts")
        pipx = os.path.join(binDir, "pipx")
        if OSUtilities.isWindowsPlatform():
            pipx += ".exe"

        return pipx

    def __runPipxProcess(self, args):
        """
        Private method to execute pipx with the given arguments.

        @param args list of command line arguments for pipx
        @type list of str
        @return tuple containing a flag indicating success and the output
            of the process
        @rtype tuple of (bool, str)
        """
        ioEncoding = Preferences.getSystem("IOEncoding")

        process = QProcess()
        process.start(self.__getPipxExecutable(), args)
        procStarted = process.waitForStarted()
        if procStarted:
            finished = process.waitForFinished(30000)
            if finished:
                if process.exitCode() == 0:
                    output = str(process.readAllStandardOutput(), ioEncoding, "replace")
                    return True, output
                else:
                    error = str(process.readAllStandardError(), ioEncoding, "replace")
                    msg = self.tr("<p>Message: {0}</p>").format(error) if error else ""
                    return (
                        False,
                        self.tr("<p>pipx exited with an error ({0}).</p>{1}").format(
                            process.exitCode(), msg
                        ),
                    )
            else:
                process.terminate()
                process.waitForFinished(2000)
                process.kill()
                process.waitForFinished(3000)
                return False, self.tr("pipx did not finish within 30 seconds.")

        return False, self.tr("pipx could not be started.")

    def __metadataDecoderHook(self, jsonDict):
        """
        Private method to allow the JSON decoding of Path objects of a spec metadata
        file as created by 'pipx list --json'.

        @param jsonDict JSON dictionary to be decoded
        @type dict
        @return decoded Path object or the dictionary unaltered
        @rtype dict or pathlib.Path
        """
        if jsonDict.get("__type__") == "Path" and "__Path__" in jsonDict:
            return pathlib.Path(jsonDict["__Path__"])
        return jsonDict

    def __runPipxAsyncProcess(self, args, callback=None, timeout=30000):
        """
        Private method to execute pipx with the given arguments asynchronously.

        @param args list of command line arguments for pipx
        @type list of str
        @param callback reference to the function to be called a success flag and the
            process output or error message (defaults to None)
        @type function (optional)
        @param timeout timeout for the process in milliseconds (defaults to 30000)
        @type int (optional)
        @return reference to the generated process object
        @rtype QProcess
        """
        process = EricProcess(timeout=timeout)
        process.finished.connect(
            functools.partial(self.__asyncProcessFinished, callback, process)
        )
        process.errorOccurred.connect(
            functools.partial(self.__asyncProcessError, process)
        )
        self.__pipxProcesses.append(process)
        process.start(self.__getPipxExecutable(), args)
        return process

    def __asyncProcessError(self, process, error):
        """
        Private method to handle a process error signal.

        @param process reference to the process
        @type QProcess
        @param error error that occurred
        @type QProcess.ProcessError
        """
        if error == QProcess.ProcessError.FailedToStart:
            with contextlib.suppress(ValueError):
                self.__pipxProcesses.remove(process)
            EricMessageBox.critical(
                None, self.tr("pipx Start Error"), self.tr("pipx could not be started.")
            )
        else:
            EricMessageBox.critical(
                None,
                self.tr("pipx Runtime Error"),
                self.tr(
                    "<p>The pipx process reported an error.</p><p>Error: {0}</p>"
                ).format(process.errorString()),
            )

    def __asyncProcessFinished(self, callback, process, _exitCode, exitStatus):
        """
        Private method to handle the process finished signal.

        @param callback reference to the function to be called a success flag and the
            process output or error message
        @type function
        @param process reference to the process
        @type QProcess
        @param _exitCode exit code of the process
        @type int
        @param exitStatus exit status of the process
        @type QProcess.ExitStatus
        """
        if process.timedOut():
            msg = self.tr("pipx did not finish within {0} seconds.").format(
                process.timeoutInterval() // 1_000
            )
            if callback:
                callback(False, msg)
            else:
                EricMessageBox.critical(None, self.tr("pipx Timeout Error"), msg)

        elif exitStatus == QProcess.ExitStatus.NormalExit:
            ioEncoding = Preferences.getSystem("IOEncoding")
            if process.exitCode() == 0:
                output = str(process.readAllStandardOutput(), ioEncoding, "replace")
                if callback:
                    callback(True, output)
            else:
                error = str(process.readAllStandardError(), ioEncoding, "replace")
                msg = self.tr("<p>Message: {0}</p>").format(error) if error else ""
                if callback:
                    callback(
                        False,
                        self.tr("<p>pipx exited with an error ({0}).</p>{1}").format(
                            process.exitCode(), msg
                        ),
                    )
                else:
                    EricMessageBox.critical(
                        None,
                        self.tr("pipx Execution Error"),
                        self.tr("<p>pipx exited with an error ({0}).</p>{1}").format(
                            process.exitCode(), msg
                        ),
                    )

        with contextlib.suppress(ValueError):
            self.__pipxProcesses.remove(process)

    ############################################################################
    ## pipx interpreter list function (modified from original to work here)
    ############################################################################

    def getPipxInterpretersList(self):
        """
        Public method returning a list all standalone interpreters.

        @return dictionary containing data of standalone interpreters
        @rtype dict
        """
        from pipx.commands.interpreter import (  # noqa: I102
            get_installed_standalone_interpreters,
            get_interpreter_users,
            get_venvs_using_standalone_interpreter,
        )
        from pipx.paths import ctx  # noqa: I102
        from pipx.venv import VenvContainer  # noqa: I102

        interpreters = get_installed_standalone_interpreters()
        venvs = get_venvs_using_standalone_interpreter(VenvContainer(ctx.venvs))

        interpretersDict = {
            "path": str(ctx.standalone_python_cachedir),
            "interpreters": {},
        }
        for interpreter in interpreters:
            usedBy = get_interpreter_users(interpreter, venvs)
            interpretersDict["interpreters"][interpreter.name] = {
                "used": bool(usedBy),
                "used_by": [
                    (p.main_package.package, p.main_package.package_version)
                    for p in usedBy
                ],
            }
        return interpretersDict

    ############################################################################
    ## Command methods
    ############################################################################

    def getInstalledPackages(self):
        """
        Public method to get the installed packages.

        @return list of dictionaries containing the installed packages and apps
        @rtype list of dict[str, str | list]
        """
        from pipx.paths import ctx  # noqa: I102

        packages = []

        ok, output = self.__runPipxProcess(["list", "--json"])
        if ok and output:
            with contextlib.suppress(json.JSONDecodeError):
                data = json.loads(output, object_hook=self.__metadataDecoderHook)
                for venvName in data["venvs"]:
                    metadata = data["venvs"][venvName]["metadata"]
                    package = {
                        "name": venvName,
                        "version": metadata["main_package"]["package_version"],
                        "apps": [],
                        "python": metadata["python_version"],
                        "is_standalone": (
                            str(metadata["source_interpreter"]).startswith(
                                str(ctx.standalone_python_cachedir.resolve())
                            )
                            if metadata["source_interpreter"]
                            else False
                        ),
                    }
                    for appPath in metadata["main_package"]["app_paths"]:
                        package["apps"].append((appPath.name, str(appPath)))
                    packages.append(package)

        return packages

    def installPackages(
        self,
        packages,
        interpreterVersion="",
        fetchMissingInterpreter=False,
        forceVenvModification=False,
        systemSitePackages=False,
    ):
        """
        Public method to install a list of packages with the given options.

        @param packages list of packages to install
        @type list of str
        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        @param forceVenvModification flag indicating to allow modification of already
            existing virtual environments (defaults to False)
        @type bool (optional)
        @param systemSitePackages flag indicating to give access to the system
            site-packages directory (defaults to False)
        @type bool (optional)
        """
        args = ["install"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        if forceVenvModification:
            args.append("--force")
        if systemSitePackages:
            args.append("--system-site-packages")
        args += packages
        dia = PipxExecDialog(self.tr("Install Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def installAllPackages(
        self,
        specFile,
        interpreterVersion="",
        fetchMissingInterpreter=False,
        forceVenvModification=False,
        systemSitePackages=False,
    ):
        """
        Public method to install all packages define by a given spec metadata file
        with given options.

        @param specFile path of the spec metadata file
        @type str
        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        @param forceVenvModification flag indicating to allow modification of already
            existing virtual environments (defaults to False)
        @type bool (optional)
        @param systemSitePackages flag indicating to give access to the system
            site-packages directory (defaults to False)
        @type bool (optional)
        """
        args = ["install-all"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        if forceVenvModification:
            args.append("--force")
        if systemSitePackages:
            args.append("--system-site-packages")
        args.append(specFile)
        dia = PipxExecDialog(self.tr("Install All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def createSpecMetadataFile(self, specFile):
        """
        Public method to create a spec metadata file.

        @param specFile path of the spec metadata file
        @type str
        @return tuple containing a flag indicating success and an error message in case
            of failure
        @rtype tuple of (bool, str)
        """
        ok, output = self.__runPipxProcess(["list", "--json"])
        if ok:
            try:
                with open(specFile, "w") as f:
                    f.write(output)
                    return True, ""
            except IOError as err:
                return False, str(err)
        else:
            return False, output

    def reinstallPackage(
        self,
        package,
        interpreterVersion="",
        fetchMissingInterpreter=False,
    ):
        """
        Public method to reinstall the given package with given options.

        @param package name of the package to reinstall
        @type str
        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        """
        args = ["reinstall"]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        args.append(package)
        dia = PipxExecDialog(self.tr("Re-Install Package"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def reinstallAllPackages(
        self,
        interpreterVersion="",
        fetchMissingInterpreter=False,
        skipPackages=None,
    ):
        """
        Public method to reinstall all packages with given options.

        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        @param skipPackages list of packages to be skipped by the 'reinstall-all'
            command (defaults to None)
        @type list of str (optional)
        """
        args = ["reinstall-all"]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        if skipPackages:
            args += ["--skip"] + skipPackages
        dia = PipxExecDialog(self.tr("Re-Install All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def uninstallPackage(self, package):
        """
        Public method to uninstall the given package.

        @param package name of the package to be uninstalled
        @type str
        """
        args = ["uninstall", package]
        dia = PipxExecDialog(self.tr("Uninstall Package"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def uninstallAllPackages(self):
        """
        Public method to uninstall all pipx managed packages.
        """
        args = ["uninstall-all"]
        dia = PipxExecDialog(self.tr("Uninstall All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def upgradePackage(self, package):
        """
        Public method to upgrade the given package.

        @param package name of the package
        @type str
        """
        args = ["upgrade"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        args.append(package)
        dia = PipxExecDialog(self.tr("Upgrade Package"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def upgradeAllPackages(self):
        """
        Public method to upgrade all package.
        """
        args = ["upgrade-all"]
        dia = PipxExecDialog(self.tr("Upgrade All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def upgradeSharedLibraries(self):
        """
        Public method to upgrade shared libraries.
        """
        args = ["upgrade-shared"]
        dia = PipxExecDialog(self.tr("Upgrade Shared Libraries"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def upgradeInterpreters(self, dialogParent=None):
        """
        Public method to upgrade the installed interpreters to the latest available
        micro/patch version.

        @param dialogParent parent widget of the execution dialog
        @type QWidget
        """
        args = ["interpreter", "upgrade"]
        dia = PipxExecDialog(self.tr("Upgrade Interpreters"), parent=dialogParent)
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def pruneInterpreters(self, dialogParent=None):
        """
        Public method to prune unused interpreters.

        @param dialogParent parent widget of the execution dialog
        @type QWidget
        """
        args = ["interpreter", "prune"]
        dia = PipxExecDialog(self.tr("Prune Unused Interpreters"), parent=dialogParent)
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def ensurePath(self):
        """
        Public method to ensure that the directory where pipx stores apps is
        in your PATH environment variable.
        """
        args = ["ensurepath"]
        dia = PipxExecDialog(self.tr("Ensure PATH Modifications"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    ############################################################################
    ## Special methods based on 'runpip'
    ############################################################################

    def checkPackageOutdated(self, package):
        """
        Public method to check, if a given package is outdated.

        @param package name of the package
        @type str
        """
        args = ["runpip", package, "list", "--outdated", "--format", "json"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        self.__runPipxAsyncProcess(
            args, callback=functools.partial(self.__checkPackageOutdatedCb, package)
        )

    def __checkPackageOutdatedCb(self, package, ok, output):
        """
        Private method handling the pipx process output of a check for an outdated
        package.

        @param package name of the package
        @type str
        @param ok flag indicating the process ended successfully
        @type bool
        @param output output of the pipx process or an error message
        @type str
        """
        if not ok:
            EricMessageBox.information(
                None,
                self.tr("Check Outdated Package"),
                self.tr(
                    "<p>The status of package <b>{0}</b> could not be determined.</p>"
                    "<p>Reason: {1}</p>"
                ).format(package, output),
            )
            self.outdatedPackage.emit(package, "", False)
            return

        outdatedList = json.loads(output)
        # check if the main package is in the list
        for outdatedPackage in outdatedList:
            if outdatedPackage["name"] == package:
                self.outdatedPackage.emit(
                    package, outdatedPackage["latest_version"], len(outdatedList) > 1
                )
                return

        self.outdatedPackage.emit(package, "", bool(outdatedList))

    def getPackageVersion(self, package):
        """
        Public method to get the version of a package.

        @param package package name
        @type str
        @return package version
        @rtype str
        """
        packagesList = self.__getPackageDependencies(package=package)
        for pack in packagesList:
            if pack["name"] == package:
                return pack["version"]
        else:
            return ""

    def __getPackageDependencies(self, package, uptodate=False, outdated=False):
        """
        Private method to get a list of dependencies of a given package.

        @param package name of the package
        @type str
        @param uptodate DESCRIPTION (defaults to False)
        @type TYPE (optional)
        @param outdated DESCRIPTION (defaults to False)
        @type TYPE (optional)
        @return list of dictionaries as returned by 'pip'
        @rtype list[dict[str: str]]
        """
        if outdated:
            args = ["runpip", package, "list", "--format", "json", "--outdated"]
        elif uptodate:
            args = ["runpip", package, "list", "--format", "json", "--uptodate"]
        else:
            args = ["runpip", package, "list", "--format", "json"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        ok, output = self.__runPipxProcess(args)
        if not ok:
            EricMessageBox.information(
                None,
                self.tr("Get Package Dependencies"),
                self.tr(
                    "<p>The status of dependencies of package <b>{0}</b> could not"
                    " be determined.</p><p>Reason: {1}</p>"
                ).format(package, output),
            )
            return []

        return json.loads(output)

    def getOutdatedPackageDependencies(self, package):
        """
        Public method to get the list of outdated package dependencies.

        @param package name of the package
        @type str
        @return list of tuples containing the dependency name, version and latest
            version
        @rtype list of tuple of (str, str, str)
        """
        outdatedList = self.__getPackageDependencies(package=package, outdated=True)
        return [
            (d["name"], d["version"], d["latest_version"])
            for d in outdatedList
            if d["name"] != package
        ]

    def getUptodatePackageDependencies(self, package):
        """
        Public method to get the list of up-to-date package dependencies.

        @param package name of the package
        @type str
        @return list of tuples containing the dependency name and version
        @rtype list of tuple of (str, str)
        """
        uptodateList = self.__getPackageDependencies(package=package, uptodate=True)
        return [(d["name"], d["version"]) for d in uptodateList if d["name"] != package]

    def getAllPackageDependencies(self, package):
        """
        Public method to get the list of package dependencies.

        @param package name of the package
        @type str
        @return list of tuples containing the dependency name and version
        @rtype list of tuple of (str, str)
        """
        dependenciesList = self.__getPackageDependencies(package=package)
        return [
            (d["name"], d["version"]) for d in dependenciesList if d["name"] != package
        ]

    def upgradePackageDependencies(self, package):
        """
        Public method to upgrade the dependencies of the given package.

        @param package name of the package
        @type str
        """
        outdatedDependencies = [
            d[0] for d in self.getOutdatedPackageDependencies(package=package)
        ]
        args = [
            "runpip",
            package,
            "install",
            "--upgrade",
            "--prefer-binary",
        ] + outdatedDependencies

        dia = PipxExecDialog(self.tr("Upgrade Dependencies"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()