initial commit
This commit is contained in:
Tilman Kranz
275
bin/simple-apt-update
Executable file
275
bin/simple-apt-update
Executable file
@ -0,0 +1,275 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import subprocess
|
||||
import selectors
|
||||
import html
|
||||
import gi
|
||||
import apt
|
||||
import apt.progress
|
||||
|
||||
gi.require_version('Gtk', '3.0')
|
||||
gi.require_version('Gdk', '3.0')
|
||||
|
||||
try:
|
||||
from gi.repository import Gdk
|
||||
except ImportError:
|
||||
print("ERROR: Could not import Gdk")
|
||||
|
||||
try:
|
||||
from gi.repository import Gtk
|
||||
except ImportError:
|
||||
print("ERROR: Could not import Gtk")
|
||||
|
||||
|
||||
class AptFetchProgress(apt.progress.base.AcquireProgress):
|
||||
def __init__(self, update_window):
|
||||
super().__init__()
|
||||
self.update_window = update_window
|
||||
|
||||
def done(self, item):
|
||||
super().done(item)
|
||||
self.update_window.append("Done fetching package updates.")
|
||||
|
||||
def fail(self, item):
|
||||
super().fail(item)
|
||||
self.update_window.append(
|
||||
"ERROR: The following item could not be downloaded: %s" % item.uri)
|
||||
|
||||
def fetch(self, item):
|
||||
super().fetch(item)
|
||||
self.update_window.append(
|
||||
"INFO: The following item was downloaded: %s" % item.uri)
|
||||
|
||||
def ims_hit(self, item):
|
||||
super().ims_hit(item)
|
||||
self.update_window.append(
|
||||
"INFO: The following item was found in the cache: %s" % item.uri)
|
||||
|
||||
def media_change(self, str, drive):
|
||||
super().media_change(str, drive)
|
||||
self.update_window.append(
|
||||
"ERROR: Media changes are not supported (str=%s, drive=%s)" %
|
||||
(str, drive))
|
||||
return False
|
||||
|
||||
def pulse(self, owner):
|
||||
super().pulse(owner)
|
||||
return True
|
||||
|
||||
|
||||
class AptInstallProgress(apt.progress.base.InstallProgress):
|
||||
def __init__(self, update_window):
|
||||
super().__init__()
|
||||
self.update_window = update_window
|
||||
self.count = 0
|
||||
|
||||
def conffile(self, current, new):
|
||||
super().conffile(current, new)
|
||||
self.update_window.append_mesg(
|
||||
"WARNING",
|
||||
"The following configuration file was not modified: %s" %
|
||||
current)
|
||||
|
||||
def error(self, pkg, errormsg):
|
||||
super().error(pkg, errormsg)
|
||||
self.update_window.append_mesg(
|
||||
"ERROR",
|
||||
"Installation of package %s failed: %s" % (pkg, errormsg))
|
||||
|
||||
def processing(self, pkg, stage):
|
||||
super().error(pkg, stage)
|
||||
self.update_window.append_mesg(
|
||||
"INFO",
|
||||
"Processing package %s, stage %s" % (pkg, stage))
|
||||
|
||||
def dpkg_status_change(self, pkg, status):
|
||||
super().dpkg_status_change(pkg, status)
|
||||
print("dpkg_status_change: pkg=%s status=%s" % (pkg, status))
|
||||
self.update_window.append_mesg(
|
||||
"INFO",
|
||||
"Package %s is now has status %s" % (pkg, status))
|
||||
|
||||
|
||||
class UpdateWindow(Gtk.ApplicationWindow):
|
||||
def clear(self):
|
||||
self.buffer.set_text("")
|
||||
|
||||
def scroll_to_bottom(self):
|
||||
adj = self.scrolledwindow.get_vadjustment()
|
||||
adj.set_value(adj.get_upper())
|
||||
self.scrolledwindow.set_vadjustment(adj)
|
||||
|
||||
def level_to_color(self, level):
|
||||
if level == "INFO":
|
||||
return "green"
|
||||
elif level == "ERROR":
|
||||
return "red"
|
||||
else:
|
||||
return "grey"
|
||||
|
||||
def prepend_mesg(self, level, text):
|
||||
self.prepend(text)
|
||||
self.prepend_color(level + ": ", self.level_to_color(level))
|
||||
|
||||
def append_mesg(self, level, text):
|
||||
self.append_color(level + ": ", self.level_to_color(level))
|
||||
self.append(text)
|
||||
|
||||
def prepend_markup(self, markup):
|
||||
self.insert_markup(markup, self.buffer.get_start_iter())
|
||||
|
||||
def append_markup(self, markup):
|
||||
self.insert_markup(markup, self.buffer.get_end_iter())
|
||||
|
||||
def insert_markup(self, markup, iter):
|
||||
self.buffer.insert_markup(iter, markup, -1)
|
||||
|
||||
def prepend_color(self, text, color):
|
||||
self.insert_color(text, color, self.buffer.get_start_iter())
|
||||
|
||||
def append_color(self, text, color):
|
||||
self.insert_color(text, color, self.buffer.get_end_iter())
|
||||
|
||||
def insert_color(self, text, color, iter):
|
||||
self.buffer.insert_markup(
|
||||
iter,
|
||||
"<span color=\"%s\">%s</span>" % (color, html.escape(text)),
|
||||
-1)
|
||||
|
||||
def prepend(self, text):
|
||||
self.insert(text, self.buffer.get_start_iter())
|
||||
|
||||
def append(self, text):
|
||||
self.insert(text, self.buffer.get_end_iter())
|
||||
self.scroll_to_bottom()
|
||||
|
||||
def insert(self, text, iter):
|
||||
self.buffer.insert(iter, text + "\n")
|
||||
|
||||
def run(self, args, ignore_stderr=False, output_msg=None, empty_msg=None):
|
||||
p = subprocess.Popen(
|
||||
args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False)
|
||||
|
||||
sel = selectors.DefaultSelector()
|
||||
sel.register(p.stdout, selectors.EVENT_READ)
|
||||
sel.register(p.stderr, selectors.EVENT_READ)
|
||||
done = False
|
||||
output = False
|
||||
error = False
|
||||
|
||||
while not done:
|
||||
for key, _ in sel.select():
|
||||
data = key.fileobj.read1().decode().rstrip()
|
||||
|
||||
if data:
|
||||
if key.fileobj is p.stdout:
|
||||
self.append(data)
|
||||
output = True
|
||||
elif not ignore_stderr:
|
||||
self.append(data)
|
||||
|
||||
exit_code = p.poll()
|
||||
|
||||
if exit_code is not None:
|
||||
if exit_code != 0:
|
||||
self.append_mesg(
|
||||
"ERROR",
|
||||
"apt-get exit code: %d\n" % exit_code
|
||||
)
|
||||
error = True
|
||||
|
||||
done = True
|
||||
break
|
||||
|
||||
if not error:
|
||||
if output:
|
||||
if output_msg is not None:
|
||||
self.prepend_mesg("INFO", output_msg)
|
||||
elif empty_msg is not None:
|
||||
self.append_mesg("INFO", empty_msg)
|
||||
|
||||
def on_upgrade(self, *args):
|
||||
self.cache.open(None)
|
||||
self.cache.upgrade(True)
|
||||
self.cache.commit(AptFetchProgress(self), AptInstallProgress(self))
|
||||
|
||||
def on_update(self, *args):
|
||||
self.cache.update()
|
||||
self.append_mesg("INFO", "The package cache was updated.")
|
||||
|
||||
def on_list(self, *args):
|
||||
self.run(
|
||||
['/usr/bin/apt', '-qq', 'list', '--upgradable'],
|
||||
ignore_stderr=True,
|
||||
output_msg="Found the following package upgrades:",
|
||||
empty_msg="Currently there are no available package upgrades.")
|
||||
|
||||
def on_ctrl_w(self, *args):
|
||||
self.app.quit()
|
||||
|
||||
def __init__(self, app):
|
||||
super(UpdateWindow, self).__init__(application=app, title="Simple APT Update")
|
||||
self.app = app
|
||||
self.init_ui()
|
||||
|
||||
def init_ui(self):
|
||||
self.set_icon_from_file('/usr/share/icons/simple-apt-update.svg')
|
||||
self.set_border_width(10)
|
||||
self.set_default_size(630, 390)
|
||||
accel = Gtk.AccelGroup()
|
||||
accel.connect(Gdk.keyval_from_name('W'), Gdk.ModifierType.CONTROL_MASK,
|
||||
0, self.on_ctrl_w)
|
||||
accel.connect(Gdk.keyval_from_name('Q'), Gdk.ModifierType.CONTROL_MASK,
|
||||
0, self.on_ctrl_w)
|
||||
self.add_accel_group(accel)
|
||||
|
||||
hbox = Gtk.Box(spacing=6, orientation=Gtk.Orientation.VERTICAL)
|
||||
self.add(hbox)
|
||||
|
||||
self.update_button = Gtk.Button.new_with_label(
|
||||
"Update the Package Cache")
|
||||
self.update_button.connect("clicked", self.on_update)
|
||||
hbox.pack_start(self.update_button, True, True, 0)
|
||||
|
||||
self.upgrade_button = Gtk.Button.new_with_label(
|
||||
"Apply all Package Upgrades")
|
||||
self.upgrade_button.connect("clicked", self.on_upgrade)
|
||||
hbox.pack_start(self.upgrade_button, True, True, 0)
|
||||
|
||||
self.list_button = Gtk.Button.new_with_label(
|
||||
"List available Package Updates")
|
||||
self.list_button.connect("clicked", self.on_list)
|
||||
hbox.pack_start(self.list_button, True, True, 0)
|
||||
|
||||
self.scrolledwindow = Gtk.ScrolledWindow()
|
||||
self.scrolledwindow.set_hexpand(True)
|
||||
self.scrolledwindow.set_vexpand(True)
|
||||
self.scrolledwindow.set_min_content_height(300)
|
||||
self.scrolledwindow.set_max_content_height(300)
|
||||
self.buffer = Gtk.TextBuffer()
|
||||
|
||||
text_view = Gtk.TextView(buffer=self.buffer)
|
||||
text_view.set_editable(False)
|
||||
text_view.set_cursor_visible(False)
|
||||
|
||||
self.scrolledwindow.add(text_view)
|
||||
hbox.pack_start(self.scrolledwindow, True, True, 0)
|
||||
|
||||
self.cache = apt.Cache()
|
||||
self.cache.update()
|
||||
self.append_mesg("INFO", "The package cache was updated.")
|
||||
|
||||
self.on_update()
|
||||
self.clear()
|
||||
self.on_list()
|
||||
|
||||
|
||||
def on_activate(app):
|
||||
win = UpdateWindow(app)
|
||||
win.present()
|
||||
win.show_all()
|
||||
|
||||
|
||||
app = Gtk.Application(application_id='org.linuxfoo.SimpleAptUpdate')
|
||||
app.connect('activate', on_activate)
|
||||
app.run(None)
|
||||
Reference in New Issue
Block a user