lubuntu-update-notifier/lubuntu-upgrader

#!/usr/bin/python3
# coding=utf-8

# Copyright (C) 2019 Hans P. Möller <hmollercl@lubuntu.me>
# Copyright (C) 2022 Simon Quigley <tsimonq2@lubuntu.me>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

# deppend on
# aptdaemon
# debconf-kde-helper
import sys
import os
from pathlib import Path
import gettext
from argparse import ArgumentParser
from PyQt5.QtWidgets import (QWidget, QApplication, QLabel, QPushButton,
                             QHBoxLayout, QVBoxLayout, QProgressBar,
                             QPlainTextEdit, QMessageBox)
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QIcon, QTextCursor, QPalette

from aptdaemon import client
from aptdaemon.errors import NotAuthorizedError, TransactionFailed
from aptdaemon.enums import (EXIT_SUCCESS,
                             EXIT_FAILED,
                             STATUS_COMMITTING,
                             get_error_description_from_enum,
                             get_error_string_from_enum,
                             get_status_string_from_enum)



class DialogUpg(QWidget):
    '''UI'''
    def __init__(self, options=None):
        QWidget.__init__(self)

        self.initUI()
        self.closeBtn.clicked.connect(self.call_reject)
        self.apt_client = client.AptClient()
        self.downloadText = ""
        self.detailText = ""
        self.old_short_desc = ""
        self.details = ""
        self.status = ""
        self.errors = []
        # TODO make a terminal work to see more info
        # self.master, self.slave = pty.openpty()
        '''proc = subprocess.Popen(['qterminal'],
                            stdin=self.slave,
                            #stdout=subprocess.PIPE,
                            stdout=self.slave,
                            #stderr=subprocess.PIPE
                            stderr=self.slave)'''

        self.trans3 = self.apt_client.fix_incomplete_install()
        self.repair_install()
        if options.fullUpgrade:
            self.trans2 = self.apt_client.upgrade_system(safe_mode=False)
            self.setWindowTitle('Full Upgrade')
        else:
            self.trans2 = self.apt_client.upgrade_system(safe_mode=True)

        if os.geteuid() == 0:
            if options.cacheUpdate:
                self.trans1 = self.apt_client.update_cache()
                self.update_cache()
            else:
                self.upgrade()

    def initUI(self):
        '''initialize UI'''
        self.label = QLabel()
        self.label.setAlignment(Qt.AlignHCenter)
        self.closeBtn = QPushButton("Close")
        self.progressBar = QProgressBar()
        self.plainTextEdit = QPlainTextEdit()
        palette = self.plainTextEdit.palette()
        palette.setColor(QPalette.Base, Qt.black)
        palette.setColor(QPalette.Text, Qt.gray)
        self.plainTextEdit.setPalette(palette)

        hbox = QHBoxLayout()
        hbox.addStretch(1)
        hbox.addWidget(self.closeBtn)
        hbox.addStretch(1)

        vbox = QVBoxLayout()
        vbox.addWidget(self.label)
        vbox.addWidget(self.progressBar)
        vbox.addWidget(self.plainTextEdit)
        vbox.addLayout(hbox)

        self.setLayout(vbox)
        self.setGeometry(300, 300, 500, 150)
        self.setWindowTitle('Upgrade')
        self.progressBar.setVisible(False)
        self.plainTextEdit.setReadOnly(True)
        self.plainTextEdit.setVisible(False)
        self.center()

    def center(self):
        '''centers UI'''
        frameGm = self.frameGeometry()
        screen = QApplication.desktop().screenNumber(
            QApplication.desktop().cursor().pos())
        centerPoint = QApplication.desktop().screenGeometry(screen).center()
        frameGm.moveCenter(centerPoint)
        self.move(frameGm.topLeft())

    def upgrade_progress(self, transaction, progress):
        '''upgrade progressbar during upgrade'''
        self.progressBar.setVisible(True)
        self.progressBar.setValue(progress)

    def update_progress(self, transaction, progress):
        '''upgrade progressbar during update'''
        self.progressBar.setVisible(True)
        self.progressBar.setValue(progress)
        self.label.setText(_("Updating cache..."))

    def repair_progress(self, transaction, progress):
        '''upgrade progressbar during update'''
        self.progressBar.setVisible(True)
        self.progressBar.setValue(progress)
        self.label.setText(_("Repairing interrupted upgrade if necessary..."))

    def update_progress_download(self, transaction, uri, status, short_desc,
                                 total_size, current_size, msg):
        '''print update info'''
        self.plainTextEdit.setVisible(True)
        if self.old_short_desc == short_desc:
            # if it's the same file we update the line, don't append new line
            self.plainTextEdit.moveCursor(QTextCursor.End)
            cursor = self.plainTextEdit.textCursor()
            cursor.movePosition(QTextCursor.End, QTextCursor.MoveAnchor)
            cursor.select(QTextCursor.LineUnderCursor)
            cursor.removeSelectedText()
            self.plainTextEdit.insertPlainText(str(current_size) + "/"
                                               + str(total_size) + " " + msg)
            cursor.movePosition(QTextCursor.End, QTextCursor.MoveAnchor)
        else:
            self.plainTextEdit.moveCursor(QTextCursor.End)
            self.plainTextEdit.appendPlainText(status + " " + short_desc +
                                               "\n")
            self.plainTextEdit.insertPlainText(str(current_size) + "/"
                                               + str(total_size) + " " + msg)
            self.plainTextEdit.moveCursor(QTextCursor.End)
            self.old_short_desc = short_desc

    def upgrade_progress_download(self, transaction, uri, status, short_desc,
                                  total_size, current_size, msg):
        '''print upgrade info'''
        self.plainTextEdit.setVisible(True)
        if self.status == "status-downloading":
            # TODO it prints the last line after installation is complete.
            if self.old_short_desc == short_desc:
                # if it's the same file we update the line, don't append
                self.plainTextEdit.moveCursor(QTextCursor.End)
                cursor = self.plainTextEdit.textCursor()
                cursor.movePosition(QTextCursor.End, QTextCursor.MoveAnchor)
                cursor.select(QTextCursor.LineUnderCursor)
                cursor.removeSelectedText()
                self.plainTextEdit.insertPlainText(str(current_size) + "/"
                                                   + str(total_size) + " "
                                                   + msg)
                cursor.movePosition(QTextCursor.End, QTextCursor.MoveAnchor)
            else:
                self.plainTextEdit.moveCursor(QTextCursor.End)
                self.plainTextEdit.appendPlainText(status + " " + short_desc
                                                   + "\n")
                self.plainTextEdit.insertPlainText(str(current_size) + "/"
                                                   + str(total_size) + " "
                                                   + msg)
                self.plainTextEdit.moveCursor(QTextCursor.End)
                self.old_short_desc = short_desc

    def update_progress_detail(self, transaction, current_items, total_items,
                               current_bytes, total_bytes, current_cps, eta):
        '''print update detail info'''
        if total_items > 0:
            self.plainTextEdit.setVisible(True)
            if self.detailText != _("Fetching") + str(
                    current_items) + " " + _("of") + " " + str(total_items):
                self.detailText = (
                    _("Fetching") + " " + str(current_items) + " " + _("of") +
                    " " + str(total_items)
                )
                self.label.setText(self.detailText + "\n" + self.downloadText)

    def upgrade_progress_detail(self, transaction, current_items, total_items,
                                current_bytes, total_bytes, current_cps, eta):
        '''print upgrade detail info'''
        if total_items > 0:
            self.plainTextEdit.setVisible(True)
            if self.detailText != _("Downloaded") + " " + str(
                    current_items) + " " + _("of") + " " + str(total_items):
                self.detailText = (
                    _("Downloaded") + " " + str(current_items) + " " +
                    _("of") + " " + str(total_items)
                )
                self.label.setText(self.detailText + "\n" + self.downloadText)

    def upgrade_finish(self, transaction, exit_state):
        '''when upgrade finish'''
        if exit_state == EXIT_FAILED:
            error_string = get_error_string_from_enum(transaction.error.code)
            error_desc = get_error_description_from_enum(
                transaction.error.code)

        text = _("Upgrade finished")

        reboot_required_path = Path("/var/run/reboot-required")
        if reboot_required_path.exists():
            text = text + "\n" + _("Reboot required")
        self.progressBar.setVisible(False)

        if len(self.errors) > 0:
            text += "\n" + _("With some Errors")
            self.plainTextEdit.appendPlainText(_("Error Resume:") + "\n")
            for error in self.errors:
                self.plainTextEdit.setEnabled(False)
                self.plainTextEdit.insertPlainText(error + "\n")
                self.plainTextEdit.insertPlainText(error_string + "\n")
                self.plainTextEdit.insertPlainText(error_desc + "\n")
                self.plainTextEdit.moveCursor(QTextCursor.End)

        self.label.setText(text)
        self.closeBtn.setVisible(True)
        self.closeBtn.setEnabled(True)
        self.plainTextEdit.setEnabled(True)

    def upgrade_error(self, transaction, error_code, error_details):
        '''if error during upgrade'''
        self.plainTextEdit.setVisible(True)
        self.errors.append("Eror Code: " + str(error_code))
        self.errors.append("Error Detail: " + error_details)
        self.plainTextEdit.setVisible(True)
        self.closeBtn.setEnabled(True)
        print("ECode: " + str(error_code) + "\n")
        print("EDetail: " + error_details + "\n")

    def upgrade_cancellable_changed(self, transaction, cancellable):
        '''when upgrade cancellable toogle'''
        self.closeBtn.setEnabled(cancellable)

    def update_cache(self):
        '''runs cache update'''
        self.closeBtn.setVisible(False)
        try:
            self.trans1.connect('finished', self.update_finish)

            self.trans1.connect('progress-changed', self.update_progress)
            self.trans1.connect('progress-details-changed',
                                self.update_progress_detail)
            self.trans1.connect('progress-download-changed',
                                self.update_progress_download)
            self.trans1.connect('error', self.upgrade_error)
            self.trans1.connect("status-changed", self.status_changed)
            # TODO make a terminal work to see more info
            # self.trans1.set_terminal(os.ttyname(self.slave))

            self.trans1.run()

        except (NotAuthorizedError, TransactionFailed) as e:
            print("Warning: install transaction not completed successfully:"
                  + "{}".format(e))

    def update_finish(self, transaction, exit_state):
        '''when update finish'''
        self.label.setText(_("Update Cache Finished"))
        if exit_state == EXIT_FAILED:
            error_string = get_error_string_from_enum(transaction.error.code)
            error_desc = get_error_description_from_enum(
                transaction.error.code)
            self.plainTextEdit.setEnabled(False)
            self.plainTextEdit.moveCursor(QTextCursor.End)
            self.plainTextEdit.insertPlainText(error_string + "\n")
            self.plainTextEdit.insertPlainText(error_desc + "\n")
            self.plainTextEdit.moveCursor(QTextCursor.End)
            self.plainTextEdit.setEnabled(True)

        self.upgrade()

    def status_changed(self, transaction, status):
        '''print info when status changed'''
        self.status = status
        self.label.setText("Status:" + get_status_string_from_enum(status))
        print("Status:" + get_status_string_from_enum(status) + " " + status
              + "\n")

    def status_details_changed(self, transaction, details):
        '''print status detail info'''
        self.plainTextEdit.setVisible(True)
        if self.details != details:
            self.details = details

            if self.status != "status-downloading":
                '''
                if "Downloading xxxxx" is handled by
                upgrade_progress_download" in short_desc
                '''
                self.plainTextEdit.appendPlainText(details)
                self.plainTextEdit.moveCursor(QTextCursor.End)
                self.label.setText(details)
            else:
                self.label.setText(self.detailText + "\n" + details)
                # if is downloading put the "Downloaded x of y" text
            # print("PTY:" + str(self.slave))
            print("Status Details:" + details)

    def config_file_conflict(self, transaction, cur, new):
        title = "Conflicting Configuration"
        text = "Updating the system will result in the following file being "
        text += "overwritten: " + cur + "\n\nWhat would you like to do?"
        query = QMessageBox()
        query.setWindowTitle(title)
        query.setText(text)
        query.setIcon(QMessageBox.Question)
        query.setStandardButtons(QMessageBox.Yes|QMessageBox.No)
        yes = query.button(QMessageBox.Yes)
        yes.setText("Overwrite")
        no = query.button(QMessageBox.No)
        no.setText("Keep Existing")
        query.setDefaultButton(no)
        query.exec_()

        if query.clickedButton() == yes:
            answer = "replace"
        elif query.clickedButton() == no:
            answer = "keep"

        transaction.resolve_config_file_conflict(config=cur, answer=answer)

    def upgrade(self):
        '''runs upgrade'''
        try:
            self.trans2.connect('progress-changed', self.upgrade_progress)
            self.trans2.connect('cancellable-changed',
                                self.upgrade_cancellable_changed)
            self.trans2.connect('progress-details-changed',
                                self.upgrade_progress_detail)
            self.trans2.connect('progress-download-changed',
                                self.upgrade_progress_download)
            self.trans2.connect('finished', self.upgrade_finish)
            self.trans2.connect('error', self.upgrade_error)
            self.trans2.connect("status-details-changed",
                                self.status_details_changed)
            self.trans2.connect("status-changed", self.status_changed)
            self.trans2.connect("config-file-conflict",
                                self.config_file_conflict)

            # TODO make a terminal work to see more info
            # self.trans2.set_terminal(os.ttyname(self.slave))

            '''
            # TODO implement this
            self.trans2.connect("medium-required", self._on_medium_required)
            remove_obsoleted_depends
            '''
            self.trans2.set_debconf_frontend('kde')
            self.trans2.run()

        except (NotAuthorizedError, TransactionFailed) as e:
            print("Warning: install transaction not completed successfully:"
                  + "{}".format(e))

    def call_reject(self):
        '''when close button is pushed, quit'''
        app.quit()

    def repair_install(self):
        self.closeBtn.setVisible(False)
        try:
            self.trans3.connect('progress-changed', self.repair_progress)
            # TODO the next line breaks the ability to see download progress
            # for... some reason. Maybe an aptd bug?
            # self.trans3.connect('status-changed', self.status_changed)
            self.trans3.connect('status-details-changed',
                                self.status_details_changed)
            self.trans3.connect('finished', self.repair_finish)
            self.trans3.connect('error', self.upgrade_error)
            self.trans3.set_debconf_frontend('kde')
            self.trans3.run()

        except (NotAuthorizedError, TransactionFailed) as e:
            print("Warning: install transaction not completed successfully:"
                  + "{}".format(e))

    def repair_finish(self, transaction, exit_state):
        '''when repair finish'''
        self.label.setText(_("Repair Finished (if repair was needed)"))
        if exit_state == EXIT_FAILED:
            error_string = get_error_string_from_enum(transaction.error.code)
            error_desc = get_error_description_from_enum(
                transaction.error.code)
            self.plainTextEdit.setEnabled(False)
            self.plainTextEdit.moveCursor(QTextCursor.End)
            self.plainTextEdit.insertPlainText(error_string + "\n")
            self.plainTextEdit.insertPlainText(error_desc + "\n")
            self.plainTextEdit.moveCursor(QTextCursor.End)
            self.plainTextEdit.setEnabled(True)

class App(QApplication):
    '''app'''
    def __init__(self, options, *args):
        QApplication.__init__(self, *args)
        self.dialog = DialogUpg(options)
        self.dialog.show()


def main(args, options):
    '''main'''
    global app
    app = App(options, args)
    app.setWindowIcon(QIcon.fromTheme("system-software-update"))

    # Check for root permissions
    if os.geteuid() != 0:
        text = _("Please run this software with administrative rights."
                 "To do so, run this program with lxqt-sudo.")
        title = "Need administrative powers"
        QMessageBox.critical(None, title, text)
        sys.exit()
    else:
        app.exec_()


if __name__ == "__main__":
    localesApp ="lubuntu-update-notifier"
    localesDir ="/usr/share/locale"
    gettext.bindtextdomain(localesApp, localesDir)
    gettext.textdomain(localesApp)
    _ = gettext.gettext

    # check arguments
    parser = ArgumentParser()
    parser.add_argument("-c",
                        "--cache-update",
                        action="store_true",
                        dest="cacheUpdate",
                        help=_("Update Cache Before Upgrade"))
    parser.add_argument("-f",
                        "--full-upgrade",
                        action="store_true",
                        dest="fullUpgrade",
                        help=_("Full upgrade same as dist-upgrade"))
    parser.add_argument('--version',
                        action='version',
                        version='%(prog)s 0.5.1')
    options = parser.parse_args()

    # run it
    main(sys.argv, options)