erplibre/script/todo/qemu_install_monitor.py

#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""Dashboard Textual : suivi des installations ERPLibre parallèles sur VM.

Principe « détachable » : chaque installation tourne dans un processus
DÉTACHÉ (setsid) qui écrit un fichier log et, à la fin, un marqueur
« __ERPLIBRE_EXIT__ <code> ». Ce module ne fait que VISUALISER ces fichiers :
quitter le dashboard n'arrête rien, on peut le rouvrir pour ré-attacher.

- launch_installs(...) : lance les process détachés + écrit un manifeste JSON.
- run_monitor(manifest_path) : ouvre le dashboard Textual sur un manifeste.
"""
from __future__ import annotations

import asyncio
import json
import os
import shlex
import shutil
import subprocess
import time
from pathlib import Path

try:
    from script.todo.todo_i18n import t
except Exception:  # pragma: no cover - repli si i18n indisponible

    def t(key: str) -> str:
        return key


EXIT_MARKER = "__ERPLIBRE_EXIT__"
SSH_OPTS = (
    "-o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null "
    "-o ConnectTimeout=8"
)


def session_dir() -> Path:
    """Répertoire des logs/manifestes d'installation (créé au besoin)."""
    base = Path(os.path.expanduser("~/.erplibre/qemu-install"))
    base.mkdir(parents=True, exist_ok=True)
    return base


def _launch_one(ip: str, remote_cmd: str, log_path: str) -> None:
    """Lance une install SSH DÉTACHÉE : attend le sshd, exécute, journalise
    la sortie puis écrit le marqueur de fin avec le code de sortie."""
    # Sonde de disponibilité : on attend que sshd réponde ET que cloud-init
    # soit TERMINÉ, via des connexions COURTES successives (jusqu'à ~20 min :
    # une architecture ÉMULÉE, s390x/arm64 sur hôte x86, boote lentement).
    # Au 1er boot, cloud-init régénère les clés d'hôte et REDÉMARRE sshd : une
    # session SSH longue (ex. « cloud-init status --wait ») serait alors tuée
    # (« Connection closed by remote host », exit 255 — cas Fedora). Chaque
    # itération étant une connexion neuve, un redémarrage de sshd ne casse que
    # la tentative en cours. On imprime toujours l'état (|| true) pour matcher
    # sur le TEXTE, « status: running » n'ayant pas de code de sortie fiable.
    ci_probe = (
        "if command -v cloud-init >/dev/null 2>&1; then "
        "cloud-init status 2>/dev/null || true; else echo nocloudinit; fi"
    )
    log_q = shlex.quote(log_path)
    # On écrit un message d'attente + un battement toutes les ~30 s : sinon le
    # log reste VIDE pendant tout le boot émulé et paraît « bloqué ».
    msg_wait = t("Waiting for the VM to start (boot + cloud-init)")
    msg_slow = t("(an emulated architecture can be slow; this is normal)")
    msg_ready = t("VM ready - starting the ERPLibre install")
    wrapper = (
        f"echo {shlex.quote('== ' + msg_wait + ' ==')} >> {log_q}; "
        f"echo {shlex.quote('   ' + msg_slow)} >> {log_q}; "
        f"for i in $(seq 1 240); do "
        f"st=$(ssh {SSH_OPTS} -o BatchMode=yes erplibre@{ip} "
        f"{shlex.quote(ci_probe)} 2>/dev/null); "
        f'case "$st" in '
        f"*done*|*disabled*|*error*|*degraded*|*nocloudinit*) break;; "
        f"esac; "
        f'if [ $((i % 6)) -eq 0 ]; then echo "   ... $((i*5))s" >> {log_q}; fi; '
        f"sleep 5; done; "
        f"echo {shlex.quote('== ' + msg_ready + ' ==')} >> {log_q}; "
        f"ssh {SSH_OPTS} erplibre@{ip} {shlex.quote(remote_cmd)} "
        f">> {log_q} 2>&1; "
        f'echo "{EXIT_MARKER} $?" >> {log_q}'
    )
    # setsid -f : le process survit à la fermeture du menu / du dashboard.
    subprocess.Popen(
        ["setsid", "-f", "bash", "-c", wrapper],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )


