447 lines
16 KiB
Python
447 lines
16 KiB
Python
import asyncio
|
|
import collections
|
|
from datetime import datetime
|
|
import itertools
|
|
import json
|
|
import os
|
|
import typing
|
|
|
|
import PyQt5 # pylint: disable=import-error
|
|
import PyQt5.QtWidgets # pylint: disable=import-error
|
|
|
|
from . import ui_qvmtemplate # pylint: disable=no-name-in-module
|
|
from . import ui_templateinstallconfirmdlg # pylint: disable=no-name-in-module
|
|
from . import ui_templateinstallprogressdlg # pylint: disable=no-name-in-module
|
|
from . import utils
|
|
|
|
#pylint: disable=invalid-name
|
|
|
|
BASE_CMD = ['qvm-template', '--enablerepo=*', '--yes']
|
|
|
|
# singleton for "no date"
|
|
ZERO_DATE = datetime.utcfromtimestamp(0)
|
|
|
|
# pylint: disable=too-few-public-methods,inherit-non-class
|
|
class Template(typing.NamedTuple):
|
|
status: str
|
|
name: str
|
|
evr: str
|
|
reponame: str
|
|
size: int
|
|
buildtime: datetime
|
|
installtime: typing.Optional[datetime]
|
|
#licence: str
|
|
#url: str
|
|
#summary: str
|
|
# ---- internal ----
|
|
description: str
|
|
default_status: str
|
|
# ------------------
|
|
|
|
COL_NAMES = [
|
|
'Status',
|
|
'Name',
|
|
'Version',
|
|
'Repository',
|
|
'Download Size (MB)',
|
|
'Build',
|
|
'Install',
|
|
#'License',
|
|
#'URL',
|
|
#'Summary'
|
|
]
|
|
|
|
@staticmethod
|
|
def build(status, entry):
|
|
cli_format = '%Y-%m-%d %H:%M:%S'
|
|
buildtime = datetime.strptime(entry['buildtime'], cli_format)
|
|
if entry['installtime']:
|
|
installtime = datetime.strptime(entry['installtime'], cli_format)
|
|
else:
|
|
installtime = ZERO_DATE
|
|
return Template(
|
|
status,
|
|
entry['name'],
|
|
'%s:%s-%s' % (entry['epoch'], entry['version'], entry['release']),
|
|
entry['reponame'],
|
|
int(entry['size']) // 1000000,
|
|
buildtime,
|
|
installtime,
|
|
#entry['license'],
|
|
#entry['url'],
|
|
entry['description'],
|
|
status
|
|
)
|
|
|
|
class Action(typing.NamedTuple):
|
|
op: str
|
|
name: str
|
|
evr: str
|
|
|
|
TYPES = [str, str, str]
|
|
COL_NAMES = ['Operation', 'Name', 'Version']
|
|
|
|
class TemplateStatusDelegate(PyQt5.QtWidgets.QStyledItemDelegate):
|
|
OPS = [
|
|
['Installed', 'Reinstall', 'Remove'],
|
|
['Extra', 'Remove'],
|
|
['Upgradable', 'Upgrade', 'Remove'],
|
|
['Downgradable', 'Downgrade', 'Remove'],
|
|
['Available', 'Install']
|
|
]
|
|
|
|
def createEditor(self, parent, option, index):
|
|
_ = option # unused
|
|
editor = PyQt5.QtWidgets.QComboBox(parent)
|
|
# Otherwise the internalPointer can be overwritten with a QComboBox
|
|
index = index.model().index(index.row(), index.column())
|
|
kind = index.internalPointer().default_status
|
|
for op_list in TemplateStatusDelegate.OPS:
|
|
if op_list[0] == kind:
|
|
for op in op_list:
|
|
editor.addItem(op)
|
|
editor.currentIndexChanged.connect(self.currentIndexChanged)
|
|
editor.showPopup()
|
|
return editor
|
|
return None
|
|
|
|
def setEditorData(self, editor, index):
|
|
#pylint: disable=no-self-use
|
|
cur = index.data()
|
|
idx = editor.findText(cur)
|
|
if idx >= 0:
|
|
editor.setCurrentIndex(idx)
|
|
|
|
def setModelData(self, editor, model, index):
|
|
#pylint: disable=no-self-use
|
|
model.setData(index, editor.currentText())
|
|
|
|
def updateEditorGeometry(self, editor, option, index):
|
|
#pylint: disable=no-self-use
|
|
_ = index # unused
|
|
editor.setGeometry(option.rect)
|
|
|
|
@PyQt5.QtCore.pyqtSlot()
|
|
def currentIndexChanged(self):
|
|
self.commitData.emit(self.sender())
|
|
|
|
class TemplateModel(PyQt5.QtCore.QAbstractItemModel):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
self.children = []
|
|
|
|
def flags(self, index):
|
|
if index.isValid() and index.column() == 0:
|
|
return super().flags(index) | PyQt5.QtCore.Qt.ItemIsEditable
|
|
return super().flags(index)
|
|
|
|
def sort(self, idx, order):
|
|
rev = (order == PyQt5.QtCore.Qt.AscendingOrder)
|
|
self.children.sort(key=lambda x: x[idx], reverse=rev)
|
|
|
|
self.dataChanged.emit(*self.row_index(0, self.rowCount() - 1))
|
|
|
|
def index(self, row, column, parent=PyQt5.QtCore.QModelIndex()):
|
|
if not self.hasIndex(row, column, parent):
|
|
return PyQt5.QtCore.QModelIndex()
|
|
|
|
return self.createIndex(row, column, self.children[row])
|
|
|
|
def parent(self, child):
|
|
#pylint: disable=no-self-use
|
|
_ = child # unused
|
|
return PyQt5.QtCore.QModelIndex()
|
|
|
|
def rowCount(self, parent=PyQt5.QtCore.QModelIndex()):
|
|
#pylint: disable=no-self-use
|
|
_ = parent # unused
|
|
return len(self.children)
|
|
|
|
def columnCount(self, parent=PyQt5.QtCore.QModelIndex()):
|
|
#pylint: disable=no-self-use
|
|
_ = parent # unused
|
|
return len(Template.COL_NAMES)
|
|
|
|
def hasChildren(self, index=PyQt5.QtCore.QModelIndex()):
|
|
#pylint: disable=no-self-use
|
|
return index == PyQt5.QtCore.QModelIndex()
|
|
|
|
def data(self, index, role=PyQt5.QtCore.Qt.DisplayRole):
|
|
if index.isValid():
|
|
data = self.children[index.row()][index.column()]
|
|
if role == PyQt5.QtCore.Qt.DisplayRole:
|
|
if data is ZERO_DATE:
|
|
return ''
|
|
if isinstance(data, datetime):
|
|
return data.strftime('%d %b %Y')
|
|
return data
|
|
if role == PyQt5.QtCore.Qt.FontRole:
|
|
font = PyQt5.QtGui.QFont()
|
|
tpl = self.children[index.row()]
|
|
font.setBold(tpl.status != tpl.default_status)
|
|
return font
|
|
if role == PyQt5.QtCore.Qt.TextAlignmentRole:
|
|
if isinstance(data, int):
|
|
return PyQt5.QtCore.Qt.AlignRight
|
|
return PyQt5.QtCore.Qt.AlignLeft
|
|
return None
|
|
|
|
def setData(self, index, value, role=PyQt5.QtCore.Qt.EditRole):
|
|
if index.isValid() and role == PyQt5.QtCore.Qt.EditRole:
|
|
old_list = list(self.children[index.row()])
|
|
old_list[index.column()] = value
|
|
new_tpl = Template(*old_list)
|
|
self.children[index.row()] = new_tpl
|
|
self.dataChanged.emit(index, index)
|
|
return True
|
|
return False
|
|
|
|
def headerData(self, section, orientation,
|
|
role=PyQt5.QtCore.Qt.DisplayRole):
|
|
#pylint: disable=no-self-use
|
|
if section < len(Template.COL_NAMES) \
|
|
and orientation == PyQt5.QtCore.Qt.Horizontal \
|
|
and role == PyQt5.QtCore.Qt.DisplayRole:
|
|
return Template.COL_NAMES[section]
|
|
return None
|
|
|
|
def removeRows(self, row, count, parent=PyQt5.QtCore.QModelIndex()):
|
|
_ = parent # unused
|
|
self.beginRemoveRows(PyQt5.QtCore.QModelIndex(), row, row + count)
|
|
del self.children[row:row+count]
|
|
self.endRemoveRows()
|
|
self.dataChanged.emit(*self.row_index(row, row + count))
|
|
|
|
def row_index(self, low, high):
|
|
return self.createIndex(low, 0), \
|
|
self.createIndex(high, self.columnCount())
|
|
|
|
def set_templates(self, templates):
|
|
self.removeRows(0, self.rowCount())
|
|
cnt = sum(len(g) for _, g in templates.items())
|
|
self.beginInsertRows(PyQt5.QtCore.QModelIndex(), 0, cnt - 1)
|
|
for status, grp in templates.items():
|
|
for tpl in grp:
|
|
self.children.append(Template.build(status, tpl))
|
|
self.endInsertRows()
|
|
self.dataChanged.emit(*self.row_index(0, self.rowCount() - 1))
|
|
|
|
def get_actions(self):
|
|
actions = []
|
|
for tpl in self.children:
|
|
if tpl.status != tpl.default_status:
|
|
actions.append(Action(tpl.status, tpl.name, tpl.evr))
|
|
return actions
|
|
|
|
async def refresh(self, refresh=True):
|
|
cmd = BASE_CMD[:]
|
|
if refresh:
|
|
# Force refresh if triggered by button press
|
|
cmd.append('--refresh')
|
|
cmd.extend(['info', '--machine-readable-json', '--installed',
|
|
'--available', '--upgrades', '--extras'])
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE)
|
|
output, stderr = await proc.communicate()
|
|
output = output.decode('ASCII')
|
|
if proc.returncode != 0:
|
|
stderr = stderr.decode('ASCII')
|
|
return False, stderr
|
|
# Default type is dict as we're going to replace the lists with
|
|
# dicts shortly after
|
|
tpls = collections.defaultdict(dict, json.loads(output))
|
|
# Remove duplicates
|
|
# Should this be done in qvm-template?
|
|
# TODO: Merge templates with same name?
|
|
# If so, we may need to have a separate UI to force versions.
|
|
local_names = set(x['name'] for x in tpls['installed'])
|
|
# Convert to dict for easier subtraction
|
|
for key in tpls:
|
|
tpls[key] = {
|
|
(x['name'], x['epoch'], x['version'], x['release']): x
|
|
for x in tpls[key]}
|
|
tpls['installed'] = {
|
|
k: v for k, v in tpls['installed'].items()
|
|
if k not in tpls['extra'] and k not in tpls['upgradable']}
|
|
tpls['available'] = {
|
|
k: v for k, v in tpls['available'].items()
|
|
if k not in tpls['installed']
|
|
and k not in tpls['upgradable']}
|
|
# If the package name is installed but the specific version is
|
|
# neither installed or an upgrade, then it must be a downgrade
|
|
tpls['downgradable'] = {
|
|
k: v for k, v in tpls['available'].items()
|
|
if k[0] in local_names}
|
|
tpls['available'] = {
|
|
k: v for k, v in tpls['available'].items()
|
|
if k not in tpls['downgradable']}
|
|
# Convert back to list
|
|
tpls = {k.title(): list(v.values()) for k, v in tpls.items()}
|
|
self.set_templates(tpls)
|
|
return True, None
|
|
|
|
class TemplateInstallConfirmDialog(
|
|
ui_templateinstallconfirmdlg.Ui_TemplateInstallConfirmDlg,
|
|
PyQt5.QtWidgets.QDialog):
|
|
def __init__(self, actions):
|
|
super().__init__()
|
|
self.setupUi(self)
|
|
|
|
model = PyQt5.QtGui.QStandardItemModel()
|
|
model.setHorizontalHeaderLabels(Action.COL_NAMES)
|
|
self.treeView.setModel(model)
|
|
|
|
for act in actions:
|
|
model.appendRow([PyQt5.QtGui.QStandardItem(x) for x in act])
|
|
|
|
class TemplateInstallProgressDialog(
|
|
ui_templateinstallprogressdlg.Ui_TemplateInstallProgressDlg,
|
|
PyQt5.QtWidgets.QDialog):
|
|
def __init__(self, actions):
|
|
super().__init__()
|
|
self.setupUi(self)
|
|
self.actions = actions
|
|
self.buttonBox.hide()
|
|
|
|
@staticmethod
|
|
def _process_cr(text):
|
|
"""Reduce lines replaced using CR character (\r)"""
|
|
while '\r' in text:
|
|
prefix, suffix = text.rsplit('\r', 1)
|
|
if '\n' in prefix:
|
|
prefix = prefix.rsplit('\n', 1)[0]
|
|
prefix += '\n'
|
|
else:
|
|
prefix = ''
|
|
text = prefix + suffix
|
|
return text
|
|
|
|
def install(self):
|
|
async def coro():
|
|
self.actions.sort()
|
|
for oper, grp in itertools.groupby(self.actions, lambda x: x[0]):
|
|
oper = oper.lower()
|
|
# No need to specify versions for local operations
|
|
if oper in ('remove', 'purge'):
|
|
specs = [x.name for x in grp]
|
|
else:
|
|
specs = [x.name + '-' + x.evr for x in grp]
|
|
# FIXME: (C)Python versions before 3.9 fully-buffers stderr in
|
|
# this context, cf. https://bugs.python.org/issue13601
|
|
# Forcing it to be unbuffered for the time being so that
|
|
# the messages can be displayed in time.
|
|
envs = os.environ.copy()
|
|
envs['PYTHONUNBUFFERED'] = '1'
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*(BASE_CMD + [oper, '--'] + specs),
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.STDOUT,
|
|
env=envs)
|
|
#pylint: disable=cell-var-from-loop
|
|
status_text = ''
|
|
while True:
|
|
line = await proc.stdout.read(100)
|
|
if line == b'':
|
|
break
|
|
line = line.decode('UTF-8')
|
|
status_text = self._process_cr(status_text + line)
|
|
self.textEdit.setPlainText(status_text)
|
|
if await proc.wait() != 0:
|
|
self.buttonBox.show()
|
|
self.progressBar.setMaximum(100)
|
|
self.progressBar.setValue(0)
|
|
return False
|
|
self.progressBar.setMaximum(100)
|
|
self.progressBar.setValue(100)
|
|
self.buttonBox.show()
|
|
return True
|
|
asyncio.create_task(coro())
|
|
|
|
class QvmTemplateWindow(
|
|
ui_qvmtemplate.Ui_QubesTemplateManager,
|
|
PyQt5.QtWidgets.QMainWindow):
|
|
def __init__(self, qt_app, qubes_app, dispatcher, parent=None):
|
|
_ = parent # unused
|
|
|
|
super().__init__()
|
|
self.setupUi(self)
|
|
self.listing.header().setSectionResizeMode(
|
|
PyQt5.QtWidgets.QHeaderView.ResizeToContents)
|
|
|
|
self.qubes_app = qubes_app
|
|
self.qt_app = qt_app
|
|
self.dispatcher = dispatcher
|
|
|
|
self.listing_model = TemplateModel()
|
|
self.listing_delegate = TemplateStatusDelegate(self.listing)
|
|
|
|
self.listing.setModel(self.listing_model)
|
|
self.listing.setItemDelegateForColumn(0, self.listing_delegate)
|
|
|
|
self.refresh(False)
|
|
self.listing.setItemDelegateForColumn(0, self.listing_delegate)
|
|
self.listing.selectionModel() \
|
|
.selectionChanged.connect(self.update_info)
|
|
|
|
self.actionRefresh.triggered.connect(lambda: self.refresh(True))
|
|
self.actionInstall.triggered.connect(self.do_install)
|
|
|
|
def update_info(self, selected):
|
|
_ = selected # unused
|
|
indices = [
|
|
x
|
|
for x in self.listing.selectionModel().selectedIndexes()
|
|
if x.column() == 0]
|
|
if len(indices) == 0:
|
|
return
|
|
self.infobox.clear()
|
|
cursor = PyQt5.QtGui.QTextCursor(self.infobox.document())
|
|
bold_fmt = PyQt5.QtGui.QTextCharFormat()
|
|
bold_fmt.setFontWeight(PyQt5.QtGui.QFont.Bold)
|
|
norm_fmt = PyQt5.QtGui.QTextCharFormat()
|
|
if len(indices) > 1:
|
|
cursor.insertText('Selected templates:\n', bold_fmt)
|
|
for idx in indices:
|
|
tpl = self.listing_model.children[idx.row()]
|
|
cursor.insertText(tpl.name + '-' + tpl.evr + '\n', norm_fmt)
|
|
else:
|
|
idx = indices[0]
|
|
tpl = self.listing_model.children[idx.row()]
|
|
cursor.insertText('Name: ', bold_fmt)
|
|
cursor.insertText(tpl.name + '\n', norm_fmt)
|
|
cursor.insertText('Description:\n', bold_fmt)
|
|
cursor.insertText(tpl.description + '\n', norm_fmt)
|
|
|
|
def refresh(self, refresh=True):
|
|
self.progressBar.show()
|
|
async def coro():
|
|
ok, stderr = await self.listing_model.refresh(refresh)
|
|
self.infobox.clear()
|
|
if not ok:
|
|
cursor = PyQt5.QtGui.QTextCursor(self.infobox.document())
|
|
fmt = PyQt5.QtGui.QTextCharFormat()
|
|
fmt.setFontWeight(PyQt5.QtGui.QFont.Bold)
|
|
cursor.insertText('Failed to fetch template list:\n', fmt)
|
|
fmt.setFontWeight(PyQt5.QtGui.QFont.Normal)
|
|
cursor.insertText(stderr, fmt)
|
|
self.progressBar.hide()
|
|
asyncio.create_task(coro())
|
|
|
|
def do_install(self):
|
|
actions = self.listing_model.get_actions()
|
|
confirm = TemplateInstallConfirmDialog(actions)
|
|
if confirm.exec_():
|
|
progress = TemplateInstallProgressDialog(actions)
|
|
progress.install()
|
|
progress.exec_()
|
|
self.refresh()
|
|
|
|
def main():
|
|
utils.run_asynchronous(QvmTemplateWindow)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|