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 Any, Mapping, Optional


# ----------------------------------------------------------------------------
#  Accès paresseux au DomainContext
# ----------------------------------------------------------------------------
#
# On importe le module à l'appel pour éviter toute dépendance circulaire
# avec `agent_v0.server_v1.domain_context` (qui ne doit pas importer l'UI).
# Si l'import échoue (contexte client sans server_v1), on retombe sur None
# et les formatters gardent leur comportement générique historique.


def _get_domain_ctx(domain_id: Optional[str]):
    """Récupérer un DomainContext si possible, sinon None (fallback)."""
    if not domain_id:
        return None
    try:
        from agent_v0.server_v1.domain_context import get_domain_context  # lazy
        return get_domain_context(domain_id)
    except Exception:
        return None


def _friendly_target(description: str, domain_id: Optional[str] = None) -> str:
    """Transformer une description technique en langage métier si possible.

    Ex (tim_codage) : "DP" → "diagnostic principal"
    Ex (comptabilite) : "TVA" → "montant de TVA"
    Retombe sur la description nettoyée si aucun domaine ne matche.
    """
    base = _nettoyer_description_cible(description)
    ctx = _get_domain_ctx(domain_id)
    if ctx is None or not base:
        return base
    try:
        return ctx._apply_synonyms(base)
    except Exception:
        return base


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,
    domain_id: Optional[str] = None,
    params: Optional[Mapping[str, Any]] = None,
) -> MessageUtilisateur:
    """Message quand Léa ne trouve pas un élément à cliquer.

    Si un domaine métier est fourni, la description de la cible est
    transformée en langage métier via le DomainContext :
      - tim_codage + "DP" → "diagnostic principal"
      - comptabilite + "TVA" → "montant de TVA"

    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.

    Args:
        description_cible: Description brute de la cible.
        titre_fenetre:     Titre de la fenêtre active (pour extraire l'app).
        domain_id:         Domaine métier pour enrichir la sortie (optionnel).
        params:            Paramètres du workflow (nom_patient, num_facture...)
                           utilisés par les templates de clarification métier.
    """
    cible = _friendly_target(description_cible, domain_id) or "l'élément"
    app = _extraire_nom_application(titre_fenetre or "")

    # Si un domaine et un template de clarification existent, préférer la
    # question métier (plus pertinente que le message générique).
    ctx = _get_domain_ctx(domain_id)
    if ctx is not None and ctx.clarification_templates:
        try:
            corps = ctx.pose_clarification_question(
                {
                    "blocked_on": "target_not_found",
                    "target": description_cible or "",
                    "app": app,
                    "params": dict(params or {}),
                }
            )
        except Exception:
            corps = ""
        if corps:
            return MessageUtilisateur(
                niveau=NiveauMessage.BLOCAGE,
                titre="Léa a besoin d'aide",
                corps=corps,
                duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE],
                persistent=True,
            )

    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,
    domain_id: Optional[str] = None,
    items_count: int = 0,
    failed_count: int = 0,
    params: Optional[Mapping[str, Any]] = None,
) -> MessageUtilisateur:
    """Message à la fin d'un workflow.

    Si un domaine métier est fourni (et qu'il expose des summary_templates),
    on utilise `DomainContext.describe_workflow_outcome` pour formuler un
    rapport en langage métier (ex: "J'ai codé 14 dossiers sur 15").

    Args:
        succes:       True si l'ensemble du workflow a réussi.
        nom_workflow: Nom du workflow.
        nb_etapes:    Nombre d'étapes techniques (pour fallback générique).
        duree_s:      Durée totale en secondes.
        domain_id:    Domaine métier (optionnel).
        items_count:  Nombre d'items métier traités (ex: 15 dossiers).
        failed_count: Nombre d'items en échec.
        params:       Infos supplémentaires passées aux templates.
    """
    ctx = _get_domain_ctx(domain_id)
    if ctx is not None and ctx.summary_templates:
        try:
            corps = ctx.describe_workflow_outcome(
                workflow_name=nom_workflow,
                success=succes,
                items_count=items_count or max(1, nb_etapes),
                failed_count=failed_count,
                elapsed_s=duree_s,
                extra=dict(params or {}),
            )
        except Exception:
            corps = ""
        if corps:
            if succes and failed_count == 0:
                return MessageUtilisateur(
                    niveau=NiveauMessage.INFO,
                    titre="Léa a terminé",
                    corps=corps,
                    duree_s=6,
                )
            if succes and failed_count > 0:
                return MessageUtilisateur(
                    niveau=NiveauMessage.ATTENTION,
                    titre="Léa a terminé partiellement",
                    corps=corps,
                    duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION],
                )
            return MessageUtilisateur(
                niveau=NiveauMessage.BLOCAGE,
                titre="Léa s'arrête",
                corps=corps,
                duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE],
                persistent=True,
            )

    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,
    domain_id: Optional[str] = None,
    params: Optional[Mapping[str, Any]] = None,
) -> 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.
    Si `domain_id` est fourni, il est propagé aux formatters spécialisés
    pour produire un message en langage métier.
    """
    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, domain_id=domain_id, params=params)

    # 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, domain_id=domain_id, params=params
        )

    # 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)


# Fenêtres parasites Windows à ignorer dans les pré-vérifications.
# Ce ne sont pas des fenêtres applicatives — c'est du bruit système
# qui prend le focus de manière imprévisible.
_FENETRES_BRUIT_SYSTEME = (
    "fenêtre de dépassement de capacité",
    "overflow",               # version anglaise systray
    "program manager",
    "barre des tâches",
    "task bar",
    "cortana",
    "action center",
    "centre de notifications",
)


def est_fenetre_bruit(titre_fenetre: str) -> bool:
    """Détecter si un titre de fenêtre est du bruit système Windows.

    Ces fenêtres prennent le focus de manière imprévisible (systray overflow,
    taskbar, Program Manager) et ne sont jamais la cible d'une action utilisateur.
    """
    if not titre_fenetre:
        return True  # pas de titre = bruit
    titre_lower = titre_fenetre.lower().strip()
    if titre_lower == "unknown_window":
        return True
    return any(p in titre_lower for p in _FENETRES_BRUIT_SYSTEME)


# 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",
)