Plugins/UiExtensionPlugins/PipInterface/PipListDialog.py

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

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

"""
Module implementing a dialog to list installed packages.
"""

from __future__ import unicode_literals
try:
    str = unicode       # __IGNORE_EXCEPTION__
except NameError:
    pass

import json

from PyQt5.QtCore import pyqtSlot, Qt, QProcess, QTimer
from PyQt5.QtWidgets import QDialog, QDialogButtonBox, QAbstractButton, \
    QApplication, QTreeWidgetItem, QHeaderView

from E5Gui import E5MessageBox

from .Ui_PipListDialog import Ui_PipListDialog

import Preferences


class PipListDialog(QDialog, Ui_PipListDialog):
    """
    Class implementing a dialog to list installed packages.
    """
    CommandArguments = {
        "list": ["list", "--format=json"],
        "uptodate": ["list", "--uptodate", "--format=json"],
        "outdated": ["list", "--outdated", "--format=json"],
    }
    
    ShowProcessGeneralMode = 0
    ShowProcessClassifiersMode = 1
    ShowProcessEntryPointsMode = 2
    ShowProcessFilesListMode = 3
    
    def __init__(self, pip, mode, indexUrl, title, parent=None):
        """
        Constructor
        
        @param pip reference to the master object
        @type Pip
        @param mode list command mode (one of 'list', 'uptodate', 'outdated')
        @type str
        @param indexUrl URL of the pypi index
        @type str
        @param title title of the dialog
        @type str
        @param parent reference to the parent widget
        @type QWidget
        """
        assert mode in PipListDialog.CommandArguments
        
        super(PipListDialog, self).__init__(parent)
        self.setupUi(self)
        self.setWindowFlags(Qt.Window)
        
        self.setWindowTitle(title)
        
        self.__refreshButton = self.buttonBox.addButton(
            self.tr("&Refresh"), QDialogButtonBox.ActionRole)
        self.__refreshButton.setEnabled(False)
        if mode == "outdated":
            self.__upgradeButton = self.buttonBox.addButton(
                self.tr("Up&grade"), QDialogButtonBox.ActionRole)
            self.__upgradeButton.setEnabled(False)
            self.__upgradeAllButton = self.buttonBox.addButton(
                self.tr("Upgrade &All"), QDialogButtonBox.ActionRole)
            self.__upgradeAllButton.setEnabled(False)
        else:
            self.__upgradeButton = None
            self.__upgradeAllButton = None
        self.__uninstallButton = self.buttonBox.addButton(
            self.tr("&Uninstall"), QDialogButtonBox.ActionRole)
        self.__uninstallButton.setEnabled(False)
        
        self.__pip = pip
        self.__mode = mode
        self.__ioEncoding = Preferences.getSystem("IOEncoding")
        self.__indexUrl = indexUrl
        self.__errors = ""
        self.__output = []
        
        self.__nothingStrings = {
            "list": self.tr("Nothing to show"),
            "uptodate": self.tr("All packages outdated"),
            "outdated": self.tr("All packages up-to-date"),
        }
        
        self.venvComboBox.addItem(self.__pip.getDefaultEnvironmentString())
        self.venvComboBox.addItems(self.__pip.getVirtualenvNames())
        
        if mode == "list":
            self.infoLabel.setText(self.tr("Installed Packages:"))
            self.packageList.setHeaderLabels([
                self.tr("Package"),
                self.tr("Version"),
            ])
        elif mode == "uptodate":
            self.infoLabel.setText(self.tr("Up-to-date Packages:"))
            self.packageList.setHeaderLabels([
                self.tr("Package"),
                self.tr("Version"),
            ])
        elif mode == "outdated":
            self.infoLabel.setText(self.tr("Outdated Packages:"))
            self.packageList.setHeaderLabels([
                self.tr("Package"),
                self.tr("Current Version"),
                self.tr("Latest Version"),
                self.tr("Package Type"),
            ])
        
        self.packageList.header().setSortIndicator(0, Qt.AscendingOrder)
        
        self.__infoLabels = {
            "name": self.tr("Name:"),
            "version": self.tr("Version:"),
            "location": self.tr("Location:"),
            "requires": self.tr("Requires:"),
            "summary": self.tr("Summary:"),
            "home-page": self.tr("Homepage:"),
            "author": self.tr("Author:"),
            "author-email": self.tr("Author Email:"),
            "license": self.tr("License:"),
            "metadata-version": self.tr("Metadata Version:"),
            "installer": self.tr("Installer:"),
            "classifiers": self.tr("Classifiers:"),
            "entry-points": self.tr("Entry Points:"),
            "files": self.tr("Files:"),
        }
        self.infoWidget.setHeaderLabels(["Key", "Value"])
        
        self.process = QProcess()
        self.process.finished.connect(self.__procFinished)
        self.process.readyReadStandardOutput.connect(self.__readStdout)
        self.process.readyReadStandardError.connect(self.__readStderr)
        
        self.show()
        QApplication.processEvents()
    
    def __stopProcess(self):
        """
        Private slot to stop the running process.
        """
        if self.process.state() != QProcess.NotRunning:
            self.process.terminate()
            QTimer.singleShot(2000, self.process.kill)
            self.process.waitForFinished(3000)
        
        QApplication.restoreOverrideCursor()
    
    def closeEvent(self, e):
        """
        Protected slot implementing a close event handler.
        
        @param e close event
        @type QCloseEvent
        """
        self.__stopProcess()
        e.accept()
    
    def __finish(self):
        """
        Private slot called when the process finished or the user pressed
        the cancel button.
        """
        self.__stopProcess()
        
        self.__processOutput()
        
        self.buttonBox.button(QDialogButtonBox.Close).setEnabled(True)
        self.buttonBox.button(QDialogButtonBox.Cancel).setEnabled(False)
        self.buttonBox.button(QDialogButtonBox.Close).setDefault(True)
        self.buttonBox.button(QDialogButtonBox.Close).setFocus(
            Qt.OtherFocusReason)
        self.__refreshButton.setEnabled(True)
        
        if self.packageList.topLevelItemCount() == 0:
            QTreeWidgetItem(self.packageList,
                            [self.__nothingStrings[self.__mode]])
            if self.__errors and not self.__errors.startswith("DEPRECATION"):
                E5MessageBox.critical(
                    self,
                    self.windowTitle(),
                    self.tr("""<p>The command failed.</p>"""
                            """<p>Reason: {0}</p>""").format(
                        self.__errors.replace("\r\n", "<br/>")
                            .replace("\n", "<br/>").replace("\r", "<br/>")
                            .replace(" ", "&nbsp;")))
            if self.__upgradeAllButton is not None:
                self.__upgradeAllButton.setEnabled(False)
        else:
            if self.__upgradeAllButton is not None:
                self.__upgradeAllButton.setEnabled(True)
        
        self.packageList.sortItems(
            0,
            self.packageList.header().sortIndicatorOrder())
        self.packageList.header().resizeSections(
            QHeaderView.ResizeToContents)
        self.packageList.header().setStretchLastSection(True)
    
    @pyqtSlot(QAbstractButton)
    def on_buttonBox_clicked(self, button):
        """
        Private slot called by a button of the button box clicked.
        
        @param button button that was clicked
        @type QAbstractButton
        """
        if button == self.buttonBox.button(QDialogButtonBox.Close):
            self.close()
        elif button == self.buttonBox.button(QDialogButtonBox.Cancel):
            self.__finish()
        elif button == self.__refreshButton:
            self.__refresh()
        elif button == self.__upgradeButton:
            self.__upgradePackages()
        elif button == self.__upgradeAllButton:
            self.__upgradeAllPackages()
        elif button == self.__uninstallButton:
            self.__uninstallPackages()
    
    def __procFinished(self, exitCode, exitStatus):
        """
        Private slot connected to the finished signal.
        
        @param exitCode exit code of the process
        @type int
        @param exitStatus exit status of the process
        @type QProcess.ExitStatus
        """
        self.__finish()
    
    def __refresh(self):
        """
        Private slot to refresh the displayed list.
        """
        self.__stopProcess()
        self.start()
    
    def start(self):
        """
        Public method to start the command.
        """
        self.packageList.clear()
        self.__errors = ""
        self.__output = []
        
        self.buttonBox.button(QDialogButtonBox.Close).setEnabled(False)
        self.buttonBox.button(QDialogButtonBox.Cancel).setEnabled(True)
        self.buttonBox.button(QDialogButtonBox.Cancel).setDefault(True)
        self.__refreshButton.setEnabled(False)
        if self.__upgradeAllButton is not None:
            self.__upgradeAllButton.setEnabled(False)
        QApplication.processEvents()
        
        QApplication.setOverrideCursor(Qt.WaitCursor)
        QApplication.processEvents()
        
        venvName = self.venvComboBox.currentText()
        interpreter = self.__pip.getVirtualenvInterpreter(venvName)
        if not interpreter:
            return
        
        args = ["-m", "pip"] + PipListDialog.CommandArguments[self.__mode]
        if self.localCheckBox.isChecked():
            args.append("--local")
        if self.notRequiredCheckBox.isChecked():
            args.append("--not-required")
        if self.userCheckBox.isChecked():
            args.append("--user")
        
        if self.__indexUrl:
            args.append("--index-url")
            args.append(self.__indexUrl + "/simple")
        
        self.process.start(interpreter, args)
        procStarted = self.process.waitForStarted(5000)
        if not procStarted:
            self.buttonBox.setFocus()
            self.__stopProcess()
            E5MessageBox.critical(
                self,
                self.tr('Process Generation Error'),
                self.tr(
                    'The process {0} could not be started.'
                ).format(interpreter))
            self.__finish()
    
    def __processOutput(self):
        """
        Private method to process the captured output.
        """
        if self.__output:
            try:
                packageData = json.loads("\n".join(self.__output))
                for package in packageData:
                    data = [
                        package["name"],
                        package["version"],
                    ]
                    if self.__mode == "outdated":
                        data.extend([
                            package["latest_version"],
                            package["latest_filetype"],
                        ])
                    QTreeWidgetItem(self.packageList, data)
            except ValueError as err:
                self.__errors += str(err) + "\n"
                self.__errors += "received output:\n"
                self.__errors += "\n".join(self.__output)
    
    def __readStdout(self):
        """
        Private slot to handle the readyReadStandardOutput signal.
        
        It reads the output of the process, formats it and inserts it into
        the contents pane.
        """
        self.process.setReadChannel(QProcess.StandardOutput)
        
        while self.process.canReadLine():
            line = str(self.process.readLine(), self.__ioEncoding,
                       'replace').strip()
            self.__output.append(line)
    
    def __readStderr(self):
        """
        Private slot to handle the readyReadStandardError signal.
        
        It reads the error output of the process and inserts it into the
        error pane.
        """
        self.__errors += str(self.process.readAllStandardError(),
                             self.__ioEncoding, 'replace')
    
    @pyqtSlot(str)
    def on_venvComboBox_activated(self, txt):
        """
        Private slot handling the selection of a virtual environment.
        
        @param txt virtual environment
        @type str
        """
        self.__refresh()
    
    @pyqtSlot(bool)
    def on_localCheckBox_clicked(self, checked):
        """
        Private slot handling the switching of the local mode.
        
        @param checked state of the local check box
        @type bool
        """
        self.__refresh()
    
    @pyqtSlot(bool)
    def on_notRequiredCheckBox_clicked(self, checked):
        """
        Private slot handling the switching of the 'not required' mode.
        
        @param checked state of the 'not required' check box
        @type bool
        """
        self.__refresh()
    
    @pyqtSlot(bool)
    def on_userCheckBox_clicked(self, checked):
        """
        Private slot handling the switching of the 'user-site' mode.
        
        @param checked state of the 'user-site' check box
        @type bool
        """
        self.__refresh()
    
    @pyqtSlot()
    def on_packageList_itemSelectionChanged(self):
        """
        Private slot handling the selection of a package.
        """
        self.infoWidget.clear()
        
        if len(self.packageList.selectedItems()) == 1:
            itm = self.packageList.selectedItems()[0]
            
            environment = self.venvComboBox.currentText()
            interpreter = self.__pip.getVirtualenvInterpreter(environment)
            if not interpreter:
                return
            
            QApplication.setOverrideCursor(Qt.WaitCursor)
            
            args = ["-m", "pip", "show"]
            if self.verboseCheckBox.isChecked():
                args.append("--verbose")
            if self.installedFilesCheckBox.isChecked():
                args.append("--files")
            args.append(itm.text(0))
            success, output = self.__pip.runProcess(args, interpreter)
            
            if success and output:
                mode = PipListDialog.ShowProcessGeneralMode
                for line in output.splitlines():
                    line = line.rstrip()
                    if line != "---":
                        if mode != PipListDialog.ShowProcessGeneralMode:
                            if line[0] == " ":
                                QTreeWidgetItem(
                                    self.infoWidget,
                                    [" ", line.strip()])
                            else:
                                mode = PipListDialog.ShowProcessGeneralMode
                        if mode == PipListDialog.ShowProcessGeneralMode:
                            try:
                                label, info = line.split(": ", 1)
                            except ValueError:
                                label = line[:-1]
                                info = ""
                            label = label.lower()
                            if label in self.__infoLabels:
                                QTreeWidgetItem(
                                    self.infoWidget,
                                    [self.__infoLabels[label], info])
                            if label == "files":
                                mode = PipListDialog.ShowProcessFilesListMode
                            elif label == "classifiers":
                                mode = PipListDialog.ShowProcessClassifiersMode
                            elif label == "entry-points":
                                mode = PipListDialog.ShowProcessEntryPointsMode
                self.infoWidget.scrollToTop()
            
            header = self.infoWidget.header()
            header.setStretchLastSection(False)
            header.resizeSections(QHeaderView.ResizeToContents)
            if header.sectionSize(0) + header.sectionSize(1) < header.width():
                header.setStretchLastSection(True)
            
            QApplication.restoreOverrideCursor()
        
        enable = (len(self.packageList.selectedItems()) > 1 or
                  (len(self.packageList.selectedItems()) == 1 and
                   self.packageList.selectedItems()[0].text(0) not in
                   self.__nothingStrings.values()))
        self.__upgradeButton and self.__upgradeButton.setEnabled(enable)
        self.__uninstallButton.setEnabled(enable)
    
    @pyqtSlot(bool)
    def on_verboseCheckBox_clicked(self, checked):
        """
        Private slot to handle a change of the verbose package information
        checkbox.
        
        @param checked state of the checkbox
        @type bool
        """
        self.on_packageList_itemSelectionChanged()
    
    @pyqtSlot(bool)
    def on_installedFilesCheckBox_clicked(self, checked):
        """
        Private slot to handle a change of the installed files information
        checkbox.
        
        @param checked state of the checkbox
        @type bool
        """
        self.on_packageList_itemSelectionChanged()
    
    def __upgradePackages(self):
        """
        Private slot to upgrade the selected packages.
        """
        packages = []
        for itm in self.packageList.selectedItems():
            packages.append(itm.text(0))
        
        if packages:
            if "pip" in packages:
                self.__upgradePip()
            else:
                self.__executeUpgradePackages(packages)
    
    def __upgradeAllPackages(self):
        """
        Private slot to upgrade all listed packages.
        """
        packages = []
        for index in range(self.packageList.topLevelItemCount()):
            itm = self.packageList.topLevelItem(index)
            packages.append(itm.text(0))
        
        if packages:
            if "pip" in packages:
                self.__upgradePip()
            else:
                self.__executeUpgradePackages(packages)
    
    def __upgradePip(self):
        """
        Private slot to upgrade pip itself.
        """
        res = self.__pip.upgradePip(
            venvName=self.venvComboBox.currentText(),
            userSite=self.userCheckBox.isChecked())
        if res:
            self.__refresh()
    
    def __executeUpgradePackages(self, packages):
        """
        Private method to execute the pip upgrade command.
        
        @param packages list of package names to be upgraded
        @type list of str
        """
        res = self.__pip.upgradePackages(
            packages, venvName=self.venvComboBox.currentText(),
            userSite=self.userCheckBox.isChecked())
        if res:
            self.activateWindow()
            self.raise_()
            self.__refresh()
    
    def __uninstallPackages(self):
        """
        Private slot to uninstall the selected packages.
        """
        packages = []
        for itm in self.packageList.selectedItems():
            packages.append(itm.text(0))
        
        if packages:
            res = self.__pip.uninstallPackages(
                packages,
                venvName=self.venvComboBox.currentText())
            if res:
                self.activateWindow()
                self.raise_()
                self.__refresh()