238 lines
7.9 KiB
Python
238 lines
7.9 KiB
Python
"""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)
|