#!/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 UI_PATH = os.path.dirname(__file__) class prefsDialog(QDialog): def __init__(self, parent=None): QDialog.__init__(self, parent) self.ui = uic.loadUi(os.path.join(UI_PATH, "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(os.path.join(UI_PATH, "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(os.path.join(UI_PATH, "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(os.path.join(UI_PATH, "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(os.path.join(UI_PATH, "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(os.path.join(UI_PATH, "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())