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" ) 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) # ── 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 _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()