def _log_header(vm: dict, branch: str, when: str) -> str:
    """En-tête du log : date, VM, distribution, version, architecture, branche.
    Permet d'identifier l'installation d'un coup d'œil (et de ne jamais laisser
    le log vide pendant l'attente du boot)."""
    distro = vm.get("distro") or "?"
    version = vm.get("version") or ""
    arch = vm.get("arch") or "?"
    bar = "=" * 64
    return (
        f"{bar}\n"
        f"  ERPLibre — {t('installation')}\n"
        f"  Date         : {when}\n"
        f"  VM           : {vm['name']}\n"
        f"  Distribution : {distro} {version}\n"
        f"  Architecture : {arch}\n"
        f"  Branche      : {branch}\n"
        f"  IP           : {vm['ip']}\n"
        f"{bar}\n\n"
    )


def launch_installs(vms: list[dict], branch: str, remote_cmd: str) -> str:
    """vms : [{name, ip, distro?, version?, arch?}]. Lance chaque install
    détachée, écrit un manifeste et retourne son chemin. remote_cmd : script
    exécuté dans chaque VM."""
    sdir = session_dir()
    stamp = time.strftime("%Y%m%d-%H%M%S")
    when = time.strftime("%Y-%m-%d %H:%M:%S")
    logdir = sdir / stamp
    logdir.mkdir(parents=True, exist_ok=True)
    entries = []
    for vm in vms:
        log_path = str(logdir / f"{vm['name']}.log")
        # En-tête d'emblée (date/distro/version/arch) : le log n'est jamais
        # vide, l'utilisateur voit tout de suite QUOI s'installe.
        Path(log_path).write_text(_log_header(vm, branch, when))
        _launch_one(vm["ip"], remote_cmd, log_path)
        entries.append(
            {
                "name": vm["name"],
                "ip": vm["ip"],
                "distro": vm.get("distro"),
                "version": vm.get("version"),
                "arch": vm.get("arch"),
                "log": log_path,
                "ssh": f"ssh erplibre@{vm['ip']}",
            }
        )
    manifest = {
        "branch": branch,
        "started": time.time(),
        "vms": entries,
    }
    manifest_path = str(logdir / "session.json")
    Path(manifest_path).write_text(json.dumps(manifest, indent=2))
    return manifest_path


def read_status(log_path: str) -> tuple[str, int | None]:
    """(état, code) d'un log : pending / running / done / failed. Ne lit que
    la FIN du fichier (le marqueur de sortie est sur la dernière ligne) : lire
    tout le log de 30 VM chaque seconde saturait la boucle d'événements du TUI
    (lag, interface figée quand l'I/O ralentit)."""
    try:
        size = os.path.getsize(log_path)
        if size == 0:
            return "pending", None
        with open(log_path, "rb") as fh:
            if size > 4096:
                fh.seek(-4096, os.SEEK_END)
            tail = fh.read().decode(errors="replace")
    except OSError:
        return "pending", None
    if not tail.strip():
        return "pending", None
    for line in reversed(tail.splitlines()):
        if EXIT_MARKER in line:
            try:
                code = int(line.split()[-1])
            except ValueError:
                code = 1
            return ("done" if code == 0 else "failed"), code
    return "running", None


# Listes d'ignore reprises de script/test/run_parallel_test.py (erreurs/
# avertissements connus et bénins) : on réutilise la MÊME logique de détection
# que la suite de tests ERPLibre pour analyser les logs d'installation.
_LST_IGNORE_WARNING = (
    "have the same label:",
    "odoo.addons.code_generator.extractor_module_file: Ignore next error about"
    " ALTER TABLE DROP CONSTRAINT.",
)
_LST_IGNORE_ERROR = (
    "fetchmail_notify_error_to_sender",
    'odoo.sql_db: bad query: ALTER TABLE "db_backup" DROP CONSTRAINT'
    ' "db_backup_db_backup_name_unique"',
    'ERROR: constraint "db_backup_db_backup_name_unique" of relation'
    ' "db_backup" does not exist',
    'odoo.sql_db: bad query: ALTER TABLE "db_backup" DROP CONSTRAINT'
    ' "db_backup_db_backup_days_to_keep_positive"',
    'ERROR: constraint "db_backup_db_backup_days_to_keep_positive" of relation'
    ' "db_backup" does not exist',
    "odoo.addons.code_generator.extractor_module_file: Ignore next error about"
    " ALTER TABLE DROP CONSTRAINT.",
)


def scan_log_errors(log_path: str) -> tuple[int, int]:
    """(nb_erreurs, nb_avertissements) dans un log d'installation, en
    réutilisant la détection de la suite de tests ERPLibre : sous-chaîne
    « error »/« warning » (insensible à la casse) moins les listes d'ignore.
    Lit le fichier COMPLET (appelé une seule fois, à la complétion d'une VM)."""
    try:
        text = Path(log_path).read_text(errors="replace")
    except OSError:
        return 0, 0
    nerr = nwarn = 0
    for line in text.splitlines():
        low = line.lower()
        if EXIT_MARKER in line:
            continue
        if "error" in low and not any(
            ig in line for ig in _LST_IGNORE_ERROR
        ):
            nerr += 1
        if "warning" in low and not any(
            ig in line for ig in _LST_IGNORE_WARNING
        ):
            nwarn += 1
    return nerr, nwarn


