Skip to content

X
ximper-system-updater
Commit 1ca753a1 authored by Roman Alifanov's avatar Roman Alifanov
Browse files
Options

converted to Gio

parent 8e767d91
Hide whitespace changes
Inline Side-by-side
Showing with 272 additions and 227 deletions
src/apm_check_updates.py
View file @ 1ca753a1
#!/usr/bin/env python3 #!/usr/bin/env python3
import dbus, json import json
from dbus.mainloop.glib import DBusGMainLoop import gi
from gi.repository import GLib gi.require_version('Gio', '2.0')
gi.require_version('GLib', '2.0')
def handle_signal(*args, **kwargs): from gi.repository import Gio, GLib
response = json.loads(args[0])
print(
f"signal handled:\n{response["message"]}\n{response["state"]}\n{response["progress"]}"
)
def handle_signal(connection, sender_name, object_path, interface_name, signal_name, parameters, *args, **kwargs):
try:
response = json.loads(parameters.unpack()[0])
print(f"signal handled:\n{response['message']}\n{response['state']}\n{response['progress']}")
except Exception as e:
print("Failed to handle signal:", e)
def handle_updates(*args, **kwargs): def handle_updates(source_object, result, user_data=None):
response = json.loads(args[0]) try:
variant = bus.call_finish(result)
print( response_str = variant.unpack()[0]
"handled method response:\n", response = json.loads(response_str)
response["data"]["info"],
) print("handled method response:")
print(response.get('data', {}).get('info', 'No info'))
if response["data"]["info"]["upgradedPackages"]: if 'data' in response and 'upgradedPackages' in response['data'].get('info', {}):
print(f"{response["data"]["message"]}\n{response["data"]["info"]["upgradedPackages"]}") upgraded = response['data']['info']['upgradedPackages']
print(f"\n{response.get('message', '')}\n{upgraded}")
except Exception as e:
print("Error during async method call:", e)
if __name__ == "__main__": if __name__ == "__main__":
# Initialize the D-Bus main loop loop = GLib.MainLoop()
DBusGMainLoop(set_as_default=True)
bus = dbus.SystemBus()
bus.add_signal_receiver(
handle_signal,
dbus_interface="org.altlinux.APM",
interface_keyword="interface",
member_keyword="member",
path_keyword="path",
signal_name="Notification"
)
print("try to get updates...") bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
bus.call_async(
reply_handler=handle_updates, bus.call(
error_handler=print,
bus_name="org.altlinux.APM", bus_name="org.altlinux.APM",
dbus_interface="org.altlinux.APM.system",
object_path="/org/altlinux/APM", object_path="/org/altlinux/APM",
method="CheckUpgrade", interface_name="org.altlinux.APM.system",
signature="s", method_name="CheckUpgrade",
args=["Ximper System Updater"], parameters=GLib.Variant("(s)", ("Ximper System Updater",)),
flags=Gio.DBusCallFlags.NONE,
reply_type=GLib.VariantType("(s)"),
timeout_msec=-1,
callback=handle_updates,
)
Gio.DBusConnection.signal_subscribe(
bus,
sender="org.altlinux.APM",
interface_name="org.altlinux.APM",
member="Notification",
object_path="/org/altlinux/APM",
arg0=None,
flags=Gio.DBusSignalFlags.NONE,
callback=handle_signal,
) )
print("try to get updates...")
print("Listening for signals from 'org.altlinux.APM'...") print("Listening for signals from 'org.altlinux.APM'...")
loop = GLib.MainLoop()
try: try:
loop.run() loop.run()
except KeyboardInterrupt: except KeyboardInterrupt:
print("\nExiting.")
loop.quit() loop.quit()
print("\nExiting.")
\ No newline at end of file
src/test.py
View file @ 1ca753a1
#!/usr/bin/env python3 #!/usr/bin/env python3
import dbus, json, threading import json
from dbus.mainloop.glib import DBusGMainLoop import gi
from gi.repository import GLib, Gtk, Adw, GObject, Gio gi.require_version('Gtk', '4.0')
gi.require_version('GLib', '2.0')
gi.require_version('Gio', '2.0')
from gi.repository import Gtk, GLib, Gio, GObject
class PackageObject(GObject.Object): class UpdaterItem(GObject.Object):
__gtype_name__ = 'PackageObject' __gtype_name__ = 'UpdaterItem'
name = GObject.Property(type=str, default="") def __init__(self, title, message, progress):
def __init__(self, name):
super().__init__() super().__init__()
self.name = name self.title = title
self.message = message
class PackageRow(Gtk.Box): self.progress = progress
__gtype_name__ = 'PackageRow'
def __init__(self):
super().__init__(
orientation=Gtk.Orientation.HORIZONTAL,
spacing=10,
margin_start=10,
margin_end=10,
margin_top=5,
margin_bottom=5
)
self.icon = Gtk.Image.new_from_icon_name("package-x-generic-symbolic") @GObject.Property(type=str)
self.icon.set_pixel_size(32) def title(self):
self.append(self.icon) return self._title
self.label = Gtk.Label(xalign=0, hexpand=True) @title.setter
self.append(self.label) def title(self, value):
self._title = value
def bind(self, package_obj): @GObject.Property(type=str)
self.label.set_label(package_obj.name) def message(self):
return self._message
class DBusManager: @message.setter
def __init__(self, callback): def message(self, value):
self.callback = callback self._message = value
self.bus = None
self.loop = None
self.thread = threading.Thread(target=self.run, daemon=True)
self.thread.start()
def run(self):
DBusGMainLoop(set_as_default=True)
self.bus = dbus.SystemBus()
self.bus.add_signal_receiver(
self.handle_signal,
dbus_interface="org.altlinux.APM",
signal_name="Notification"
)
self.check_updates() @GObject.Property(type=str)
def progress(self):
return self._progress
self.loop = GLib.MainLoop() @progress.setter
self.loop.run() def progress(self, value):
self._progress = value
def check_updates(self):
try:
proxy = self.bus.get_object("org.altlinux.APM", "/org/altlinux/APM")
system = dbus.Interface(proxy, "org.altlinux.APM.system")
response = system.CheckUpgrade("Ximper System Updater", timeout=60) class DBusManager:
self.handle_updates(response) def __init__(self, app):
except Exception as e: self.app = app
print(f"D-Bus ошибка: {e}") self.connection = None
self.callback("error", error=str(e)) self.signal_subscription_id = 0
self.connect()
def handle_signal(self, *args): def connect(self):
try: try:
response = json.loads(args[0]) self.connection = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
print(f"Сигнал: {response['message']}\nСостояние: {response['state']}\nПрогресс: {response['progress']}")
except Exception as e: # Сохраняем ID подписки
print(f"Ошибка обработки сигнала: {e}") self.signal_subscription_id = self.connection.signal_subscribe(
"org.altlinux.APM",
def handle_updates(self, response): "org.altlinux.APM",
"Notification",
"/org/altlinux/APM",
None,
Gio.DBusSignalFlags.NONE,
self.handle_signal,
None
)
self.app.log_message("Подключено к D-Bus. Ожидание сигналов...")
except GLib.Error as e:
self.app.log_message(f"Ошибка подключения к D-Bus: {e.message}", is_error=True)
def handle_signal(self, connection, sender_name, object_path, interface_name, signal_name, parameters, user_data):
try: try:
response = json.loads(response) response = json.loads(parameters[0])
info = response["data"]["info"]
packages = info.get("upgradedPackages", [])
self.callback("packages", packages=packages)
except Exception as e:
print(f"Ошибка обработки обновлений: {e}")
self.callback("error", error="Ошибка обработки ответа")
class PackageRowFactory(Gtk.SignalListItemFactory):
def __init__(self, item_clicked_callback):
super().__init__()
self.item_clicked_callback = item_clicked_callback
self.connect("setup", self.on_setup)
self.connect("bind", self.on_bind)
def on_setup(self, factory, list_item):
row = PackageRow()
list_item.set_child(row)
click_controller = Gtk.GestureClick()
click_controller.connect("pressed", self.on_item_clicked, list_item)
row.add_controller(click_controller)
def on_bind(self, factory, list_item):
package_obj = list_item.get_item()
row = list_item.get_child()
row.bind(package_obj)
def on_item_clicked(self, controller, n_press, x, y, list_item):
package_obj = list_item.get_item()
if package_obj:
self.item_clicked_callback(package_obj.name)
class MainWindow(Adw.ApplicationWindow):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.set_default_size(600, 400)
self.set_title("Ximper System Updater")
self.toolbar_view = Adw.ToolbarView()
self.set_content(self.toolbar_view)
header_bar = Adw.HeaderBar()
self.refresh_button = Gtk.Button(
icon_name="view-refresh-symbolic",
tooltip_text="Проверить обновления"
)
self.refresh_button.connect("clicked", self.on_refresh_clicked)
header_bar.pack_end(self.refresh_button)
self.toolbar_view.add_top_bar(header_bar)
self.status_page = Adw.StatusPage(
title="Обновления не проверены",
description="Нажмите кнопку обновления для проверки",
icon_name="system-software-update-symbolic"
)
self.package_store = Gio.ListStore.new(PackageObject) # Формируем заголовок на основе типа сигнала
title = "Сигнал"
self.selection_model = Gtk.NoSelection.new(self.package_store) if "state" in response:
self.list_view = Gtk.ListView.new( title = f"Состояние: {response['state']}"
self.selection_model,
PackageRowFactory(self.on_package_clicked)
)
self.scrolled = Gtk.ScrolledWindow() message = response.get("message", "Нет сообщения")
self.scrolled.set_child(self.list_view) progress = response.get("progress", "")
self.scrolled.set_vexpand(True)
self.scrolled.set_hexpand(True)
self.content_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) # Обновляем UI через главный поток
self.content_box.append(self.status_page) GLib.idle_add(self.app.add_update_item, title, message, progress)
self.toolbar_view.set_content(self.content_box) except json.JSONDecodeError as e:
self.app.log_message(f"Ошибка декодирования сигнала: {e}", is_error=True)
except Exception as e:
self.app.log_message(f"Неожиданная ошибка в обработчике сигнала: {e}", is_error=True)
self.dbus_manager = DBusManager(self.handle_dbus_event) def shutdown(self):
if self.connection and self.signal_subscription_id:
self.connection.signal_unsubscribe(self.signal_subscription_id)
def on_refresh_clicked(self, button):
self.show_loading_state()
threading.Thread(target=self.dbus_manager.check_updates, daemon=True).start()
def on_package_clicked(self, package_name): class ListItemWidget(Gtk.Box):
print(f"Клик по пакету: {package_name}") def __init__(self):
super().__init__(orientation=Gtk.Orientation.VERTICAL,
spacing=4,
margin_start=12,
margin_end=12,
margin_top=6,
margin_bottom=6)
self.title_label = Gtk.Label(xalign=0, css_classes=['title'])
self.append(self.title_label)
self.message_label = Gtk.Label(xalign=0, wrap=True, wrap_mode=2)
self.append(self.message_label)
self.progress_label = Gtk.Label(xalign=0, css_classes=['dim-label'])
self.append(self.progress_label)
def update(self, item):
self.title_label.set_label(item.title)
self.message_label.set_label(item.message)
self.progress_label.set_label(item.progress)
class MainWindow(Gtk.ApplicationWindow):
def __init__(self, app):
super().__init__(application=app,
title="Системное обновление",
default_width=600,
default_height=400)
self.app = app
self.setup_ui()
def setup_ui(self):
# Создаем главный контейнер
main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
self.set_child(main_box)
# Создаем заголовочную панель
header = Gtk.HeaderBar()
self.set_titlebar(header)
# Кнопка для проверки обновлений
self.check_button = Gtk.Button(label="Проверить обновления")
self.check_button.connect("clicked", self.on_check_updates)
header.pack_start(self.check_button)
# Создаем список для отображения сигналов
factory = Gtk.SignalListItemFactory()
factory.connect("setup", self.on_factory_setup)
factory.connect("bind", self.on_factory_bind)
self.selection = Gtk.SingleSelection.new(self.app.store)
self.list_view = Gtk.ListView(model=self.selection, factory=factory)
# Добавляем список в прокручиваемую область
scrolled = Gtk.ScrolledWindow()
scrolled.set_child(self.list_view)
scrolled.set_vexpand(True)
main_box.append(scrolled)
# Строка состояния
self.status_label = Gtk.Label()
main_box.append(self.status_label)
def on_factory_setup(self, factory, list_item):
# Создаем виджет для элемента списка
list_item.set_child(ListItemWidget())
def on_factory_bind(self, factory, list_item):
# Обновляем виджет данными
widget = list_item.get_child()
item = list_item.get_item()
widget.update(item)
def on_check_updates(self, button):
# Запрос на обновления только по явному нажатию кнопки
self.app.log_message("Запрос на проверку обновлений...")
if not self.app.dbus_manager.connection:
self.app.log_message("Нет подключения к D-Bus", is_error=True)
return
def handle_dbus_event(self, event_type, **kwargs): self.app.dbus_manager.connection.call(
if event_type == "packages": "org.altlinux.APM",
GLib.idle_add(self.update_package_list, kwargs["packages"]) "/org/altlinux/APM",
elif event_type == "error": "org.altlinux.APM.system",
GLib.idle_add(self.show_error_state, kwargs["error"]) "CheckUpgrade",
GLib.Variant("(s)", ["System Updater"]),
None,
Gio.DBusCallFlags.NONE,
-1,
None,
self.handle_updates,
None
)
def show_loading_state(self): def handle_updates(self, connection, task, user_data):
self.status_page.set_title("Проверка обновлений...") try:
self.status_page.set_description("Идёт запрос к системе обновлений") result = connection.call_finish(task)
self.status_page.set_icon_name("system-software-update-symbolic") response = json.loads(result[0])
self.show_content(self.status_page)
if "data" in response and "message" in response["data"]:
self.app.log_message(f"Ответ: {response['data']['message']}")
else:
self.app.log_message("Получен ответ о проверке обновлений")
except GLib.Error as e:
self.app.log_message(f"Ошибка проверки обновлений: {e.message}", is_error=True)
except json.JSONDecodeError as e:
self.app.log_message(f"Ошибка декодирования ответа: {e}", is_error=True)
except Exception as e:
self.app.log_message(f"Неожиданная ошибка: {e}", is_error=True)
def show_package_list(self): def log_message(self, message, is_error=False):
self.show_content(self.scrolled) if is_error:
self.status_label.add_css_class('error')
else:
self.status_label.remove_css_class('error')
self.status_label.set_label(message)
def show_no_updates(self): def add_update_item(self, title, message, progress):
self.status_page.set_title("Обновления не найдены") item = UpdaterItem(title=title, message=message, progress=progress)
self.status_page.set_description("Ваша система полностью обновлена") self.app.store.append(item)
self.status_page.set_icon_name("emblem-ok-symbolic")
self.show_content(self.status_page)
def show_error_state(self, error): # Автоматическая прокрутка к новому элементу
self.status_page.set_title("Ошибка при проверке обновлений") pos = self.app.store.get_n_items() - 1
self.status_page.set_description(error) self.selection.set_selected(pos)
self.status_page.set_icon_name("dialog-error-symbolic")
self.show_content(self.status_page)
def show_content(self, widget):
for child in self.content_box:
self.content_box.remove(child)
self.content_box.append(widget) class DBusUpdaterApp(Gtk.Application):
def __init__(self):
super().__init__(application_id='org.example.DBusUpdater',
flags=Gio.ApplicationFlags.FLAGS_NONE)
def update_package_list(self, packages): # Модель данных для списка
if not packages: self.store = Gio.ListStore.new(UpdaterItem)
self.show_no_updates() self.dbus_manager = None
return self.window = None
self.show_package_list() def do_activate(self):
if not self.window:
self.window = MainWindow(self)
self.dbus_manager = DBusManager(self)
self.window.present()
self.package_store.remove_all() def do_shutdown(self):
for package in packages: if self.dbus_manager:
self.package_store.append(PackageObject(package)) self.dbus_manager.shutdown()
super().do_shutdown()
self.set_title(f"Ximper System Updater ({len(packages)} обновлений)") def log_message(self, message, is_error=False):
if self.window:
GLib.idle_add(self.window.log_message, message, is_error)
class SystemUpdaterApp(Adw.Application): def add_update_item(self, title, message, progress):
def __init__(self): if self.window:
super().__init__(application_id='com.ximper.SystemUpdater') GLib.idle_add(self.window.add_update_item, title, message, progress)
def do_activate(self):
win = MainWindow(application=self)
win.present()
if __name__ == "__main__": if __name__ == '__main__':
app = SystemUpdaterApp() app = DBusUpdaterApp()
app.run() app.run(None)
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment