"""Installateur exécuté juste avant l'extinction du PC. Installe les paquets qui nécessitaient un redémarrage (kernel, systemd, etc.) puis laisse systemd terminer la séquence d'arrêt normalement. """ import logging import os import subprocess import sys import threading import time from datetime import datetime import state as state_mod from package_manager import upgrade_cached_packages, is_pacman_locked logging.basicConfig( level=logging.INFO, format="%(asctime)s [SHUTDOWN] %(message)s", stream=sys.stdout, ) logger = logging.getLogger(__name__) # ── Écran de progression ──────────────────────────────────────────────────── CYAN = "\033[1;36m" GREEN = "\033[1;32m" RED = "\033[1;31m" YELLOW = "\033[1;33m" RESET = "\033[0m" BOLD = "\033[1m" CLEAR = "\033[2J\033[H" # Efface l'écran, curseur en haut à gauche HIDE_CURSOR = "\033[?25l" SHOW_CURSOR = "\033[?25h" SPINNER_FRAMES = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏" class ProgressDisplay: """Affiche un écran de progression pendant l'installation des mises à jour. Utilise Plymouth s'il est actif (CachyOS avec splash activé), sinon prend le contrôle du TTY1 directement. """ def __init__(self, packages: list[dict]): self._packages = packages self._active = False self._status = "Installation en cours…" self._spin_thread: threading.Thread | None = None self._tty_fd: int | None = None self._use_plymouth = self._probe_plymouth() def _probe_plymouth(self) -> bool: try: r = subprocess.run( ["plymouth", "--ping"], capture_output=True, timeout=2 ) return r.returncode == 0 except Exception: return False # ── Interface publique ────────────────────────────────────────────────── def start(self): self._active = True if self._use_plymouth: self._plymouth_show() else: self._tty_start() def update(self, message: str): self._status = message if self._use_plymouth: self._plym_msg(f"⟳ {message}") def finish(self, success: bool): self._active = False if success: final = f"✓ Mises à jour installées avec succès" color = GREEN else: final = f"✗ Échec de l'installation — voir les journaux" color = RED if self._use_plymouth: self._plym_msg(final) time.sleep(1.5) else: if self._spin_thread: self._spin_thread.join(timeout=1) self._tty_finish(final, color) time.sleep(2) self._tty_close() # ── Plymouth ──────────────────────────────────────────────────────────── def _plymouth_show(self): names = ", ".join(p["name"] for p in self._packages[:5]) extra = f" (+{len(self._packages) - 5})" if len(self._packages) > 5 else "" self._plym_msg(f"⟳ Installation des mises à jour… {names}{extra}") def _plym_msg(self, text: str): try: subprocess.run( ["plymouth", "message", f"--text={text}"], capture_output=True, timeout=5 ) except Exception: pass # ── TTY direct ───────────────────────────────────────────────────────── def _tty_start(self): try: subprocess.run(["chvt", "1"], capture_output=True, timeout=2) except Exception: pass for path in ("/dev/tty1", "/dev/console"): try: self._tty_fd = os.open(path, os.O_WRONLY | os.O_NOCTTY) break except Exception: continue self._tty_write(self._build_screen()) self._spin_thread = threading.Thread(target=self._spin_loop, daemon=True) self._spin_thread.start() def _build_screen(self) -> str: n = len(self._packages) pkg_lines = "" for pkg in self._packages[:10]: name = pkg["name"] old_v = pkg.get("old_version", "") new_v = pkg.get("new_version", "") pkg_lines += f" {CYAN}•{RESET} {BOLD}{name:<30}{RESET} {old_v} → {new_v}\n" if n > 10: pkg_lines += f" {CYAN}•{RESET} … et {n - 10} autre(s) paquet(s)\n" return ( f"{CLEAR}{HIDE_CURSOR}" f"\n\n\n" f" {CYAN}{'─' * 60}{RESET}\n" f"\n" f" {BOLD}CachyOS Updater{RESET}\n" f"\n" f" Installation de {BOLD}{n} mise{'s' if n > 1 else ''} à jour{RESET} " f"avant extinction…\n" f"\n" f"{pkg_lines}" f"\n" f" {YELLOW}Ne pas éteindre ou débrancher votre PC.{RESET}\n" f"\n" f" {CYAN}{'─' * 60}{RESET}\n" f"\n" f" " # Point de départ du spinner (sera écrasé) ) def _spin_loop(self): i = 0 while self._active: c = SPINNER_FRAMES[i % len(SPINNER_FRAMES)] self._tty_write(f"\r {CYAN}{c}{RESET} {self._status}\033[K") i += 1 time.sleep(0.1) def _tty_finish(self, message: str, color: str): self._tty_write(f"\r {color}{message}{RESET}\033[K\n") def _tty_close(self): if self._tty_fd is not None: self._tty_write(SHOW_CURSOR) os.close(self._tty_fd) self._tty_fd = None def _tty_write(self, text: str): if self._tty_fd is not None: try: os.write(self._tty_fd, text.encode("utf-8")) except Exception: pass # ── Point d'entrée ────────────────────────────────────────────────────────── def run(): state = state_mod.load() pending = state.get("pending_restart", []) if not pending: logger.info("Aucune mise à jour en attente — extinction normale") sys.exit(0) names = [p["name"] for p in pending] logger.info( f"Installation de {len(names)} paquet(s) avant extinction : {', '.join(names)}" ) if is_pacman_locked(): logger.error("pacman est verrouillé, impossible d'installer les mises à jour") sys.exit(1) display = ProgressDisplay(pending) display.start() try: display.update(f"Installation de {len(names)} paquet(s)…") result = upgrade_cached_packages(names) except Exception as e: logger.exception(f"Erreur inattendue : {e}") display.finish(success=False) sys.exit(1) if result.success: logger.info("Mises à jour installées avec succès avant extinction") now = datetime.now().isoformat() for pkg in pending: state["installed_history"].append({**pkg, "installed_at": now}) state["installed_history"] = state["installed_history"][-200:] state["pending_restart"] = [] installed_set = set(names) state["available_updates"] = [ u for u in state.get("available_updates", []) if u["name"] not in installed_set ] state_mod.save(state) display.finish(success=True) sys.exit(0) else: logger.error(f"Échec de l'installation : {result.error}") display.finish(success=False) sys.exit(1)