def _read_new(path: str, offset: int) -> tuple[str, int]:
    """Lit le log à partir de `offset` (lecture incrémentale). Renvoie
    (nouveau_texte, nouvel_offset). Bloquant -> à appeler dans un thread."""
    try:
        with open(path, "r", errors="replace") as fh:
            fh.seek(offset)
            data = fh.read()
            return data, fh.tell()
    except OSError:
        return "", offset


# --------------------------------------------------------------------------- #
# Télémétrie, historique de durées (ETA), navigateur CLI
# --------------------------------------------------------------------------- #
def _stats_path() -> Path:
    """Fichier d'historique persistant des installations. DÉDIÉ dans
    .venv.erplibre du dépôt (repli sur ~/.erplibre si le venv est absent)."""
    try:
        venv = Path(__file__).resolve().parents[2] / ".venv.erplibre"
        venv.mkdir(parents=True, exist_ok=True)
        return venv / "qemu_install_stats.json"
    except OSError:
        return session_dir() / "stats.json"


def load_stats() -> dict:
    try:
        return json.loads(_stats_path().read_text())
    except (OSError, ValueError):
        return {}


def record_duration(distro, version, arch, secs) -> None:
    """Enregistre une install (distro + version + archi + durée + horodatage)
    dans l'historique, pour l'ETA et les moyennes par archi/distro. Garde les
    500 derniers runs."""
    data = load_stats()
    runs = data.setdefault("runs", [])
    runs.append(
        {
            "distro": distro or "?",
            "version": version or "?",
            "arch": arch or "?",
            "seconds": int(secs),
            "ts": int(time.time()),
        }
    )
    data["runs"] = runs[-500:]
    try:
        _stats_path().write_text(json.dumps(data, ensure_ascii=False))
    except OSError:
        pass


def _runs(stats=None):
    return (stats or load_stats()).get("runs", []) or []


