pyssh-client/app/local_terminal.py

"""Terminal local (PowerShell / Invite de commandes) via pseudo-console Windows (pywinpty).

Independant de toute session enregistree : permet d'ouvrir un terminal Windows
fonctionnel en un clic, comme le fait MobaXterm avec ses "terminaux locaux".
"""
from __future__ import annotations

import queue
import threading

import winpty
from PySide6.QtCore import QThread, Signal

SHELL_POWERSHELL = "powershell"
SHELL_CMD = "cmd"

SHELL_LABELS = {
    SHELL_POWERSHELL: "PowerShell",
    SHELL_CMD: "Invite de commandes (cmd)",
}
SHELL_COMMANDS = {
    SHELL_POWERSHELL: "powershell.exe -NoLogo",
    SHELL_CMD: "cmd.exe",
}


class LocalTerminalConnection(QThread):
    """Meme interface de signaux que SSHConnection, pour reutiliser TerminalWidget tel quel."""

    connected = Signal()
    data_received = Signal(bytes)
    disconnected = Signal(str)
    error = Signal(str)

    def __init__(self, shell: str = SHELL_POWERSHELL, cwd: str | None = None,
                 cols: int = 80, rows: int = 24, parent=None):
        super().__init__(parent)
        self.shell = shell
        self.cwd = cwd
        self.cols = cols
        self.rows = rows
        self.pty: winpty.PtyProcess | None = None
        self._stop = threading.Event()

        # File d'envoi + thread dedie : send() est appele directement depuis le thread GUI
        # (une frappe clavier). pty.write() peut bloquer si la console Windows n'est pas
        # assez rapide a consommer son pipe d'entree ; sans decouplage, ce blocage gele
        # toute l'interface le temps de l'ecriture (pty.read() n'a pas de timeout pour
        # entrelacer l'ecriture dans la meme boucle, contrairement au canal SSH).
        self._out_queue: queue.Queue[bytes] = queue.Queue()
        self._writer_thread: threading.Thread | None = None

    def run(self) -> None:
        try:
            command = SHELL_COMMANDS.get(self.shell, SHELL_COMMANDS[SHELL_POWERSHELL])
            self.pty = self._spawn(command)

            # La taille reelle du widget est connue apres la mise en page, souvent *apres*
            # la creation de la connexion : on reapplique ici la derniere taille demandee,
            # sinon la console reste en 80x24 alors que l'affichage est bien plus grand
            # (desynchronisation = retours a la ligne et redraws incoherents).
            try:
                self.pty.setwinsize(self.rows, self.cols)
            except Exception:
                pass

            self._writer_thread = threading.Thread(target=self._writer_loop, daemon=True)
            self._writer_thread.start()
            self.connected.emit()

            while not self._stop.is_set():
                try:
                    data = self.pty.read(4096)
                except EOFError:
                    break
                if not data:
                    break
                self.data_received.emit(data.encode("utf-8", errors="replace"))

            self.disconnected.emit("Terminal ferme")
        except Exception as exc:  # noqa: BLE001
            self.error.emit(str(exc))
        finally:
            self._cleanup()

    def _spawn(self, command: str) -> winpty.PtyProcess:
        """Ouvre la pseudo-console, en privilegiant ConPTY.

        ConPTY est l'API pseudo-console native de Windows (10 1809+) : c'est un vrai
        passe-plat. L'ancien backend WinPTY, lui, *scrute* le tampon d'ecran de la console
        et resynthetise l'affichage : mesure faite, il perd des caracteres a la frappe
        (16 echos sur 23) et affiche ~32 ms de latence contre ~2,5 ms pour ConPTY, avec des
        artefacts de curseur qui saute et de saisie reecrite. On ne retombe dessus que si
        ConPTY est indisponible (Windows trop ancien).
        """
        try:
            return winpty.PtyProcess.spawn(
                command,
                dimensions=(self.rows, self.cols),
                cwd=self.cwd,
                backend=winpty.Backend.ConPTY,
            )
        except Exception:
            return winpty.PtyProcess.spawn(
                command,
                dimensions=(self.rows, self.cols),
                cwd=self.cwd,
                backend=winpty.Backend.WinPTY,
            )

    def send(self, data: bytes) -> None:
        """Appele depuis le thread GUI (frappe clavier) : ne doit jamais bloquer."""
        self._out_queue.put(data)

    def _writer_loop(self) -> None:
        """Tourne dans son propre thread : effectue les ecritures potentiellement bloquantes.

        Une erreur d'ecriture ponctuelle ne doit pas tuer le thread : sinon, toute frappe
        suivante serait silencieusement perdue pour le reste de la session (le process
        local restant lui-meme actif). On n'arrete la boucle que si le process est mort."""
        while not self._stop.is_set():
            try:
                data = self._out_queue.get(timeout=0.2)
            except queue.Empty:
                continue
            if not (self.pty and self.pty.isalive()):
                continue
            try:
                self.pty.write(data.decode("utf-8", errors="ignore"))
            except Exception:
                if not (self.pty and self.pty.isalive()):
                    break

    def resize(self, cols: int, rows: int) -> None:
        self.cols, self.rows = cols, rows
        if self.pty and self.pty.isalive():
            try:
                self.pty.setwinsize(rows, cols)
            except Exception:
                pass

    def close_connection(self) -> None:
        self._stop.set()
        self._cleanup()

    def _cleanup(self) -> None:
        try:
            if self.pty and self.pty.isalive():
                self.pty.terminate(force=True)
        except Exception:
            pass