381 lines
15 KiB
Python
381 lines
15 KiB
Python
import gi
|
|
gi.require_version("Gtk", "4.0")
|
|
gi.require_version("Adw", "1")
|
|
from gi.repository import Adw, Gio, GLib, Gtk
|
|
|
|
import json
|
|
import socket as sock_module
|
|
import threading
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from config import SOCKET_PATH, STATE_FILE
|
|
|
|
|
|
class UpdaterWindow(Adw.ApplicationWindow):
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.set_title("Mises à jour")
|
|
self.set_default_size(500, -1)
|
|
self.set_resizable(True)
|
|
|
|
self._pending_rows: list[Adw.ActionRow] = []
|
|
self._recent_rows: list[Adw.ActionRow] = []
|
|
|
|
self._build_ui()
|
|
self._setup_file_monitor()
|
|
self._load_state()
|
|
|
|
# ─── Construction de l'interface ────────────────────────────────────────
|
|
|
|
def _build_ui(self):
|
|
self.toast_overlay = Adw.ToastOverlay()
|
|
|
|
toolbar_view = Adw.ToolbarView()
|
|
|
|
header = Adw.HeaderBar()
|
|
self.refresh_btn = Gtk.Button.new_from_icon_name("view-refresh-symbolic")
|
|
self.refresh_btn.set_tooltip_text("Vérifier maintenant")
|
|
self.refresh_btn.connect("clicked", self._on_refresh_clicked)
|
|
header.pack_end(self.refresh_btn)
|
|
toolbar_view.add_top_bar(header)
|
|
|
|
scroll = Gtk.ScrolledWindow()
|
|
scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
|
|
scroll.set_propagate_natural_height(True)
|
|
|
|
main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=20)
|
|
main_box.set_margin_top(20)
|
|
main_box.set_margin_bottom(28)
|
|
main_box.set_margin_start(20)
|
|
main_box.set_margin_end(20)
|
|
|
|
# Carte de statut principal
|
|
self._build_status_card(main_box)
|
|
|
|
# Groupe "En attente du redémarrage"
|
|
self.pending_group = Adw.PreferencesGroup()
|
|
self.pending_group.set_title("En attente du redémarrage")
|
|
self.pending_group.set_description(
|
|
"Ces paquets seront installés automatiquement à la prochaine extinction du PC"
|
|
)
|
|
|
|
self.install_now_btn = Gtk.Button.new_with_label("Installer maintenant")
|
|
self.install_now_btn.add_css_class("suggested-action")
|
|
self.install_now_btn.add_css_class("pill")
|
|
self.install_now_btn.connect("clicked", self._on_install_pending_clicked)
|
|
self.pending_group.set_header_suffix(self.install_now_btn)
|
|
|
|
main_box.append(self.pending_group)
|
|
|
|
# Groupe "Récemment installées"
|
|
self.recent_group = Adw.PreferencesGroup()
|
|
self.recent_group.set_title("Récemment installées")
|
|
main_box.append(self.recent_group)
|
|
|
|
scroll.set_child(main_box)
|
|
toolbar_view.set_content(scroll)
|
|
self.toast_overlay.set_child(toolbar_view)
|
|
self.set_content(self.toast_overlay)
|
|
|
|
def _build_status_card(self, parent: Gtk.Box):
|
|
card = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0)
|
|
card.add_css_class("card")
|
|
|
|
# Côté gauche : icône ou spinner
|
|
left = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
|
left.set_valign(Gtk.Align.CENTER)
|
|
left.set_margin_start(20)
|
|
left.set_margin_end(16)
|
|
left.set_margin_top(20)
|
|
left.set_margin_bottom(20)
|
|
|
|
self.status_icon = Gtk.Image()
|
|
self.status_icon.set_pixel_size(40)
|
|
self.status_icon.set_valign(Gtk.Align.CENTER)
|
|
|
|
self.status_spinner = Gtk.Spinner()
|
|
self.status_spinner.set_size_request(40, 40)
|
|
self.status_spinner.set_valign(Gtk.Align.CENTER)
|
|
self.status_spinner.set_visible(False)
|
|
|
|
left.append(self.status_icon)
|
|
left.append(self.status_spinner)
|
|
|
|
# Côté droit : texte
|
|
text_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
|
|
text_box.set_valign(Gtk.Align.CENTER)
|
|
text_box.set_hexpand(True)
|
|
text_box.set_margin_end(20)
|
|
text_box.set_margin_top(18)
|
|
text_box.set_margin_bottom(18)
|
|
|
|
self.status_title_lbl = Gtk.Label()
|
|
self.status_title_lbl.set_halign(Gtk.Align.START)
|
|
self.status_title_lbl.add_css_class("title-3")
|
|
self.status_title_lbl.set_wrap(True)
|
|
|
|
self.status_sub_lbl = Gtk.Label()
|
|
self.status_sub_lbl.set_halign(Gtk.Align.START)
|
|
self.status_sub_lbl.add_css_class("caption")
|
|
self.status_sub_lbl.add_css_class("dim-label")
|
|
self.status_sub_lbl.set_wrap(True)
|
|
|
|
text_box.append(self.status_title_lbl)
|
|
text_box.append(self.status_sub_lbl)
|
|
|
|
card.append(left)
|
|
card.append(text_box)
|
|
parent.append(card)
|
|
|
|
# ─── Surveillance du fichier d'état ────────────────────────────────────
|
|
|
|
def _setup_file_monitor(self):
|
|
gfile = Gio.File.new_for_path(str(STATE_FILE))
|
|
try:
|
|
self._monitor = gfile.monitor_file(Gio.FileMonitorFlags.NONE, None)
|
|
self._monitor.connect("changed", self._on_file_changed)
|
|
except Exception:
|
|
GLib.timeout_add_seconds(5, self._poll_state)
|
|
|
|
def _on_file_changed(self, _monitor, _file, _other, event):
|
|
if event in (Gio.FileMonitorEvent.CHANGED, Gio.FileMonitorEvent.CREATED):
|
|
GLib.timeout_add(300, self._load_state)
|
|
|
|
def _poll_state(self) -> bool:
|
|
self._load_state()
|
|
return True
|
|
|
|
def _load_state(self) -> bool:
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
state = json.load(f)
|
|
self._update_ui(state)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
self._show_service_unavailable()
|
|
return False # Ne pas répéter si appelé via timeout_add
|
|
|
|
# ─── Mise à jour de l'interface ────────────────────────────────────────
|
|
|
|
def _show_service_unavailable(self):
|
|
self._set_icon("network-offline-symbolic")
|
|
self.status_title_lbl.set_text("Service non disponible")
|
|
self.status_sub_lbl.set_text(
|
|
"Démarrez le service : systemctl start cachyos-updater"
|
|
)
|
|
self.pending_group.set_visible(False)
|
|
self.recent_group.set_visible(False)
|
|
self.refresh_btn.set_sensitive(False)
|
|
|
|
def _set_icon(self, icon_name: str, color_class: str = ""):
|
|
self.status_spinner.stop()
|
|
self.status_spinner.set_visible(False)
|
|
self.status_icon.set_visible(True)
|
|
self.status_icon.set_from_icon_name(icon_name)
|
|
for cls in ("success", "warning", "error", "accent", "dim-label"):
|
|
self.status_icon.remove_css_class(cls)
|
|
if color_class:
|
|
self.status_icon.add_css_class(color_class)
|
|
|
|
def _set_spinner(self):
|
|
self.status_icon.set_visible(False)
|
|
self.status_spinner.set_visible(True)
|
|
self.status_spinner.start()
|
|
|
|
def _update_ui(self, state: dict):
|
|
status = state.get("status", "idle")
|
|
pending = state.get("pending_restart", [])
|
|
history = state.get("installed_history", [])
|
|
last_check = state.get("last_check")
|
|
errors = state.get("errors", [])
|
|
|
|
# ── Carte de statut ──────────────────────────────────────────────
|
|
if status == "checking":
|
|
self._set_spinner()
|
|
self.status_title_lbl.set_text("Vérification des mises à jour…")
|
|
self.status_sub_lbl.set_text("")
|
|
self.refresh_btn.set_sensitive(False)
|
|
|
|
elif status == "downloading":
|
|
self._set_spinner()
|
|
self.status_title_lbl.set_text("Téléchargement des mises à jour…")
|
|
self.status_sub_lbl.set_text("Installation dans quelques instants")
|
|
self.refresh_btn.set_sensitive(False)
|
|
|
|
elif status == "installing":
|
|
self._set_spinner()
|
|
self.status_title_lbl.set_text("Installation en cours…")
|
|
self.status_sub_lbl.set_text("Ne pas éteindre le PC")
|
|
self.refresh_btn.set_sensitive(False)
|
|
|
|
elif status == "error":
|
|
self._set_icon("dialog-error-symbolic", "error")
|
|
self.status_title_lbl.set_text("Erreur lors de la mise à jour")
|
|
self.status_sub_lbl.set_text(
|
|
errors[-1]["message"] if errors else "Erreur inconnue"
|
|
)
|
|
self.refresh_btn.set_sensitive(True)
|
|
|
|
else: # idle
|
|
self.refresh_btn.set_sensitive(True)
|
|
if pending:
|
|
n = len(pending)
|
|
self._set_icon("software-update-available-symbolic", "warning")
|
|
self.status_title_lbl.set_text(
|
|
f"{n} mise{'s' if n > 1 else ''} à jour "
|
|
f"en attente de redémarrage"
|
|
)
|
|
self.status_sub_lbl.set_text(
|
|
"Seront installées automatiquement à la prochaine extinction"
|
|
)
|
|
else:
|
|
self._set_icon("emblem-ok-symbolic", "success")
|
|
self.status_title_lbl.set_text("Système à jour")
|
|
sub = ""
|
|
if last_check:
|
|
try:
|
|
dt = datetime.fromisoformat(last_check)
|
|
sub = f"Dernière vérification : {self._relative_time(dt)}"
|
|
except Exception:
|
|
pass
|
|
self.status_sub_lbl.set_text(sub)
|
|
|
|
# ── Bouton "Installer maintenant" ────────────────────────────────
|
|
can_install = status in ("idle", "error") and bool(pending)
|
|
self.install_now_btn.set_sensitive(can_install)
|
|
|
|
# ── Liste des paquets en attente de redémarrage ──────────────────
|
|
for row in self._pending_rows:
|
|
self.pending_group.remove(row)
|
|
self._pending_rows.clear()
|
|
|
|
for pkg in pending:
|
|
row = Adw.ActionRow()
|
|
row.set_title(GLib.markup_escape_text(pkg.get("name", "")))
|
|
row.set_subtitle(
|
|
f"{pkg.get('old_version', '')} → {pkg.get('new_version', '')}"
|
|
)
|
|
icon = Gtk.Image.new_from_icon_name("system-restart-symbolic")
|
|
icon.add_css_class("warning")
|
|
row.add_prefix(icon)
|
|
self.pending_group.add(row)
|
|
self._pending_rows.append(row)
|
|
|
|
self.pending_group.set_visible(bool(pending))
|
|
|
|
# ── Historique des installations récentes ────────────────────────
|
|
for row in self._recent_rows:
|
|
self.recent_group.remove(row)
|
|
self._recent_rows.clear()
|
|
|
|
recent = list(reversed(history[-30:]))
|
|
for pkg in recent:
|
|
row = Adw.ActionRow()
|
|
row.set_title(GLib.markup_escape_text(pkg.get("name", "")))
|
|
subtitle = (
|
|
f"{pkg.get('old_version', '')} → {pkg.get('new_version', '')}"
|
|
)
|
|
installed_at = pkg.get("installed_at", "")
|
|
if installed_at:
|
|
try:
|
|
dt = datetime.fromisoformat(installed_at)
|
|
subtitle += f" • {self._relative_time(dt)}"
|
|
except Exception:
|
|
pass
|
|
row.set_subtitle(subtitle)
|
|
icon = Gtk.Image.new_from_icon_name("emblem-ok-symbolic")
|
|
icon.add_css_class("success")
|
|
row.add_prefix(icon)
|
|
self.recent_group.add(row)
|
|
self._recent_rows.append(row)
|
|
|
|
self.recent_group.set_visible(bool(recent))
|
|
|
|
# ─── Utilitaires ───────────────────────────────────────────────────────
|
|
|
|
def _relative_time(self, dt: datetime) -> str:
|
|
now = datetime.now()
|
|
secs = (now - dt).total_seconds()
|
|
if secs < 60:
|
|
return "à l'instant"
|
|
if secs < 3600:
|
|
m = int(secs // 60)
|
|
return f"il y a {m} min"
|
|
if secs < 86400:
|
|
h = int(secs // 3600)
|
|
return f"il y a {h}h"
|
|
days = int(secs // 86400)
|
|
if days == 1:
|
|
return f"hier à {dt.strftime('%H:%M')}"
|
|
if days < 7:
|
|
return f"il y a {days} jours"
|
|
return dt.strftime("%d/%m/%Y")
|
|
|
|
# ─── Actions utilisateur ───────────────────────────────────────────────
|
|
|
|
def _on_refresh_clicked(self, _btn):
|
|
self.refresh_btn.set_sensitive(False)
|
|
threading.Thread(target=self._send_check_now, daemon=True).start()
|
|
|
|
def _on_install_pending_clicked(self, _btn):
|
|
self.install_now_btn.set_sensitive(False)
|
|
threading.Thread(target=self._send_install_pending, daemon=True).start()
|
|
|
|
def _send_install_pending(self):
|
|
try:
|
|
s = sock_module.socket(sock_module.AF_UNIX, sock_module.SOCK_STREAM)
|
|
s.settimeout(5)
|
|
s.connect(str(SOCKET_PATH))
|
|
s.sendall(json.dumps({"action": "install_pending"}).encode())
|
|
s.recv(1024)
|
|
s.close()
|
|
except PermissionError:
|
|
GLib.idle_add(
|
|
self._show_toast,
|
|
"Permission refusée — êtes-vous dans le groupe wheel ?",
|
|
)
|
|
GLib.idle_add(self.install_now_btn.set_sensitive, True)
|
|
except Exception as e:
|
|
GLib.idle_add(
|
|
self._show_toast,
|
|
f"Impossible de contacter le service : {e}",
|
|
)
|
|
GLib.idle_add(self.install_now_btn.set_sensitive, True)
|
|
|
|
def _send_check_now(self):
|
|
try:
|
|
s = sock_module.socket(sock_module.AF_UNIX, sock_module.SOCK_STREAM)
|
|
s.settimeout(5)
|
|
s.connect(str(SOCKET_PATH))
|
|
s.sendall(json.dumps({"action": "check_now"}).encode())
|
|
s.recv(1024)
|
|
s.close()
|
|
except PermissionError:
|
|
GLib.idle_add(
|
|
self._show_toast,
|
|
"Permission refusée — êtes-vous dans le groupe wheel ?",
|
|
)
|
|
GLib.idle_add(self.refresh_btn.set_sensitive, True)
|
|
except Exception as e:
|
|
GLib.idle_add(
|
|
self._show_toast,
|
|
f"Impossible de contacter le service : {e}",
|
|
)
|
|
GLib.idle_add(self.refresh_btn.set_sensitive, True)
|
|
|
|
def _show_toast(self, message: str) -> bool:
|
|
toast = Adw.Toast.new(message)
|
|
toast.set_timeout(4)
|
|
self.toast_overlay.add_toast(toast)
|
|
return False
|
|
|
|
|
|
class UpdaterApp(Adw.Application):
|
|
def __init__(self):
|
|
super().__init__(application_id="org.cachyos.Updater")
|
|
self.connect("activate", self._on_activate)
|
|
|
|
def _on_activate(self, app):
|
|
win = UpdaterWindow(application=app)
|
|
win.present()
|