path: root/mageiaSync/mageiasync.py

#!/usr/bin/python3

from PyQt6.QtWidgets import (
    QProgressDialog,
    QMainWindow,
    QDialog,
    QFileDialog,
    QApplication,
    QAbstractItemView,
)
from PyQt6.QtGui import (
    QStandardItemModel,
    QStandardItem,
    QDesktopServices,
    QIcon,
)
from PyQt6.QtCore import QLibraryInfo, QUrl, QItemSelectionModel, QFileInfo
from PyQt6 import QtCore, uic  # , Qt, QThread, QObject, pyqtSignal)
import sys
import glob
import os

try:
    from . import mageiaSyncExt
except:
    import mageiaSyncExt

class prefsDialog(QDialog):
    def __init__(self, parent=None):
        QDialog.__init__(self, parent)
        self.ui = uic.loadUi("mageiaSyncDBprefs.ui", self)
        self.ui.selectDest.clicked.connect(isosSync.selectDestination)


class prefsDialog0(QDialog):
    def __init__(self, parent=None):
        QDialog.__init__(self, parent)
        self.ui = uic.loadUi("mageiaSyncDBprefs0.ui", self)


class renameDialog(QDialog):
    #   Display a dialog box to choose to rename an old collection of ISOs to a new one

    def __init__(self, parent=None):
        QDialog.__init__(self, parent)
        self.ui = uic.loadUi("mageiaSyncDBrename.ui", self)
        self.ui.chooseDir.clicked.connect(isosSync.renameDir)


class aboutDialog(QDialog):
    def __init__(self, parent=None):
        QDialog.__init__(self, parent)
        self.ui = uic.loadUi("mageiaSyncAbout.ui", self)
        self.ui.creditsButton.clicked.connect(isosSync.credits)


class creditsDialog(QDialog):
    def __init__(self, parent=None):
        QDialog.__init__(self, parent)
        self.ui = uic.loadUi("mageiaSyncCredits.ui", self)
        text = ""
        try:
            with open("/usr/share/doc/mageiasync/README.md", "r") as f:
                block = f.readlines()
                for line in block:
                    text += line
        except:
            pass
        self.ui.Readme.setText(text)


class LogWindow(QProgressDialog):
    #   Display a box at start during the remote directory list loading

    def __init__(self, parent=None):
        super(LogWindow, self).__init__(parent)
        self.setWindowModality(QtCore.Qt.WindowModality.WindowModal)
        self.setWindowTitle("Loading...")
        self.setLabelText(self.tr("Loading images list from repository."))
        self.setMinimum(0)
        self.setMaximum(0)
        self.setAutoReset(False)
        self.setAutoClose(False)
        self.setMinimumDuration(1)

    def perform(self):
        self.progressDialog.setValue(self.progress)


