rpa_vision_v3/agent_v0/agent_v1/ui/messages.py

# agent_v1/ui/messages.py
"""
Formatage des messages utilisateur pour Léa.

Convertit les codes d'erreur techniques (`target_not_found`, `no_screen_change`...)
en phrases en français naturel, orientées action, adaptées à un utilisateur non
technique (secrétaire médicale, TIM).

Trois niveaux de sévérité sont définis :
  - INFO      — Léa fait son travail normalement
  - ATTENTION — Quelque chose de léger (ralentissement, retry)
  - BLOCAGE   — Léa a besoin d'aide, elle rend la main

Le module est 100% pur (pas d'I/O, pas d'UI) : testable sans mocks lourds.
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from enum import Enum
from typing import Optional


class NiveauMessage(Enum):
    """Niveaux hiérarchiques des messages affichés à l'utilisateur."""

    INFO = "info"            # Fond vert clair, disparaît tout seul, 3-5s
    ATTENTION = "attention"  # Fond orange clair, disparaît tout seul, 7s
    BLOCAGE = "blocage"      # Fond rouge clair, reste affiché, 15s+


# Durée d'affichage par défaut (secondes), par niveau
DUREE_PAR_NIVEAU: dict[NiveauMessage, int] = {
    NiveauMessage.INFO: 4,
    NiveauMessage.ATTENTION: 7,
    NiveauMessage.BLOCAGE: 15,
}

# Icône textuelle par niveau (compatible plyer/Windows/Linux)
ICONE_PAR_NIVEAU: dict[NiveauMessage, str] = {
    NiveauMessage.INFO: "i",
    NiveauMessage.ATTENTION: "!",
    NiveauMessage.BLOCAGE: "?",
}


@dataclass
class MessageUtilisateur:
    """Un message prêt à être affiché à l'utilisateur.

    Attributes:
        niveau:  Hiérarchie (info/attention/blocage)
        titre:   Titre court de la notification (≤60 caractères)
        corps:   Corps du message en français naturel
        duree_s: Durée d'affichage recommandée (secondes)
        persistent: Si True, l'utilisateur doit fermer manuellement
    """

    niveau: NiveauMessage
    titre: str
    corps: str
    duree_s: int
    persistent: bool = False

    def to_dict(self) -> dict:
        """Sérialiser le message (utile pour les tests et le logging)."""
        return {
            "niveau": self.niveau.value,
            "titre": self.titre,
            "corps": self.corps,
            "duree_s": self.duree_s,
            "persistent": self.persistent,
        }


# ============================================================================
#  Helpers d'extraction
# ============================================================================


def _extraire_nom_application(titre_fenetre: str) -> str:
    """Extraire le nom de l'application à partir d'un titre de fenêtre.

    Les titres Windows suivent généralement le format :
        "Document.txt – Bloc-notes"
        "Ma Page - Google Chrome"
        "Sans titre — Paint"

    On retourne la partie après le dernier séparateur, ou le titre entier.
    """
    if not titre_fenetre:
        return ""
    titre = titre_fenetre.strip()
    # Chercher le dernier séparateur parmi " – ", " — ", " - "
    for sep in (" – ", " — ", " - "):
        if sep in titre:
            return titre.rsplit(sep, 1)[-1].strip()
    return titre


def _nettoyer_description_cible(description: str) -> str:
    """Nettoyer la description technique d'une cible pour l'afficher.

    Supprime les caractères techniques (guillemets inutiles, ':').
    """
    if not description:
        return ""
    desc = description.strip()
    # Retirer les guillemets encapsulants
    desc = desc.strip("'\"`")
    # Limiter la longueur
    if len(desc) > 80:
        desc = desc[:77] + "..."
    return desc


# ============================================================================
#  Formattage des messages techniques → humains
# ============================================================================


def formatter_cible_non_trouvee(
    description_cible: str,
    titre_fenetre: Optional[str] = None,
) -> MessageUtilisateur:
    """Message quand Léa ne trouve pas un élément à cliquer.

    Exemple avant :
        target_not_found: 'bonjour' dans *bonjour, – Bloc-notes
    Exemple après :
        Léa a besoin d'aide
        Je ne trouve pas "bonjour" dans le Bloc-notes. Peux-tu cliquer
        dessus toi-même ? Je reprends ensuite.
    """
    cible = _nettoyer_description_cible(description_cible) or "l'élément"
    app = _extraire_nom_application(titre_fenetre or "")

    if app:
        corps = (
            f"Je ne trouve pas « {cible} » dans {app}. "
            f"Peux-tu cliquer dessus toi-même ? Je reprends ensuite."
        )
    else:
        corps = (
            f"Je ne trouve pas « {cible} » à l'écran. "
            f"Peux-tu le faire toi-même ? Je reprends ensuite."
        )

    return MessageUtilisateur(
        niveau=NiveauMessage.BLOCAGE,
        titre="Léa a besoin d'aide",
        corps=corps,
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE],
        persistent=True,
    )