def eta_reference(stats, arch):
    """Durée d'install de RÉFÉRENCE (médiane) pour cette archi ; repli toutes
    archis confondues. None si aucun historique."""
    runs = _runs(stats)
    secs = [r["seconds"] for r in runs if r.get("arch") == arch]
    if not secs:
        secs = [r["seconds"] for r in runs]
    if not secs:
        return None
    s = sorted(secs)
    return s[len(s) // 2]


def _avg(field, value, stats=None):
    secs = [r["seconds"] for r in _runs(stats) if r.get(field) == value]
    return (sum(secs) / len(secs)) if secs else None


def avg_by_arch(arch, stats=None):
    """(moyenne_secondes, nb_runs) pour cette archi, ou (None, 0)."""
    secs = [r["seconds"] for r in _runs(stats) if r.get("arch") == arch]
    return (sum(secs) / len(secs), len(secs)) if secs else (None, 0)


def avg_by_distro(distro, stats=None):
    """(moyenne_secondes, nb_runs) pour cette distro, ou (None, 0)."""
    secs = [r["seconds"] for r in _runs(stats) if r.get("distro") == distro]
    return (sum(secs) / len(secs), len(secs)) if secs else (None, 0)


def avg_by_version(distro, version, stats=None):
    """(moyenne_secondes, nb_runs) pour cette (distro, version), ou (None, 0)."""
    secs = [
        r["seconds"]
        for r in _runs(stats)
        if r.get("distro") == distro and r.get("version") == version
    ]
    return (sum(secs) / len(secs), len(secs)) if secs else (None, 0)


def last_run(stats=None):
    """Dernier run enregistré (dict) ou None."""
    runs = _runs(stats)
    return runs[-1] if runs else None


def _fmt_size(nbytes) -> str:
    """Octets -> « 1.2G » / « 345M » / « 12K »."""
    if nbytes is None:
        return "-"
    for unit, div in (("T", 1 << 40), ("G", 1 << 30), ("M", 1 << 20)):
        if nbytes >= div:
            return f"{nbytes / div:.1f}{unit}"
    return f"{nbytes // 1024}K"


def _fmt_secs(secs) -> str:
    """Secondes -> « 45s » / « 12m » / « 1h05 »."""
    secs = int(secs)
    if secs < 60:
        return f"{secs}s"
    if secs < 3600:
        return f"{secs // 60}m"
    return f"{secs // 3600}h{(secs % 3600) // 60:02d}"


def vm_disk_path(vm: dict) -> str:
    """Chemin du qcow2 de la VM (défaut libvirt si non fourni)."""
    return vm.get("disk") or f"/var/lib/libvirt/images/{vm['name']}.qcow2"


def disk_actual_size(path: str) -> int | None:
    """Taille RÉELLEMENT occupée du qcow2 (creux) via st_blocks."""
    try:
        st = os.stat(path)
        return int(getattr(st, "st_blocks", 0)) * 512
    except OSError:
        return None


# Navigateurs web en ligne de commande, par ordre de préférence (rendu JS
# d'abord — utile pour l'UI Odoo — puis navigateurs texte classiques).
CLI_BROWSERS = ("browsh", "carbonyl", "w3m", "links", "elinks", "lynx")


def cli_browser() -> str | None:
    """Premier navigateur CLI disponible dans le PATH, sinon None."""
    for name in CLI_BROWSERS:
        if shutil.which(name):
            return name
    return None


def _os_id() -> str:
    """ID de la distribution hôte (/etc/os-release), ex. « ubuntu », « fedora »."""
    try:
        for line in open("/etc/os-release", encoding="utf-8"):
            if line.startswith("ID="):
                return line.split("=", 1)[1].strip().strip('"').lower()
    except OSError:
        pass
    return ""


# Navigateurs CLI installables via apt/dnf/pacman (nom de paquet = binaire).
# browsh/carbonyl ne sont pas dans les dépôts standard -> non proposés ici.
INSTALLABLE_BROWSERS = (
    ("w3m", "w3m — léger, rend un peu de HTML"),
    ("lynx", "lynx — navigateur texte"),
    ("links", "links — texte / graphique"),
    ("elinks", "elinks — texte, onglets"),
)


def browser_install_command(browser="w3m") -> list | None:
    """Commande d'installation du navigateur CLI `browser` adaptée à l'OS hôte :
    apt (Ubuntu/Debian), dnf (Fedora), pacman (Arch). None si gestionnaire
    inconnu."""
    apt = ["sudo", "apt-get", "install", "-y", browser]
    dnf = ["sudo", "dnf", "install", "-y", browser]
    pac = ["sudo", "pacman", "-S", "--needed", "--noconfirm", browser]
    by_id = {
        "ubuntu": apt,
        "debian": apt,
        "linuxmint": apt,
        "fedora": dnf,
        "arch": pac,
    }
    cmd = by_id.get(_os_id())
    if cmd:
        return cmd
    if shutil.which("apt-get"):
        return apt
    if shutil.which("dnf"):
        return dnf
    if shutil.which("pacman"):
        return pac
    return None


def virsh_domstates() -> dict:
    """{nom: état} de tous les domaines libvirt (« virsh list --all »). Sert à
    détecter une VM EN PAUSE ou EFFACÉE pendant le suivi. Un seul appel virsh
    pour tout le parc (à interroger à intervalle LENT)."""
    try:
        res = subprocess.run(
            ["sudo", "virsh", "list", "--all"],
            capture_output=True,
            text=True,
            timeout=15,
        )
    except (OSError, subprocess.SubprocessError):
        return {}
    states = {}
    for line in res.stdout.splitlines():
        parts = line.split()
        # Ignore l'en-tête (« Id Name State ») et le séparateur (« ---- »,
        # un seul token). Une VM éteinte a « - » en Id : à NE PAS ignorer.
        if len(parts) < 3 or parts[0] == "Id":
            continue
        states[parts[1]] = " ".join(parts[2:])
    return states


# --------------------------------------------------------------------------- #
# Dashboard Textual
# --------------------------------------------------------------------------- #
def run_monitor(manifest_path: str, run_app: bool = True):
    """Ouvre le dashboard Textual sur un manifeste d'installation. `run_app`
    à False renvoie l'instance sans la lancer (tests headless)."""
    from textual.app import App, ComposeResult
    from textual.containers import Horizontal
    from textual.widgets import DataTable, Footer, Header, RichLog, Static

    manifest = json.loads(Path(manifest_path).read_text())
    started = manifest.get("started", time.time())
    vms = manifest["vms"]

    ICON = {
        "pending": "⏳",
        "running": "⏳",
        "done": "✅",
        "failed": "❌",
    }

    class Monitor(App):
        CSS = """
        DataTable { width: 58; border: solid $accent; }
        RichLog { border: solid $accent; }
        #telemetry { height: 1; color: $text-muted; }
        #stats { height: 1; color: $accent; }
        #statsdetail { display: none; height: auto; color: $text-muted; }
        #sshbar { height: 2; color: $text-muted; }
        """
        BINDINGS = [
            ("q", "quit", "Quitter (détaché)"),
            ("s", "ssh", "SSH"),
            ("w", "web", "Web (navigateur CLI)"),
            ("f", "follow", "Suivre"),
            ("c", "copy_log", "Copier log"),
            ("p", "pause_all", "Pause tout"),
            ("o", "resume_all", "Reprendre tout"),
        ]

        def __init__(self):
            super().__init__()
            self._offsets = {vm["name"]: 0 for vm in vms}
            self._selected = vms[0]["name"] if vms else None
            self._follow = True
            # Statuts TERMINAUX mémorisés : une VM finie n'est plus relue
            # (réduit fortement l'I/O sur un gros parc). Valeur = (état, code,
            # durée à la complétion).
            self._final = {}
            # Cache des cellules AFFICHÉES : on n'appelle update_cell (donc on
            # ne re-render) que si la valeur CHANGE -> plus de churn de rendu.
            self._cells = {}
            # Historique de durées (ETA) + dossier disque à surveiller.
            self._stats = load_stats()
            self._disk_dir = os.path.dirname(vm_disk_path(vms[0])) if vms else "/"
            # État libvirt (running/paused/gone), rafraîchi à intervalle LENT.
            self._domstate = {}
            # Erreurs détectées dans le log à la complétion : {nom: (err, warn)}.
            self._errcount = {}
            # Sommaire de stats déplié (clic) ou non.
            self._stats_open = False

        @staticmethod
        def _fmt(secs):
            mm, ss = divmod(int(secs), 60)
            return f"{mm:02d}:{ss:02d}"

        def _set_cell(self, table, name, col, value):
            key = (name, col)
            if self._cells.get(key) == value:
                return
            self._cells[key] = value
            table.update_cell(name, col, value)

        def compose(self) -> ComposeResult:
            yield Header(show_clock=True)
            table = DataTable(id="vms", cursor_type="row")
            # Clés de colonnes explicites : update_cell() les référence.
            # La colonne « ⚠ » (erreurs détectées) est à GAUCHE d'« État ».
            table.add_column("VM", key="vm")
            table.add_column("⚠", key="err")
            table.add_column("État", key="state")
            table.add_column("Durée", key="elapsed")
            table.add_column("Disque", key="disk")
            for vm in vms:
                table.add_row(
                    vm["name"], "", "⏳", "--:--", "-", key=vm["name"]
                )
            # max_lines borne la mémoire/rendu (un install verbeux × 30 VM).
            self._log = RichLog(
                id="log", highlight=False, markup=False, max_lines=5000
            )
            with Horizontal():
                yield table
                yield self._log
            # Barre de télémétrie hôte (CPU, disque, ETA parc).
            yield Static("", id="telemetry")
            # Sommaire de stats en CHIFFRES (cliquable -> détail).
            yield Static("", id="stats")
            yield Static("", id="statsdetail")
            yield Static("", id="sshbar")
            yield Footer()

        def on_mount(self) -> None:
            # Le NOMBRE de VM figure dans le titre ; le sous-titre suit la
            # progression (terminées / total + durée globale).
            self.title = f"ERPLibre — {t('install monitoring')} ({len(vms)} VM)"
            self.sub_title = f"0/{len(vms)} {t('completed')}"
            self._refresh_ssh()
            self._load_selected_log(reset=True)
            # Table toutes les 2 s (30 lectures de fin de log), suivi du log
            # sélectionné toutes les 1 s (une seule lecture incrémentale).
            self.set_interval(2.0, self._tick_table)
            self.set_interval(1.0, self._tick_log)
            # État libvirt (pause / effacée) : check LENT (appel virsh) toutes
            # les 10 s ; le tableau applique le cache à chaque tick (2 s).
            self.set_interval(10.0, self._tick_domstate)
            # Premier relevé domstate immédiat (async -> via worker).
            self.run_worker(self._tick_domstate(), exclusive=False)

        # -- helpers -------------------------------------------------------- #
        def _vm_by_name(self, name):
            return next((v for v in vms if v["name"] == name), None)

        def _refresh_ssh(self):
            vm = self._vm_by_name(self._selected)
            bar = self.query_one("#sshbar", Static)
            if vm:
                bar.update(
                    f"  {vm['ssh']}    (s = SSH · w = web :8069 · "
                    "c = copier le log · Maj+glisser = sélectionner)\n"
                    f"  Log : {vm['log']}"
                )

        def _load_selected_log(self, reset=False):
            # Lecture SYNCHRONE (mount / changement de ligne). Le suivi périodique
            # passe par _tick_log (async, I/O en thread).
            vm = self._vm_by_name(self._selected)
            if not vm:
                return
            name = vm["name"]
            if reset:
                self._log.clear()
                self._offsets[name] = 0
            new, off = _read_new(vm["log"], self._offsets.get(name, 0))
            self._offsets[name] = off
            for line in new.splitlines():
                if EXIT_MARKER not in line:
                    self._log.write(line)

        def _collect_tele(self):
            """(THREAD) chaîne de télémétrie hôte : CPU % + disque des images."""
            try:
                ncpu = os.cpu_count() or 1
                load1 = os.getloadavg()[0]
                du = shutil.disk_usage(self._disk_dir)
                used_pct = int(du.used / du.total * 100) if du.total else 0
                return (
                    f"  ⚙ CPU {min(999, int(load1 / ncpu * 100))}% "
                    f"(charge {load1:.1f}/{ncpu})   "
                    f"💽 {self._disk_dir}: {_fmt_size(du.used)}/"
                    f"{_fmt_size(du.total)} ({used_pct}%) · "
                    f"libre {_fmt_size(du.free)}"
                )
            except Exception:
                return ""

        def _collect_table(self):
            """(THREAD) statut + taille disque de chaque VM + télémétrie. AUCUNE
            mise à jour d'UI ici : uniquement des I/O bloquantes déportées."""
            disks, status, errors = {}, {}, {}
            for vm in vms:
                name = vm["name"]
                disks[name] = _fmt_size(disk_actual_size(vm_disk_path(vm)))
                if (
                    name not in self._final
                    and self._domstate.get(name) != "gone"
                ):
                    st = read_status(vm["log"])
                    status[name] = st
                    # À la complétion (succès OU échec), on ANALYSE le log
                    # complet pour signaler les erreurs — même un « succès »
                    # peut contenir des erreurs passées inaperçues.
                    if st[0] in ("done", "failed") and name not in errors:
                        errors[name] = scan_log_errors(vm["log"])
            return disks, status, self._collect_tele(), errors

        async def _tick_table(self):
            # I/O (lectures de logs, stat disque, /proc) DÉPORTÉES en thread ->
            # la boucle d'événements Textual reste fluide même sous forte
            # charge ou disque lent. Les mises à jour d'UI restent sur la boucle.
            try:
                disks, status, tele, errors = await asyncio.to_thread(
                    self._collect_table
                )
            except Exception:
                return
            self._errcount.update(errors)
            try:
                table = self.query_one("#vms", DataTable)
                now = time.time()
                remaining = []
                for vm in vms:
                    name = vm["name"]
                    self._set_cell(table, name, "disk", disks.get(name, "-"))
                    if name in self._final:
                        continue
                    ds = self._domstate.get(name)
                    if ds == "gone":
                        self._final[name] = ("deleted", None, now - started)
                        self._set_cell(
                            table, name, "state", f"❌ {t('deleted')}"
                        )
                        continue
                    st = status.get(name)
                    if st is None:
                        continue
                    state, code = st
                    if state in ("done", "failed"):
                        elapsed = now - started
                        self._final[name] = (state, code, elapsed)
                        if state == "done":
                            record_duration(
                                vm.get("distro"),
                                vm.get("version"),
                                vm.get("arch"),
                                elapsed,
                            )
                            self._stats = load_stats()
                        lbl = "✅" if state == "done" else f"❌ ({code})"
                        self._set_cell(table, name, "state", lbl)
                        self._set_cell(
                            table, name, "elapsed", self._fmt(elapsed)
                        )
                        # Colonne ⚠ (à gauche d'État) : erreurs détectées dans
                        # le log, y compris pour un « succès ».
                        self._set_cell(
                            table, name, "err",
                            self._err_label(self._errcount.get(name)),
                        )
                    elif ds == "paused":
                        self._set_cell(
                            table, name, "state", f"⏸ {t('paused')}"
                        )
                    else:
                        self._set_cell(table, name, "state", ICON[state])
                        ref = eta_reference(self._stats, vm.get("arch"))
                        if ref is not None:
                            remaining.append(max(0, ref - (now - started)))
                done = len(self._final)
                eta = (
                    f" · ETA ~{_fmt_secs(max(remaining))}" if remaining else ""
                )
                self.sub_title = (
                    f"{done}/{len(vms)} {t('completed')} · "
                    f"{self._fmt(now - started)}{eta}"
                )
                if tele:
                    self.query_one("#telemetry", Static).update(tele)
                self._update_stats()
            except Exception:
                pass

        @staticmethod
        def _err_label(counts):
            """Libellé de la colonne ⚠ : « ⚠N » si erreurs, « ⚡N » si seulement
            des avertissements, « ✓ » si log propre."""
            if not counts:
                return ""
            nerr, nwarn = counts
            if nerr:
                return f"⚠{nerr}"
            if nwarn:
                return f"⚡{nwarn}"
            return "✓"

        def _stats_counts(self):
            """Compte par catégorie (terminées, échecs, erreurs, etc.)."""
            done = fail = deleted = err_vms = warn_vms = 0
            for name, (state, _code, _el) in self._final.items():
                if state == "done":
                    done += 1
                elif state == "failed":
                    fail += 1
                elif state == "deleted":
                    deleted += 1
                counts = self._errcount.get(name)
                if counts:
                    if counts[0]:
                        err_vms += 1
                    elif counts[1]:
                        warn_vms += 1
            running = paused = 0
            for vm in vms:
                if vm["name"] in self._final:
                    continue
                if self._domstate.get(vm["name"]) == "paused":
                    paused += 1
                else:
                    running += 1
            return {
                "total": len(vms), "done": done, "fail": fail,
                "deleted": deleted, "running": running, "paused": paused,
                "err_vms": err_vms, "warn_vms": warn_vms,
            }

        def _update_stats(self):
            c = self._stats_counts()
            line = (
                f"  📊 {c['total']} VM · ✅ {c['done']} · ❌ {c['fail']} · "
                f"⏳ {c['running']} · ⏸ {c['paused']} · 🗑 {c['deleted']} · "
                f"⚠ {c['err_vms']} · ⚡ {c['warn_vms']}   "
                f"({t('click to expand')})"
            )
            self.query_one("#stats", Static).update(line)
            if self._stats_open:
                self.query_one("#statsdetail", Static).update(
                    self._render_stats_detail()
                )

        def _render_stats_detail(self):
            """Détail déplié : VM avec erreurs/avertissements + moyennes hist."""
            lines = []
            for name, (state, code, el) in self._final.items():
                counts = self._errcount.get(name)
                if counts and (counts[0] or counts[1]):
                    lines.append(
                        f"    • {name}: ⚠{counts[0]} erreurs, "
                        f"⚡{counts[1]} avert. ({self._fmt(el)})"
                    )
            if not lines:
                lines.append(f"    {t('No error detected.')}")
            return "\n".join(lines)

        async def _tick_log(self):
            if not self._follow:
                return
            vm = self._vm_by_name(self._selected)
            if not vm:
                return
            name = vm["name"]
            try:
                new, off = await asyncio.to_thread(
                    _read_new, vm["log"], self._offsets.get(name, 0)
                )
            except Exception:
                return
            self._offsets[name] = off
            for line in new.splitlines():
                if EXIT_MARKER not in line:
                    self._log.write(line)

        async def _tick_domstate(self):
            # Relevé LENT (subprocess virsh) DÉPORTÉ en thread : ne bloque plus
            # la boucle. Une VM absente de « virsh list » est EFFACÉE (« gone »).
            try:
                states = await asyncio.to_thread(virsh_domstates)
            except Exception:
                return
            for vm in vms:
                self._domstate[vm["name"]] = states.get(vm["name"], "gone")

        # -- events --------------------------------------------------------- #
        def on_data_table_row_highlighted(self, event) -> None:
            name = event.row_key.value
            if name and name != self._selected:
                self._selected = name
                self._refresh_ssh()
                self._load_selected_log(reset=True)

        def action_follow(self) -> None:
            self._follow = not self._follow

        def action_ssh(self) -> None:
            vm = self._vm_by_name(self._selected)
            if not vm:
                return
            with self.suspend():
                os.system(f"ssh {SSH_OPTS} erplibre@{vm['ip']} || true")

        def action_web(self) -> None:
            """Ouvre l'UI web de la VM (Odoo :8069) dans un navigateur CLI
            (browsh/carbonyl/w3m/links/elinks/lynx). Surtout utile une fois
            l'installation TERMINÉE."""
            vm = self._vm_by_name(self._selected)
            if not vm or not vm.get("ip"):
                self.notify("Pas d'IP pour cette VM.", title="Web")
                return
            browser = cli_browser()
            if not browser:
                browser = self._install_cli_browser()
                if not browser:
                    self.notify(
                        "Aucun navigateur CLI disponible.",
                        title="Web",
                        severity="warning",
                    )
                    return
            url = f"http://{vm['ip']}:8069"
            with self.suspend():
                print(f"→ {browser} {url}")
                rc = os.system(f"{browser} {shlex.quote(url)}")
                # Diagnostic : sinon le navigateur « clignote » et revient au
                # TUI sans qu'on voie l'erreur (souvent Odoo pas démarré).
                print(f"\n[{browser}] terminé (code {rc}).")
                if rc != 0:
                    print(
                        "La page ne s'est peut-être pas affichée : Odoo n'est "
                        "pas démarré sur :8069, ou réseau/pare-feu. Vérifie que "
                        "le service Odoo tourne dans la VM (make run / systemd)."
                    )
                try:
                    input("Entrée pour revenir au suivi… ")
                except EOFError:
                    pass

        def _install_cli_browser(self):
            """Demande QUEL navigateur CLI installer (w3m/lynx/links/elinks),
            affiche la commande, l'exécute après validation. Renvoie le binaire
            désormais disponible, ou None."""
            with self.suspend():
                print("Aucun navigateur CLI installé. Lequel installer ?")
                for i, (b, desc) in enumerate(INSTALLABLE_BROWSERS, 1):
                    print(f"  [{i}] {desc}{' *' if i == 1 else ''}")
                sel = input("Choix (numéro, vide = w3m) : ").strip()
                browser = INSTALLABLE_BROWSERS[0][0]
                try:
                    idx = int(sel) - 1
                    if 0 <= idx < len(INSTALLABLE_BROWSERS):
                        browser = INSTALLABLE_BROWSERS[idx][0]
                except ValueError:
                    pass
                cmd = browser_install_command(browser)
                if not cmd:
                    print(
                        "Gestionnaire de paquets inconnu : installez "
                        f"« {browser} » manuellement."
                    )
                    input("Entrée… ")
                    return None
                printable = " ".join(cmd)
                print(f"Commande : {printable}")
                ans = input(
                    "Installer maintenant ? (o/N) : "
                ).strip().lower()
                if ans not in ("o", "oui", "y", "yes"):
                    return None
                rc = os.system(printable)
                print(f"\nInstallation terminée (code {rc}).")
                input("Entrée pour continuer… ")
            return cli_browser()

        def action_copy_log(self) -> None:
            """Copie le log complet de la VM sélectionnée dans le
            presse-papiers (OSC 52 ; marche aussi à travers SSH)."""
            vm = self._vm_by_name(self._selected)
            if not vm:
                return
            try:
                text = Path(vm["log"]).read_text(errors="replace")
            except OSError:
                return
            self.copy_to_clipboard(text)
            self.notify(
                f"Log de {vm['name']} copié ({len(text)} car.)",
                title="Presse-papiers",
            )

        def on_click(self, event) -> None:
            # Clic sur le sommaire de stats -> déplie / replie le détail.
            w = getattr(event, "widget", None)
            if w is not None and getattr(w, "id", None) == "stats":
                self._stats_open = not self._stats_open
                self.query_one("#statsdetail").display = self._stats_open
                self._update_stats()

        # -- pause / reprise de tout le parc -------------------------------- #
        @staticmethod
        def _virsh_bulk(action, names):
            for n in names:
                try:
                    subprocess.run(
                        ["sudo", "virsh", action, n],
                        capture_output=True,
                        text=True,
                        timeout=30,
                    )
                except (OSError, subprocess.SubprocessError):
                    pass

        def action_pause_all(self) -> None:
            """Met en PAUSE (virsh suspend) toutes les VM en cours d'exécution.
            L'install reprend là où elle en était après « Reprendre »."""
            self.run_worker(self._bulk_worker("suspend"), exclusive=False)

        def action_resume_all(self) -> None:
            """REPREND (virsh resume) toutes les VM en pause ; le suivi des
            logs continue automatiquement (offsets conservés)."""
            self.run_worker(self._bulk_worker("resume"), exclusive=False)

        async def _bulk_worker(self, action):
            want = "running" if action == "suspend" else "paused"
            targets = [
                vm["name"]
                for vm in vms
                if self._domstate.get(vm["name"]) == want
            ]
            if not targets:
                self.notify(
                    t("No running VM to pause.")
                    if action == "suspend"
                    else t("No paused VM to resume.")
                )
                return
            await asyncio.to_thread(self._virsh_bulk, action, targets)
            verb = t("paused") if action == "suspend" else t("resumed")
            self.notify(f"{len(targets)} VM {verb}.")
            # Rafraîchit tout de suite l'état libvirt (pause/reprise visible).
            self.run_worker(self._tick_domstate(), exclusive=False)

    app = Monitor()
    if run_app:
        app.run()
    return app