class IsosViewer(QMainWindow):
    #   Display the main window
    def __init__(self, parent=None):
        super(IsosViewer, self).__init__(parent)
        self.ui = uic.loadUi("mageiaSyncUI.ui", self)
        self.connectActions()
        self.ui.IprogressBar.setMinimum(0)
        self.ui.IprogressBar.setMaximum(100)
        self.ui.IprogressBar.setValue(0)
        self.ui.IprogressBar.setEnabled(False)
        self.ui.selectAllState = True
        self.ui.stop.setEnabled(False)
        self.destination = ""
        self.rsyncThread = mageiaSyncExt.syncThread(
            self
        )  # create a thread to launch rsync
        self.rsyncThread.progressSignal.connect(self.setProgress)
        self.rsyncThread.speedSignal.connect(self.setSpeed)
        self.rsyncThread.sizeSignal.connect(self.setSize)
        self.rsyncThread.remainSignal.connect(self.setRemain)
        self.rsyncThread.endSignal.connect(self.syncEnd)
        self.rsyncThread.lvM.connect(self.lvMessage)
        self.rsyncThread.checkSignal.connect(self.checks)
        self.checkThreads = []  #   A list of thread for each iso

        #      Model for local list view in a table
        self.model = QStandardItemModel(0, 4, self)
        headers = [self.tr("Directory"), self.tr("Name"), self.tr("Size"), "SHA3-512"]
        i = 0
        for label in headers:
            self.model.setHeaderData(i, QtCore.Qt.Orientation.Horizontal, label)
            i += 1

        #      Model for remote list view in a table
        self.modelRemote = QStandardItemModel(0, 4, self)
        headers = [
            self.tr("Directory"),
            self.tr("Name"),
            self.tr("Size"),
            self.tr("Date"),
        ]
        i = 0
        for label in headers:
            self.modelRemote.setHeaderData(i, QtCore.Qt.Orientation.Horizontal, label)
            i += 1

        #       settings for the local list view
        self.ui.localList.setModel(self.model)
        self.ui.localList.setColumnWidth(0, 160)
        self.ui.localList.setColumnWidth(1, 350)
        self.ui.localList.setColumnWidth(2, 120)
        self.ui.localList.horizontalHeader().setStretchLastSection(True)
        self.ui.localList.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
        # settings for local iso names management
        self.localListNames = []

        #       settings for the remote list view
        self.ui.listIsos.setModel(self.modelRemote)
        self.ui.listIsos.setColumnWidth(0, 160)
        self.ui.listIsos.setColumnWidth(1, 340)
        self.ui.listIsos.setColumnWidth(2, 120)
        self.ui.listIsos.horizontalHeader().setStretchLastSection(True)
        self.ui.listIsos.setSelectionMode(QAbstractItemView.SelectionMode.MultiSelection)
        self.listIsosNames = []

    def add(self, path, iso, isoSize, date):
        #   Add an remote ISO in list
        itemPath = QStandardItem(path)
        itemPath.setData(path, 3)  # Add tooltip
        itemIso = QStandardItem(iso)
        itemIso.setData(iso, 3)  # Add tooltip
        if isoSize == 0:
            itemSize = QStandardItem("--")
        else:
            formatedSize = isoSize.replace(",", " ")
            itemSize = QStandardItem(formatedSize)
            itemSize.setData(formatedSize, 3)  # Add tooltip
        itemSize.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignVCenter | QtCore.Qt.AlignmentFlag.AlignHCenter)

        itemDate = QStandardItem(date)
        itemDate.setData(date, 3)  # Add tooltip
        itemDate.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignVCenter | QtCore.Qt.AlignmentFlag.AlignHCenter)
        self.modelRemote.appendRow(
            [
                itemPath,
                itemIso,
                itemSize,
                itemDate,
            ]
        )
        self.listIsosNames.append([path, iso])

    def localAdd(self, path, iso, isoSize):
        #   Add an entry in local ISOs list, with indications about checking
        itemPath = QStandardItem(path)
        itemPath.setData(path, 3)  # Add tooltip
        itemIso = QStandardItem(iso)
        itemIso.setData(iso, 3)  # Add tooltip
        if isoSize == 0:
            itemSize = QStandardItem("--")
        else:
            formatedSize = "{:n}".format(isoSize).replace(",", " ")
            itemSize = QStandardItem(formatedSize)
            itemSize.setData(formatedSize, 3)  # Add tooltip
        itemSize.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignVCenter | QtCore.Qt.AlignmentFlag.AlignHCenter)
        itemCheck1 = QStandardItem("--")
        itemCheck1.setTextAlignment(QtCore.Qt.AlignmentFlag.AlignVCenter | QtCore.Qt.AlignmentFlag.AlignHCenter)
        self.model.appendRow(
            [
                itemPath,
                itemIso,
                itemSize,
                itemCheck1,
            ]
        )
        self.localListNames.append([path, iso])

    def setProgress(self, value):
        #   Update the progress bar
        self.IprogressBar.setValue(value)

    def setSpeed(self, value):
        #   Update the speed field
        self.speedLCD.display(value)

    def setSize(self, size):
        #   Update the size field
        self.Lsize.setText(size + self.tr(" bytes"))

    def setRemain(self, remainTime):
        content = QtCore.QTime.fromString(remainTime, "h:mm:ss")
        self.timeRemaining.setTime(content)

    def manualChecks(self):
        remoteRow = -1
        for isoIndex in self.listIsos.selectionModel().selectedIndexes():
            if remoteRow != isoIndex.row():
                remoteRow = isoIndex.row()
                path = self.modelRemote.data(self.modelRemote.index(remoteRow, 0))
                name = self.modelRemote.data(self.modelRemote.index(remoteRow, 1))
                try:
                    #   Look for ISO in local list
                    item = self.model.findItems(name, QtCore.Qt.MatchFlag.MatchExactly, 1)[0]
                except:
                    #   Remote ISO is not yet in local directory. We add it in localList and create the directory
                    self.localAdd(path, name, 0)
                    basedir = QtCore.QDir(self.destination)
                    basedir.mkdir(path)
                    item = self.model.findItems(name, QtCore.Qt.MatchFlag.MatchExactly, 1)[0]
                row = self.model.indexFromItem(item).row()
                self.checks(row)

    def checks(self, isoIndex):
        #   processes a checking for each iso
        #   launches a thread for each iso
        newThread = mageiaSyncExt.checkThread(self)
        self.checkThreads.append(newThread)
        self.checkThreads[-1].setup(
            self.destination,
            self.model.data(self.model.index(isoIndex, 0)),
            self.model.data(self.model.index(isoIndex, 1)),
            isoIndex,
        )
        self.checkThreads[-1].sha3Signal.connect(self.sha3Check)
        #        self.checkThreads[-1].dateSignal.connect(self.dateCheck)
        self.checkThreads[-1].sizeFinalSignal.connect(self.sizeUpdate)
        self.checkThreads[-1].checkStartSignal.connect(self.checkStart)
        self.checkThreads[-1].start()

    def checkStart(self, isoIndex):
        #   the function indicates that checking is in progress
        #   the hundred contains index of the value to check, the minor value contains the row
        col = (int)(isoIndex / 100)
        row = isoIndex - col * 100
        self.model.setData(
            self.model.index(row, col, QtCore.QModelIndex()), self.tr("Checking")
        )

    def sha3Check(self, check):
        verified = False
        signed = True
        if check >= 128:
            val = self.tr("OK")
            row = check - 128
            if row >= 64:
                verified = True
                row -= 64
            if row >= 32:
                row -= 32
                signed = False
        else:
            val = self.tr("Failed")
            row = check
            if row >= 64:
                verified = True
                row -= 64
            if row >= 32:
                signed = False
                row -= 32
            print(row)
        if not signed:
            self.lvMessage(
                "Signature for %s.sha3 not found"
                % self.model.data(self.model.index(row, 1))
            )
        if verified:
            #   we add an icon for the GPG key
            self.lvMessage("Sha3 signature  OK")
            self.model.setData(
                self.model.index(row, 3, QtCore.QModelIndex()),
                QIcon("preflight-verifier"),
                1,
            )
        self.model.setData(self.model.index(row, 3, QtCore.QModelIndex()), val)

    def sizeUpdate(self, signal, isoSize):
        col = (int)(signal / 100)
        row = signal - col * 100
        self.model.setData(self.model.index(row, col, QtCore.QModelIndex()), isoSize)

    def syncEnd(self, rc):
        if rc == 1:
            self.lvMessage(self.tr("Command rsync not found"))
        elif rc == 2:
            self.lvMessage(self.tr("Error in rsync parameters"))
        elif rc == 3:
            self.lvMessage(self.tr("Unknown error in rsync"))
        self.IprogressBar.setEnabled(False)
        self.syncGo.setEnabled(True)
        self.listIsos.setEnabled(True)
        self.selectAll.setEnabled(True)
        self.stop.setEnabled(False)

    def prefsInit(self):
        #   Load the parameters at first
        params = QtCore.QSettings("Mageia", "mageiaSync")
        paramRelease = ""
        try:
            paramRelease = params.value("release", type="QString")
            #   the parameters already initialised?
        except:
            pass
        if paramRelease == "":
            # Values are not yet set
            self.pd0 = prefsDialog0()
            self.pd0.user.setFocus()
            answer = self.pd0.exec()
            if answer:
                #   Update params
                self.user = self.pd0.user.text()
                self.password = self.pd0.password.text()
                self.location = self.pd0.location.text()
                params = QtCore.QSettings("Mageia", "mageiaSync")
                params.setValue("user", self.user)
                params.setValue("password", self.password)
                params.setValue("location", self.location)
            else:
                pass
            #                answer=QDialogButtonBox(QDialogButtonBox.Ok)
            # the user must set values or default values
            self.pd0.close()
            self.pd = prefsDialog()
            if self.password != "":
                code, list = mageiaSyncExt.findRelease(
                    "rsync://" + self.user + "@bcd.mageia.org/isos/", self.password
                )
                if code == 0:
                    for item in list:
                        self.pd.release.addItem(item)
            self.pd.password.setText(self.password)
            self.pd.user.setText(self.user)
            self.pd.location.setText(self.location)
            self.pd.selectDest.setText(QtCore.QDir.currentPath())
            self.pd.release.setFocus()
            answer = self.pd.exec()
            if answer:
                #   Update params
                self.user = self.pd.user.text()
                self.password = self.pd.password.text()
                self.location = self.pd.location.text()
                params = QtCore.QSettings("Mageia", "mageiaSync")
                self.release = self.pd.release.currentText()
                self.destination = self.pd.selectDest.text()
                self.bwl = self.pd.bwl.value()
                params.setValue("release", self.release)
                params.setValue("user", self.user)
                params.setValue("password", self.password)
                params.setValue("location", self.location)
                params.setValue("destination", self.destination)
                params.setValue("bwl", str(self.bwl))
            else:
                pass
                print(self.tr("the user must set values or default values"))
            self.pd.close()
        else:
            self.release = params.value("release", type="QString")
            self.user = params.value("user", type="QString")
            self.location = params.value("location", type="QString")
            self.password = params.value("password", type="QString")
            self.destination = params.value("destination", type="QString")
            self.bwl = params.value("bwl", type=int)
        dest = QFileInfo(self.destination)
        if dest.exists() and dest.isDir():
            self.ui.localDirLabel.setText(self.tr("Local directory: ") + self.destination)
        else:
            # ; {} is the placeholder the directory anme
            self.ui.localDirLabel.setText(
                "/!\\ "
                + self.tr(
                    "Local directory {} doesn't exists or isn't accessible. Check mounts or settings."
                ).format(self.destination)
            )
        if self.location != "":
            self.remoteDirLabel.setText(self.tr("Remote directory: ") + self.location)

    def selectDestination(self):
        #   dialog box to select the destination (local directory)
        directory = QFileDialog.getExistingDirectory(
            self, self.tr("Select a directory"), self.destination
        )
        if directory != "":
            self.pd.selectDest.setText(directory)

    def selectAllIsos(self):
        #   Select or unselect the ISOs in remote list
        if self.selectAllState:
            selectMode = QItemSelectionModel.SelectionFlag.Select
            self.selectAll.setText(self.tr("Unselect &All"))
        else:
            selectMode = QItemSelectionModel.SelectionFlag.Deselect
            self.selectAll.setText(self.tr("Select &All"))
        for i in range(self.modelRemote.rowCount()):
            self.listIsos.selectionModel().select(
                self.modelRemote.index(i, 0), QItemSelectionModel.SelectionFlag.Rows | selectMode
            )  # set flag to selected or deselected
        self.selectAllState = not self.selectAllState

    def connectActions(self):
        self.ui.actionQuit.triggered.connect(self._close)
        self.ui.quit.clicked.connect(self._close)
        self.ui.actionRename.triggered.connect(self.rename)
        self.ui.actionUpdate.triggered.connect(self.updateList)
        self.ui.actionCheck.triggered.connect(self.manualChecks)
        self.ui.actionPreferences.triggered.connect(self.prefs)
        self.ui.syncGo.clicked.connect(self.launchSync)
        self.ui.selectAll.clicked.connect(self.selectAllIsos)
        self.ui.actionAbout.triggered.connect(self.about)
        self.ui.actionOnline_help.triggered.connect(self.help)

    def updateList(self):
        # From the menu entry
        self.lw = LogWindow()
        self.lw.show()
        self.modelRemote.removeRows(0, self.modelRemote.rowCount())
        self.model.removeRows(0, self.model.rowCount())
        if self.location == "":
            self.nameWithPath = (
                "rsync://" + self.user + "@bcd.mageia.org/isos/" + self.release + "/"
            )
        #            print self.nameWithPath
        else:
            self.nameWithPath = self.location + "/"
        self.lvMessage(self.tr("Source: ") + self.nameWithPath)
        self.fillList = mageiaSyncExt.findIsos()
        self.fillList.setup(self.nameWithPath, self.password, self.destination)
        self.fillList.endSignal.connect(self.closeFill)
        self.fillList.start()
        # Reset the button
        self.ui.selectAll.setText(self.tr("Select &All"))
        self.selectAllState = True

    def lvMessage(self, message):
        #   Add a line in the logview
        self.ui.lvText.append(message)

    def renameDir(self):
        #   Choose the directory where isos are stored
        directory = QFileDialog.getExistingDirectory(
            self, self.tr("Select a directory"), self.destination
        )
        self.rd.chooseDir.setText(directory)

    def rename(self):
        #   rename old isos and directories to a new release
        self.rd = renameDialog()
        prefix = os.path.commonprefix([x for x in glob.glob("*", root_dir=self.destination) if os.path.isdir(os.path.join(self.destination,x))])
        self.rd.oldRelease.setText(prefix)
        self.rd.newRelease.setText(prefix)
        self.rd.chooseDir.setText(self.destination)
        answer = self.rd.exec()
        if answer:
            nbf, nbr = mageiaSyncExt.rename(
                self.rd.chooseDir.text(),
                self.rd.oldRelease.text(),
                str(self.rd.newRelease.text()),
                0,
                0,
            )
            returnMsg = (self.tr("Renaming {0} files and {1} directories")).format(
                nbf, nbr
            )
            self.lvMessage(returnMsg)
            self.updateList()
        self.rd.close()

    def prefs(self):
        # From the menu entry
        self.pd = prefsDialog()
        if self.password != "":
            code, list = mageiaSyncExt.findRelease(
                "rsync://" + self.user + "@bcd.mageia.org/isos/", self.password
            )
            if code == 0:
                for item in list:
                    self.pd.release.addItem(item)
        self.pd.release.setCurrentText(self.release)
        self.pd.password.setText(self.password)
        self.pd.user.setText(self.user)
        self.pd.location.setText(self.location)
        self.pd.selectDest.setText(self.destination)
        self.pd.bwl.setValue(self.bwl)
        params = QtCore.QSettings("Mageia", "mageiaSync")
        answer = self.pd.exec()
        if answer:
            params.setValue("release", self.pd.release.currentText())
            params.setValue("user", self.pd.user.text())
            params.setValue("password", self.pd.password.text())
            params.setValue("location", self.pd.location.text())
            params.setValue("destination", self.pd.selectDest.text())
            params.setValue("bwl", str(self.pd.bwl.value()))
            self.prefsInit()
            self.updateList()
        self.pd.close()

    def about(self):
        ad = aboutDialog()
        answer = ad.exec()
        if answer:
            ad.close()

    def credits(self):
        ad = creditsDialog()
        answer = ad.exec()
        if answer:
            ad.close()

    def help(self):
        # Open page in browser
        l = QDesktopServices.openUrl(
            QUrl("https://wiki.mageia.org/en/ISO_testing_rsync_tools")
        )

    def launchSync(self):
        dest = QFileInfo(self.destination)
        if (not dest.exists()) or (not dest.isDir()):
            self.lvMessage(
                "/!\\ "
                + self.tr(
                    "Local directory {} doesn't exists or isn't accessible. Check mounts or settings."
                ).format(self.destination)
            )
            return
        if not dest.isWritable():
            self.lvMessage("/!\\ " + self.tr("Local directory {} isn't writable"))
            return
        self.IprogressBar.setEnabled(True)
        self.stop.setEnabled(True)
        self.syncGo.setEnabled(False)
        self.listIsos.setEnabled(False)
        self.selectAll.setEnabled(False)
        # Connect the button Stop
        self.stop.clicked.connect(self.stopSync)
        self.rsyncThread.params(self.password, self.bwl)
        remoteRow = -1
        for isoIndex in self.listIsos.selectionModel().selectedIndexes():
            if remoteRow != isoIndex.row():
                remoteRow = isoIndex.row()
                path = self.modelRemote.data(self.modelRemote.index(remoteRow, 0))
                name = self.modelRemote.data(self.modelRemote.index(remoteRow, 1))
                try:
                    #   Look for ISO in local list
                    item = self.model.findItems(name, QtCore.Qt.MatchFlag.MatchExactly, 1)[0]
                except:
                    #   Remote ISO is not yet in local directory. We add it in localList and create the directory
                    self.localAdd(path, name, 0)
                    basedir = QtCore.QDir(self.destination)
                    basedir.mkdir(path)
                    item = self.model.findItems(name, QtCore.Qt.MatchFlag.MatchExactly, 1)[0]
                row = self.model.indexFromItem(item).row()
                if self.location == "":
                    self.nameWithPath = (
                        "rsync://"
                        + self.user
                        + "@bcd.mageia.org/isos/"
                        + self.release
                        + "/"
                        + path
                    )
                else:
                    self.nameWithPath = self.location + path
                if not str(path).endswith("/"):
                    self.nameWithPath += "/"
                self.rsyncThread.setup(
                    self.nameWithPath, self.destination + "/" + path + "/", row
                )
        self.rsyncThread.start()  # start the thread

    def closeFill(self, code):
        if code == 0:  #   list returned
            list = self.fillList.getList()
            for size, date, longName in list:
                path = longName.split("/")
                self.add(path[0], path[-1], size, date)
        elif code == 1:
            self.lvMessage(self.tr("Command rsync not found"))
        elif code == 2:
            self.lvMessage(self.tr("Error in rsync parameters"))
        elif code == 3:
            self.lvMessage(self.tr("Unknown error in rsync"))
        list = self.fillList.getLocal()
        for path, iso, isoSize in list:
            self.localAdd(path, iso, isoSize)
        self.fillList.quit()
        self.lw.hide()

    def stopSync(self):
        self.rsyncThread.stop()
        self.IprogressBar.setEnabled(False)
        self.stop.setEnabled(False)
        self.syncGo.setEnabled(True)
        self.listIsos.setEnabled(True)
        self.selectAll.setEnabled(True)

    def main(self):
        self.show()
        #   Load or look for intitial parameters
        self.prefsInit()
        # look for Isos list and add it to the isoSync list. Update preferences
        self.updateList()

    def _close(self):
        self.rsyncThread.stop()
        self.close()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    locale = QtCore.QLocale.system().name()
    qtTranslator = QtCore.QTranslator()
    if qtTranslator.load(
        "qt_" + locale, QLibraryInfo.path(QLibraryInfo.LibraryPath(10)) # QLibraryInfo.TranslationsPath
    ):
        app.installTranslator(qtTranslator)
    appTranslator = QtCore.QTranslator()
    if appTranslator.load("mageiaSync_" + locale, "/usr/share/mageiasync/translations"):
        app.installTranslator(appTranslator)
    isosSync = IsosViewer()
    isosSync.main()
    sys.exit(app.exec())