def formatter_fenetre_incorrecte(
    titre_actuel: str,
    titre_attendu: str,
) -> MessageUtilisateur:
    """Message quand la fenêtre active n'est pas celle attendue.

    Exemple avant :
        Fenêtre incorrecte: 'Program Manager' (attendu: 'Lea : Explorateur de fichiers')
    Exemple après :
        Léa attend une fenêtre
        J'attends « Explorateur de fichiers » mais c'est « Program Manager »
        qui est affiché. Peux-tu ouvrir la bonne fenêtre ?
    """
    app_actuelle = _extraire_nom_application(titre_actuel) or "une autre fenêtre"
    app_attendue = _extraire_nom_application(titre_attendu) or titre_attendu

    corps = (
        f"J'attends « {app_attendue} » mais c'est « {app_actuelle} » "
        f"qui est affiché. Peux-tu ouvrir la bonne fenêtre ?"
    )

    return MessageUtilisateur(
        niveau=NiveauMessage.BLOCAGE,
        titre="Léa attend une fenêtre",
        corps=corps,
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE],
        persistent=True,
    )


def formatter_ecran_inchange(action_type: str = "") -> MessageUtilisateur:
    """Message quand l'action n'a pas eu d'effet visible.

    Exemple avant :
        Ecran inchange apres l'action
    Exemple après :
        Léa vérifie
        Mon clic n'a pas eu l'air de marcher. Je vais réessayer ou te
        rendre la main si ça ne passe pas.
    """
    actions_fr = {
        "click": "Mon clic",
        "type": "Ma saisie",
        "key_combo": "Mon raccourci clavier",
        "scroll": "Mon défilement",
    }
    quoi = actions_fr.get(action_type, "Mon action")

    corps = (
        f"{quoi} n'a pas eu l'air de marcher. Je vais réessayer, "
        f"ou te rendre la main si ça ne passe pas."
    )

    return MessageUtilisateur(
        niveau=NiveauMessage.ATTENTION,
        titre="Léa vérifie",
        corps=corps,
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION],
    )


def formatter_connexion_perdue(hote_serveur: str = "") -> MessageUtilisateur:
    """Message quand la connexion avec le serveur est perdue.

    Rassurant : on dit qu'on va réessayer automatiquement.
    """
    corps = (
        "J'ai perdu le lien avec le serveur. Je retente automatiquement, "
        "pas besoin d'intervenir."
    )

    return MessageUtilisateur(
        niveau=NiveauMessage.ATTENTION,
        titre="Léa est déconnectée",
        corps=corps,
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION],
    )


def formatter_connexion_retablie() -> MessageUtilisateur:
    """Message quand la connexion serveur est rétablie."""
    return MessageUtilisateur(
        niveau=NiveauMessage.INFO,
        titre="Léa",
        corps="C'est bon, la connexion est revenue. Je continue.",
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.INFO],
    )


def formatter_debut_workflow(nom_workflow: str, nb_etapes: int = 0) -> MessageUtilisateur:
    """Message au démarrage d'un workflow de replay."""
    if nb_etapes > 0:
        corps = (
            f"Je démarre « {nom_workflow} » ({nb_etapes} étapes). "
            f"Je t'indique mon avancement."
        )
    else:
        corps = f"Je démarre « {nom_workflow} ». Je t'indique mon avancement."

    return MessageUtilisateur(
        niveau=NiveauMessage.INFO,
        titre="Léa démarre",
        corps=corps,
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.INFO],
    )


def formatter_etape_workflow(
    etape_actuelle: int,
    nb_etapes: int,
    description: str = "",
) -> MessageUtilisateur:
    """Message pour la progression d'une étape."""
    if description:
        desc = _nettoyer_description_cible(description)
        corps = f"Étape {etape_actuelle}/{nb_etapes} — {desc}"
    else:
        corps = f"Étape {etape_actuelle}/{nb_etapes}"

    return MessageUtilisateur(
        niveau=NiveauMessage.INFO,
        titre="Léa avance",
        corps=corps,
        duree_s=3,
    )


def formatter_retry(action_type: str = "", tentative: int = 2) -> MessageUtilisateur:
    """Message quand Léa retente une action."""
    corps = (
        f"Je retente (tentative {tentative}). Ça arrive parfois, "
        f"l'écran était peut-être en cours de chargement."
    )
    return MessageUtilisateur(
        niveau=NiveauMessage.ATTENTION,
        titre="Léa retente",
        corps=corps,
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION],
    )


def formatter_ralentissement() -> MessageUtilisateur:
    """Message quand Léa prend plus de temps que prévu."""
    return MessageUtilisateur(
        niveau=NiveauMessage.ATTENTION,
        titre="Léa prend son temps",
        corps="Je vais plus lentement que prévu. L'écran met du temps à répondre.",
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION],
    )


