rpa_vision_v3/agent_v0/server_v1/vm_controller.py

# server_v1/vm_controller.py
"""
Contrôleur de VM Windows via libvirt (virsh).
Injection d'événements HID (souris/clavier) au niveau de l'hyperviseur.
C'est le "bras armé" du Stagiaire pour l'exécution GHOST (sans agent).
"""

import subprocess
import logging
import time

logger = logging.getLogger("vm_controller")

class VMController:
    def __init__(self, domain_name: str):
        self.domain_name = domain_name

    def start_vm(self):
        """Démarre la VM si elle est éteinte."""
        try:
            logger.info(f"🚀 Démarrage de la VM {self.domain_name}...")
            subprocess.run(f"virsh start {self.domain_name}", shell=True, check=True)
            return True
        except Exception as e:
            logger.error(f"❌ Impossible de démarrer la VM: {e}")
            return False

    def stop_vm(self, force=False):
        """Arrête la VM proprement (ou force l'arrêt)."""
        cmd = "destroy" if force else "shutdown"
        try:
            logger.info(f"🛑 Arrêt de la VM {self.domain_name} ({cmd})...")
            subprocess.run(f"virsh {cmd} {self.domain_name}", shell=True, check=True)
            return True
        except Exception as e:
            logger.error(f"❌ Erreur lors de l'arrêt: {e}")
            return False

    def get_status(self) -> str:
        """Retourne l'état actuel de la VM (running, shut off, etc.)."""
        try:
            res = subprocess.check_output(f"virsh domstate {self.domain_name}", shell=True)
            return res.decode().strip()
        except:
            return "unknown"

    def create_checkpoint(self, checkpoint_name: str = "before_workflow"):
        """Crée un snapshot de la VM pour pouvoir revenir en arrière en cas d'erreur."""
        try:
            logger.info(f"📸 Création du checkpoint '{checkpoint_name}' pour {self.domain_name}...")
            # On utilise --atomic pour garantir l'intégrité
            subprocess.run(f"virsh snapshot-create-as {self.domain_name} {checkpoint_name} --atomic", shell=True, check=True)
            return True
        except Exception as e:
            logger.error(f"❌ Échec de création du checkpoint: {e}")
            return False

    def restore_checkpoint(self, checkpoint_name: str = "before_workflow"):
        """Restaure la VM à un état précédent instantanément."""
        try:
            logger.warning(f"🔄 Restauration du checkpoint '{checkpoint_name}' pour {self.domain_name}...")
            # On force la restauration
            subprocess.run(f"virsh snapshot-revert {self.domain_name} {checkpoint_name} --force", shell=True, check=True)
            return True
        except Exception as e:
            logger.error(f"❌ Échec de la restauration: {e}")
            return False

    def inject_click(self, x_pct: float, y_pct: float, button: str = "left"):
        """
        Injecte un clic de souris aux coordonnées proportionnelles (0.0-1.0).
        Utilise l'interface QEMU via virsh pour la précision absolue.
        """
        try:
            # Note: Pour QEMU/KVM, on utilise souvent l'interface moniteur 'qemu-monitor-command'
            # pour envoyer des coordonnées absolues si une tablette USB est présente (évite le drift).
            
            # Exemple de commande QEMU Monitor pour un clic absolu
            # (nécessite que la VM ait un périphérique tablette USB configuré)
            cmd = f"virsh qemu-monitor-command {self.domain_name} --hmp 'mouse_set 0 0 0 {x_pct} {y_pct}'"
            subprocess.run(cmd, shell=True, check=True)
            
            # Simulation du clic bouton
            click_cmd = f"virsh qemu-monitor-command {self.domain_name} --hmp 'mouse_button 1'" # 1 = Left
            subprocess.run(click_cmd, shell=True, check=True)
            time.sleep(0.05)
            release_cmd = f"virsh qemu-monitor-command {self.domain_name} --hmp 'mouse_button 0'"
            subprocess.run(release_cmd, shell=True, check=True)
            
            logger.info(f"🖱️ Clic GHOST injecté dans {self.domain_name} à ({x_pct}, {y_pct})")
        except Exception as e:
            logger.error(f"❌ Erreur Injection Clic: {e}")

    def inject_text(self, text: str):
        """
        Injecte du texte dans la VM en traduisant les caractères en séquences de touches.
        Gère les majuscules (via Shift) et les caractères standards.
        """
        try:
            logger.info(f"⌨️ Saisie GHOST dans {self.domain_name} : '{text}'")
            for char in text:
                self._send_char(char)
                # Petit délai pour simuler une frappe humaine et éviter la saturation du buffer
                time.sleep(0.02)
        except Exception as e:
            logger.error(f"❌ Erreur Injection Texte: {e}")

    def inject_key_combo(self, keys: list):
        """
        Exécute une combinaison de touches (ex: ['ctrl', 'alt', 'delete']).
        """
        try:
            combo = "+".join(keys)
            cmd = f"virsh sendkey {self.domain_name} {combo}"
            subprocess.run(cmd, shell=True, check=True)
            logger.info(f"⌨️ Combo GHOST : {combo}")
        except Exception as e:
            logger.error(f"❌ Erreur Combo: {e}")

    def _send_char(self, char: str):
        """Traduit un caractère unique en commande virsh sendkey."""
        # Mapping des caractères spéciaux pour virsh
        special_map = {
            " ": "space", "\n": "enter", "\t": "tab", ".": "dot", 
            ",": "comma", "-": "minus", "_": "underscore", "/": "slash"
        }

        if char in special_map:
            key = special_map[char]
            cmd = f"virsh sendkey {self.domain_name} {key}"
        elif char.isupper():
            key = char.lower()
            cmd = f"virsh sendkey {self.domain_name} shift+{key}"
        else:
            key = char
            cmd = f"virsh sendkey {self.domain_name} {key}"
            
        subprocess.run(cmd, shell=True, check=True)

if __name__ == "__main__":
    # Test rapide sur une VM de démo
    controller = VMController("win10_demo")
    # controller.inject_click(0.5, 0.5) # Clic au centre