def formatter_fin_workflow(
    succes: bool,
    nom_workflow: str = "",
    nb_etapes: int = 0,
    duree_s: float = 0.0,
) -> MessageUtilisateur:
    """Message à la fin d'un workflow."""
    if succes:
        if nom_workflow and nb_etapes > 0:
            corps = (
                f"C'est fait ! « {nom_workflow} » est terminé "
                f"({nb_etapes} étapes en {int(duree_s)}s)."
            )
        else:
            corps = "C'est fait ! Tout s'est bien passé."
        return MessageUtilisateur(
            niveau=NiveauMessage.INFO,
            titre="Léa a terminé",
            corps=corps,
            duree_s=6,
        )
    else:
        corps = (
            "Je n'ai pas pu terminer. Je te rends la main, "
            "tu peux continuer à partir de là où je me suis arrêtée."
        )
        return MessageUtilisateur(
            niveau=NiveauMessage.BLOCAGE,
            titre="Léa s'arrête",
            corps=corps,
            duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE],
            persistent=True,
        )


def formatter_erreur_generique(message_technique: str) -> MessageUtilisateur:
    """Formater un message d'erreur technique non catégorisé.

    On essaie de détecter les motifs connus dans le message technique pour
    le router vers le bon formatter spécialisé, sinon on emballe le message.
    """
    if not message_technique:
        return MessageUtilisateur(
            niveau=NiveauMessage.ATTENTION,
            titre="Léa",
            corps="J'ai rencontré un petit souci. Je continue.",
            duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION],
        )

    msg_lower = message_technique.lower()

    # target_not_found[:...]
    if "target_not_found" in msg_lower:
        # Essayer d'extraire la description après le ':'
        match = re.match(r"target_not_found[:\s]*(.*)", message_technique, re.IGNORECASE)
        desc = match.group(1).strip() if match else ""
        return formatter_cible_non_trouvee(desc)

    # Fenêtre incorrecte: 'X' (attendu: 'Y')
    if "fenêtre incorrecte" in msg_lower or "fenetre incorrecte" in msg_lower:
        # Extraire actuel et attendu
        m_actuel = re.search(r"[:,]\s*['\"]([^'\"]+)['\"]", message_technique)
        m_attendu = re.search(r"attendu[:\s]*['\"]([^'\"]+)['\"]", message_technique)
        actuel = m_actuel.group(1) if m_actuel else ""
        attendu = m_attendu.group(1) if m_attendu else ""
        return formatter_fenetre_incorrecte(actuel, attendu)

    # Ecran inchangé
    if "inchang" in msg_lower or "no_screen_change" in msg_lower:
        return formatter_ecran_inchange()

    # Policy abort / supervise
    if "policy_abort" in msg_lower or "visual_resolve_failed" in msg_lower:
        return formatter_cible_non_trouvee(message_technique)

    # Fallback : message technique tronqué
    msg_tronque = message_technique.strip()
    if len(msg_tronque) > 120:
        msg_tronque = msg_tronque[:117] + "..."

    return MessageUtilisateur(
        niveau=NiveauMessage.ATTENTION,
        titre="Léa",
        corps=f"J'ai rencontré un souci : {msg_tronque}",
        duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION],
    )


# ============================================================================
#  Détection fenêtre Léa (utilisé par l'executor pour ignorer sa propre UI)
# ============================================================================


# Motifs qui identifient une fenêtre appartenant à Léa (l'agent lui-même).
# On utilise des regex avec \b pour éviter les faux positifs sur des noms
# contenant "lea" (ex: "cléa.txt", "leapfrog", "replay").
_MOTIFS_FENETRE_LEA_REGEX = (
    r"\bléa\b",
    r"\blea\b(?!p)",  # "lea" mot entier, pas "leapfrog"
    r"lea\s*[—–\-:]",  # "Lea —", "Lea -", "Lea :"
    r"léa\s*[—–\-:]",
    r"\bassistante ia\b",
    r"\bléa ia\b",
    r"\blea ia\b",
)


def est_fenetre_lea(titre_fenetre: str) -> bool:
    """Détecter si un titre de fenêtre appartient à l'agent Léa lui-même.

    Utilisé pour éviter que Léa ne se considère comme une fenêtre intrusive
    dans ses propres pré-vérifications.

    Utilise des regex avec des word boundaries pour éviter les faux positifs
    sur des noms de fichiers contenant "lea" (ex: "cléa.txt", "replay.log").
    """
    if not titre_fenetre:
        return False
    titre_lower = titre_fenetre.lower().strip()
    return any(re.search(motif, titre_lower) for motif in _MOTIFS_FENETRE_LEA_REGEX)


# Conservé pour rétro-compatibilité avec le code qui listait MOTIFS_FENETRE_LEA
MOTIFS_FENETRE_LEA = (
    "léa",
    "lea —",
    "léa —",
    "lea -",
    "léa -",
    "lea assistante",
    "léa assistante",
    "lea : ",
    "léa : ",
    "assistante ia",
)