# agent_v1/ui/chat_window.py """ Fenetre de chat Lea integree au systray — version tkinter native. Remplace l'approche Edge browser par une vraie fenetre tkinter integree. Design professionnel, theme clair, ancree en bas a droite de l'ecran. Tourne dans son propre thread daemon pour ne pas bloquer pystray. """ import logging import os import threading import time from datetime import datetime from typing import Any, Callable, Dict, Optional logger = logging.getLogger(__name__) # FeedbackBus : import fail-safe (le ChatWindow doit tourner même si python-socketio # n'est pas installé sur le poste client, par exemple ancienne installation Pauline) try: from ..network.feedback_bus import FeedbackBusClient _HAS_FEEDBACK_BUS = True except Exception: FeedbackBusClient = None # type: ignore _HAS_FEEDBACK_BUS = False # --------------------------------------------------------------------------- # Theme — palette professionnelle claire # --------------------------------------------------------------------------- BG_COLOR = "#FAFBFC" # Fond principal HEADER_BG = "#2563EB" # Bleu en-tete HEADER_FG = "#FFFFFF" # Texte en-tete blanc MSG_LEA_BG = "#EFF6FF" # Fond messages Lea (bleu clair) MSG_USER_BG = "#E8F5E9" # Fond messages utilisateur (vert plus marque) MSG_LEA_FG = "#1E40AF" # Texte Lea (bleu fonce) MSG_USER_FG = "#1B5E20" # Texte utilisateur (vert plus fonce, meilleur contraste) TIMESTAMP_FG = "#9CA3AF" # Horodatages gris BTN_BG = "#3B82F6" # Boutons bleu BTN_HOVER_BG = "#2563EB" # Boutons hover BTN_FG = "#FFFFFF" # Texte boutons blanc INPUT_BG = "#FFFFFF" # Champ de saisie blanc INPUT_FG = "#1F2937" # Texte saisie noir BORDER_COLOR = "#E5E7EB" # Bordures gris clair STATUS_CONNECTED = "#22C55E" # Vert connecte STATUS_DISCONNECTED = "#EF4444" # Rouge deconnecte QUICK_BTN_BG = "#F3F4F6" # Fond boutons rapides QUICK_BTN_FG = "#374151" # Texte boutons rapides QUICK_BTN_HOVER = "#E5E7EB" # Hover boutons rapides SCROLLBAR_BG = "#E5E7EB" # Fond scrollbar SCROLLBAR_FG = "#9CA3AF" # Curseur scrollbar MSG_BORDER_COLOR = "#D1D5DB" # Bordure subtile des bulles de messages # Bulle paused_need_help (J3.5) — alerte non bloquante, asset démo majeur PAUSED_BG = "#FEF3C7" # Jaune pâle PAUSED_BORDER = "#F59E0B" # Orange ambré PAUSED_FG = "#92400E" # Brun foncé (lisible sur fond jaune) PAUSED_BTN_RESUME_BG = "#22C55E" # Vert PAUSED_BTN_RESUME_HOVER = "#16A34A" PAUSED_BTN_ABORT_BG = "#9CA3AF" # Gris neutre (pas dramatique) PAUSED_BTN_ABORT_HOVER = "#6B7280" # Bulle "Léa exécute" (J3.4) — distincte des bulles chat normales ACTION_BG = "#F1F5F9" # Gris très clair (différencie d'une réponse chat) ACTION_BORDER = "#CBD5E1" # Gris pâle ACTION_FG = "#1E293B" # Gris foncé ACTION_META_FG = "#94A3B8" # Métadonnées en gris discret ACTION_ICON_RUN = "#3B82F6" # Bleu (en cours) ACTION_ICON_OK = "#22C55E" # Vert (succès) ACTION_ICON_ERR = "#EF4444" # Rouge (échec) ACTION_ICON_INFO = "#64748B" # Gris (neutre) # Dimensions — confortables WIN_WIDTH = 600 WIN_HEIGHT = 800 MARGIN = 14 MSG_WRAP_WIDTH = WIN_WIDTH - 90 # Tailles de police — bien lisibles FONT_TITLE = ("Segoe UI", 15, "bold") FONT_MSG = ("Segoe UI", 13) FONT_MSG_ITALIC = ("Segoe UI", 12, "italic") FONT_TIMESTAMP = ("Segoe UI", 10) FONT_SYSTEM = ("Segoe UI", 10, "italic") FONT_QUICK_BTN = ("Segoe UI", 11) FONT_INPUT = ("Segoe UI", 13) FONT_STATUS = ("Segoe UI", 10) FONT_CLOSE_BTN = ("Segoe UI", 13) FONT_SEND_BTN = ("Segoe UI", 13) FONT_RESIZE_GRIP = ("Segoe UI", 10) # --------------------------------------------------------------------------- # Templates de bulles "Léa exécute" (J3.4) # Chaque template prend un payload et retourne (icon, icon_color, title). # Les libellés sont volontairement neutres : le contexte métier vient du # payload (workflow, action, message), pas de hardcoding. # --------------------------------------------------------------------------- def _tpl_action_started(payload: Dict[str, Any]) -> tuple: wf = payload.get("workflow") or "?" return ("▶", ACTION_ICON_RUN, f"Démarrage : {wf}") def _tpl_action_progress(payload: Dict[str, Any]) -> tuple: cur = payload.get("current", "?") tot = payload.get("total", "?") step = payload.get("step") title = step if step else f"Étape {cur}/{tot}" return ("⋯", ACTION_ICON_RUN, str(title)) def _tpl_done(payload: Dict[str, Any]) -> tuple: success = bool(payload.get("success", True)) msg = payload.get("message") or ("Terminé" if success else "Échec") if success: return ("✓", ACTION_ICON_OK, str(msg)) return ("✗", ACTION_ICON_ERR, str(msg)) def _tpl_need_confirm(payload: Dict[str, Any]) -> tuple: action = payload.get("action") or {} desc = action.get("description") if isinstance(action, dict) else None title = desc or "Validation requise" return ("?", ACTION_ICON_RUN, str(title)) def _tpl_step_result(payload: Dict[str, Any]) -> tuple: status = (payload.get("status") or "").lower() msg = payload.get("message") or status or "Étape terminée" if status in ("ok", "success", "approved"): return ("✓", ACTION_ICON_OK, str(msg)) if status in ("error", "failed"): return ("✗", ACTION_ICON_ERR, str(msg)) return ("·", ACTION_ICON_INFO, str(msg)) def _tpl_resumed(payload: Dict[str, Any]) -> tuple: return ("→", ACTION_ICON_OK, "Reprise") _ACTION_TEMPLATES = { "lea:action_started": _tpl_action_started, "lea:action_progress": _tpl_action_progress, "lea:done": _tpl_done, "lea:need_confirm": _tpl_need_confirm, "lea:step_result": _tpl_step_result, "lea:resumed": _tpl_resumed, } def _extract_meta(payload: Dict[str, Any]) -> str: """Métadonnées techniques en pied de bulle (workflow, étape, replay_id court).""" parts = [] wf = payload.get("workflow") if wf: parts.append(str(wf)) cur, tot = payload.get("current"), payload.get("total") if cur is not None and tot is not None: parts.append(f"étape {cur}/{tot}") rid = payload.get("replay_id") if rid: parts.append(f"#{str(rid)[-6:]}") return " • ".join(parts) class ChatWindow: """Fenetre de chat Lea en tkinter natif. Tourne dans un thread daemon independant. Thread-safe via root.after(). Interface compatible avec l'ancien ChatWindow (toggle, show, hide, destroy). """ def __init__( self, server_client: Optional[Any] = None, on_start_callback: Optional[Callable[[str], None]] = None, server_host: str = "localhost", chat_port: int = 5004, shared_state: Optional[Any] = None, ) -> None: self._server_client = server_client self._on_start_callback = on_start_callback self._server_host = server_host self._chat_port = chat_port # Etat partage avec le systray (source de verite unique) self._shared_state = shared_state # Etat self._visible = False self._destroyed = False self._root = None self._ready = threading.Event() self._messages = [] # historique local self._bus: Optional[Any] = None # FeedbackBusClient (J3.3, peut rester None) self._active_paused_bubble: Optional[Dict[str, Any]] = None # bulle paused active (J3.5) # S'abonner aux changements de l'etat partage if self._shared_state is not None: self._shared_state.on_change(self._on_shared_state_change) # Demarrer tkinter dans un thread daemon self._thread = threading.Thread( target=self._run_tk_loop, daemon=True, name="chat-window-tk", ) self._thread.start() # Attendre que la fenetre soit prete (max 5s) self._ready.wait(timeout=5.0) logger.info("ChatWindow tkinter initialisee") # ====================================================================== # Interface publique (thread-safe) # ====================================================================== def toggle(self) -> None: """Afficher/masquer la fenetre de chat.""" if self._destroyed or self._root is None: return if self._visible: self.hide() else: self.show() def show(self) -> None: """Afficher la fenetre.""" if self._destroyed or self._root is None: return self._root.after(0, self._do_show) def hide(self) -> None: """Masquer la fenetre (sans la detruire).""" if self._destroyed or self._root is None: return self._root.after(0, self._do_hide) def is_visible(self) -> bool: """Verifie si la fenetre est affichee.""" return self._visible def destroy(self) -> None: """Fermer definitivement la fenetre et arreter le thread.""" if self._destroyed: return self._destroyed = True if self._root is not None: try: self._root.after(0, self._do_destroy) except Exception: pass def update_server_client(self, server_client: Any) -> None: """Mettre a jour le client serveur (appele si cree apres la fenetre).""" self._server_client = server_client def _on_shared_state_change(self, state) -> None: """Callback appele quand l'etat partage change (depuis le systray ou ailleurs). Affiche un message dans le chat si un enregistrement demarre ou s'arrete depuis le systray. """ if self._root is None or self._destroyed: return # Detecter la transition enregistrement demarre (depuis le systray) if state.is_recording and not getattr(self, "_last_known_recording", False): name = state.recording_name self._add_lea_message( f"Enregistrement en cours : \u00ab {name} \u00bb\n" "Montrez-moi les \u00e9tapes, j'observe !" ) # Detecter la transition enregistrement arrete (depuis le systray) if not state.is_recording and getattr(self, "_last_known_recording", False): count = state.actions_count self._add_lea_message( f"C'est not\u00e9 ! J'ai m\u00e9moris\u00e9 {count} actions." ) # Detecter la transition replay if state.is_replay_active and not getattr(self, "_last_known_replay", False): self._add_lea_message("Replay en cours...") if not state.is_replay_active and getattr(self, "_last_known_replay", False): self._add_lea_message("Replay termin\u00e9. C'est fait !") # Memoriser l'etat pour detecter les transitions self._last_known_recording = state.is_recording self._last_known_replay = state.is_replay_active # ====================================================================== # Construction de la fenetre (thread tkinter) # ====================================================================== def _run_tk_loop(self) -> None: """Boucle principale tkinter dans un thread daemon.""" try: # Activer le DPI awareness sur Windows AVANT de créer Tk() # Sans ça, tkinter rend tout minuscule sur les écrans haute résolution try: import ctypes ctypes.windll.shcore.SetProcessDpiAwareness(1) except Exception: pass import tkinter as tk from tkinter import font as tkfont self._tk = tk root = tk.Tk() self._root = root # Appliquer le scaling DPI de Windows à tkinter try: dpi = root.winfo_fpixels('1i') scale_factor = dpi / 72.0 root.tk.call('tk', 'scaling', scale_factor) except Exception: pass # Fenetre avec barre de titre native (redimensionnable par l'utilisateur) root.title("Léa — Assistante") root.configure(bg=BG_COLOR) root.attributes('-topmost', True) root.minsize(400, 500) # Taille et position (bas-droite de l'ecran) screen_w = root.winfo_screenwidth() screen_h = root.winfo_screenheight() x = screen_w - WIN_WIDTH - 12 y = screen_h - WIN_HEIGHT - 50 root.geometry(f"{WIN_WIDTH}x{WIN_HEIGHT}+{x}+{y}") # Intercepter la fermeture (X) pour masquer au lieu de fermer root.protocol("WM_DELETE_WINDOW", self._do_hide) # Taille minimum pour eviter une fenetre trop petite root.minsize(350, 450) # Ombre / bordure simulee root.configure(highlightbackground="#CBD5E1", highlightthickness=1) # Construire les widgets self._build_header(root) self._build_status_bar(root) self._build_messages_area(root) self._build_quick_actions(root) self._build_input_area(root) self._build_resize_grip(root) # Message d'accueil — divulgation IA obligatoire (Article 50, Reglement IA) self._add_lea_message( "Bonjour ! Je suis L\u00e9a, une assistante bas\u00e9e sur " "l'intelligence artificielle.\n" "Je peux apprendre vos t\u00e2ches r\u00e9p\u00e9titives " "et les refaire \u00e0 votre place.\n" "Que puis-je faire pour vous ?" ) # Demarrer masquee root.withdraw() self._visible = False # Verification connexion periodique self._check_connection_status() # Signaler que la fenetre est prete self._ready.set() # Demarrer le bus feedback Lea (events 'lea:*' temps reel) self._start_feedback_bus() # Boucle tkinter root.mainloop() except Exception as e: logger.error("Erreur initialisation ChatWindow tkinter : %s", e) self._ready.set() def _build_header(self, root) -> None: """Barre de titre personnalisee (draggable).""" tk = self._tk header = tk.Frame(root, bg=HEADER_BG, height=44) header.pack(fill=tk.X, side=tk.TOP) header.pack_propagate(False) # Titre — police agrandie title_label = tk.Label( header, text="L\u00e9a \u2014 Assistante", bg=HEADER_BG, fg=HEADER_FG, font=FONT_TITLE, anchor="w", padx=12, ) title_label.pack(side=tk.LEFT, fill=tk.X, expand=True) # Bouton fermer close_btn = tk.Label( header, text="\u2715", bg=HEADER_BG, fg=HEADER_FG, font=FONT_CLOSE_BTN, padx=14, cursor="hand2", ) close_btn.pack(side=tk.RIGHT) close_btn.bind("", lambda e: self.hide()) close_btn.bind("", lambda e: close_btn.configure(bg="#DC2626")) close_btn.bind("", lambda e: close_btn.configure(bg=HEADER_BG)) # Drag support (deplacer la fenetre) self._drag_data = {"x": 0, "y": 0} header.bind("", self._on_drag_start) header.bind("", self._on_drag_motion) title_label.bind("", self._on_drag_start) title_label.bind("", self._on_drag_motion) def _build_status_bar(self, root) -> None: """Barre de statut avec indicateur de connexion.""" tk = self._tk status_frame = tk.Frame(root, bg="#F8FAFC", height=26) status_frame.pack(fill=tk.X, side=tk.TOP) status_frame.pack_propagate(False) # Separateur tk.Frame(root, bg=BORDER_COLOR, height=1).pack(fill=tk.X, side=tk.TOP) # Indicateur de connexion (point colore) self._status_dot = tk.Label( status_frame, text="\u25CF", fg=STATUS_DISCONNECTED, bg="#F8FAFC", font=FONT_STATUS, padx=8, ) self._status_dot.pack(side=tk.LEFT) self._status_label = tk.Label( status_frame, text="D\u00e9connect\u00e9e", bg="#F8FAFC", fg=TIMESTAMP_FG, font=FONT_STATUS, ) self._status_label.pack(side=tk.LEFT) def _build_messages_area(self, root) -> None: """Zone de messages scrollable.""" tk = self._tk # Conteneur avec scrollbar msg_container = tk.Frame(root, bg=BG_COLOR) msg_container.pack(fill=tk.BOTH, expand=True, side=tk.TOP) # Canvas + Scrollbar pour le scroll self._canvas = tk.Canvas( msg_container, bg=BG_COLOR, highlightthickness=0, bd=0, ) scrollbar = tk.Scrollbar( msg_container, orient=tk.VERTICAL, command=self._canvas.yview, bg=SCROLLBAR_BG, troughcolor=BG_COLOR, width=10, ) self._canvas.configure(yscrollcommand=scrollbar.set) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) self._canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) # Frame interne scrollable self._msg_frame = tk.Frame(self._canvas, bg=BG_COLOR) self._msg_frame_id = self._canvas.create_window( (0, 0), window=self._msg_frame, anchor=tk.NW, width=WIN_WIDTH - 22, # largeur moins scrollbar ) # Bindings pour le scroll self._msg_frame.bind( "", lambda e: self._canvas.configure(scrollregion=self._canvas.bbox("all")), ) self._canvas.bind( "", lambda e: self._canvas.itemconfig(self._msg_frame_id, width=e.width), ) # Molette souris self._canvas.bind_all( "", lambda e: self._canvas.yview_scroll(int(-1 * (e.delta / 120)), "units"), ) def _build_quick_actions(self, root) -> None: """Barre de boutons d'actions rapides.""" tk = self._tk # Separateur tk.Frame(root, bg=BORDER_COLOR, height=1).pack(fill=tk.X, side=tk.TOP) actions_frame = tk.Frame(root, bg="#F8FAFC", height=48) actions_frame.pack(fill=tk.X, side=tk.TOP, padx=0, pady=0) actions_frame.pack_propagate(False) buttons = [ ("\U0001f393 Apprenez-moi", self._on_quick_record), ("\u25b6\ufe0f Lancer une t\u00e2che", self._on_quick_tasks), ("\u23f9\ufe0f Arr\u00eater", self._on_quick_stop), ] for text, cmd in buttons: btn = tk.Label( actions_frame, text=text, bg=QUICK_BTN_BG, fg=QUICK_BTN_FG, font=FONT_QUICK_BTN, padx=8, pady=5, cursor="hand2", relief=tk.FLAT, ) btn.pack(side=tk.LEFT, padx=4, pady=5) btn.bind("", lambda e, c=cmd: c()) btn.bind("", lambda e, b=btn: b.configure(bg=QUICK_BTN_HOVER)) btn.bind("", lambda e, b=btn: b.configure(bg=QUICK_BTN_BG)) # Separateur tk.Frame(root, bg=BORDER_COLOR, height=1).pack(fill=tk.X, side=tk.TOP) def _build_input_area(self, root) -> None: """Zone de saisie avec bouton envoyer.""" tk = self._tk input_frame = tk.Frame(root, bg=BG_COLOR, height=48) input_frame.pack(fill=tk.X, side=tk.BOTTOM, padx=MARGIN, pady=(MARGIN, 4)) input_frame.pack_propagate(False) # Bouton joindre un fichier (📎) attach_btn = tk.Label( input_frame, text="\U0001f4ce", bg=BG_COLOR, fg=BTN_BG, font=FONT_SEND_BTN, padx=6, pady=3, cursor="hand2", ) attach_btn.pack(side=tk.LEFT, padx=(0, 4)) attach_btn.bind("", lambda e: self._on_attach_file()) attach_btn.bind("", lambda e: attach_btn.configure(fg=BTN_HOVER_BG)) attach_btn.bind("", lambda e: attach_btn.configure(fg=BTN_BG)) # Champ de saisie — police agrandie self._input_var = tk.StringVar() self._input_entry = tk.Entry( input_frame, textvariable=self._input_var, bg=INPUT_BG, fg=INPUT_FG, font=FONT_INPUT, relief=tk.FLAT, bd=0, highlightthickness=1, highlightbackground=BORDER_COLOR, highlightcolor=BTN_BG, insertbackground=INPUT_FG, ) self._input_entry.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0, 8)) self._input_entry.insert(0, "") self._input_entry.bind("", lambda e: self._on_send()) # Placeholder self._input_entry.insert(0, "Votre message...") self._input_entry.configure(fg=TIMESTAMP_FG) self._input_entry.bind("", self._on_input_focus_in) self._input_entry.bind("", self._on_input_focus_out) # Bouton envoyer send_btn = tk.Label( input_frame, text="\u27A4", bg=BTN_BG, fg=BTN_FG, font=FONT_SEND_BTN, padx=12, pady=3, cursor="hand2", ) send_btn.pack(side=tk.RIGHT) send_btn.bind("", lambda e: self._on_send()) send_btn.bind("", lambda e: send_btn.configure(bg=BTN_HOVER_BG)) send_btn.bind("", lambda e: send_btn.configure(bg=BTN_BG)) def _build_resize_grip(self, root) -> None: """Poignee de redimensionnement en bas a droite de la fenetre.""" tk = self._tk grip_frame = tk.Frame(root, bg=BG_COLOR, height=16) grip_frame.pack(fill=tk.X, side=tk.BOTTOM) # Symbole de grip aligne a droite self._resize_grip = tk.Label( grip_frame, text="\u25E2", # triangle bas-droite bg=BG_COLOR, fg="#B0B8C4", font=FONT_RESIZE_GRIP, cursor="bottom_right_corner", padx=4, ) self._resize_grip.pack(side=tk.RIGHT, anchor=tk.SE) # Donnees de resize self._resize_data = {"x": 0, "y": 0, "w": 0, "h": 0} self._resize_grip.bind("", self._on_resize_start) self._resize_grip.bind("", self._on_resize_motion) # ====================================================================== # Placeholder input # ====================================================================== def _on_input_focus_in(self, event) -> None: """Retire le placeholder quand le champ prend le focus.""" if self._input_entry.get() == "Votre message...": self._input_entry.delete(0, self._tk.END) self._input_entry.configure(fg=INPUT_FG) def _on_input_focus_out(self, event) -> None: """Remet le placeholder si le champ est vide.""" if not self._input_entry.get().strip(): self._input_entry.delete(0, self._tk.END) self._input_entry.insert(0, "Votre message...") self._input_entry.configure(fg=TIMESTAMP_FG) # ====================================================================== # Drag (deplacer la fenetre) # ====================================================================== def _on_drag_start(self, event) -> None: self._drag_data["x"] = event.x self._drag_data["y"] = event.y def _on_drag_motion(self, event) -> None: if self._root is None: return dx = event.x - self._drag_data["x"] dy = event.y - self._drag_data["y"] x = self._root.winfo_x() + dx y = self._root.winfo_y() + dy self._root.geometry(f"+{x}+{y}") # ====================================================================== # Resize (redimensionner la fenetre via le grip) # ====================================================================== def _on_resize_start(self, event) -> None: """Debut du redimensionnement — memorise la position et la taille.""" if self._root is None: return self._resize_data["x"] = event.x_root self._resize_data["y"] = event.y_root self._resize_data["w"] = self._root.winfo_width() self._resize_data["h"] = self._root.winfo_height() def _on_resize_motion(self, event) -> None: """Redimensionne la fenetre en suivant le curseur.""" if self._root is None: return dx = event.x_root - self._resize_data["x"] dy = event.y_root - self._resize_data["y"] new_w = max(350, self._resize_data["w"] + dx) new_h = max(450, self._resize_data["h"] + dy) x = self._root.winfo_x() y = self._root.winfo_y() self._root.geometry(f"{new_w}x{new_h}+{x}+{y}") # ====================================================================== # Actions (thread tkinter) # ====================================================================== def _do_show(self) -> None: """Affiche la fenetre (appele dans le thread tkinter).""" if self._root is None: return self._root.deiconify() self._root.lift() self._root.focus_force() self._input_entry.focus_set() self._visible = True def _do_hide(self) -> None: """Masque la fenetre (appele dans le thread tkinter).""" if self._root is None: return self._root.withdraw() self._visible = False def _do_destroy(self) -> None: """Detruit la fenetre (appele dans le thread tkinter).""" if self._bus is not None: try: self._bus.stop() except Exception: pass self._bus = None if self._root is not None: try: self._root.quit() self._root.destroy() except Exception: pass self._root = None self._visible = False # ====================================================================== # FeedbackBus — bulles temps reel pendant l'execution (J3.3) # ====================================================================== def _start_feedback_bus(self) -> None: """Demarrer la connexion au bus 'lea:*' si flag actif et lib disponible.""" if not _HAS_FEEDBACK_BUS: logger.debug("FeedbackBus non disponible (python-socketio manquant)") return flag = os.environ.get("LEA_FEEDBACK_BUS", "0").lower() if flag not in ("1", "true", "yes", "on"): return try: url = f"http://{self._server_host}:{self._chat_port}" token = os.environ.get("RPA_API_TOKEN", "") or None self._bus = FeedbackBusClient(url, token=token, on_event=self._on_lea_event) self._bus.start() logger.info("FeedbackBus demarre : %s", url) except Exception: logger.debug("FeedbackBus init silenced", exc_info=True) self._bus = None def _on_lea_event(self, event: str, payload: Dict[str, Any]) -> None: """Callback bus → bulle Lea. Thread-safe : helpers utilisent root.after.""" payload = payload or {} # J3.5 : la pause supervisée a sa propre bulle interactive if event == "lea:paused": self._add_paused_bubble(payload) return if event in ("lea:resumed", "lea:done"): self._close_active_paused_bubble(reason=event) # on continue pour afficher la bulle d'action (cf. dispatch ci-dessous) # Acks bus (resume_acked, abort_acked) : silencieux côté UI if event in ("lea:resume_acked", "lea:abort_acked"): return # J3.4 : bulle "Léa exécute" stylisée (séparée des bulles chat normales) rendered = _ACTION_TEMPLATES.get(event) if rendered is None: # Event inconnu : on affiche en bulle d'action neutre self._add_action_bubble( icon="·", icon_color=ACTION_ICON_INFO, title=event.removeprefix("lea:"), meta=_extract_meta(payload), ) return icon, icon_color, title = rendered(payload) self._add_action_bubble( icon=icon, icon_color=icon_color, title=title, meta=_extract_meta(payload), ) # ------------------------------------------------------------------ # Bulle "Léa exécute" stylisée (J3.4) # ------------------------------------------------------------------ def _add_action_bubble( self, icon: str, icon_color: str, title: str, meta: str = "", ) -> None: if self._root is None: return self._root.after(0, lambda: self._render_action_bubble(icon, icon_color, title, meta)) def _render_action_bubble( self, icon: str, icon_color: str, title: str, meta: str, ) -> None: tk = self._tk if getattr(self, "_msg_frame", None) is None: return now = datetime.now().strftime("%H:%M") container = tk.Frame(self._msg_frame, bg=BG_COLOR) container.pack(fill=tk.X, padx=MARGIN, pady=3) inner = tk.Frame( container, bg=ACTION_BG, padx=10, pady=6, highlightbackground=ACTION_BORDER, highlightthickness=1, ) inner.pack(anchor=tk.W, padx=(0, 70), fill=tk.X) row = tk.Frame(inner, bg=ACTION_BG) row.pack(fill=tk.X, anchor=tk.W) tk.Label( row, text=icon, bg=ACTION_BG, fg=icon_color, font=("Segoe UI", 13, "bold"), padx=4, ).pack(side=tk.LEFT) tk.Label( row, text=title, bg=ACTION_BG, fg=ACTION_FG, font=FONT_MSG, anchor="w", justify=tk.LEFT, wraplength=MSG_WRAP_WIDTH - 60, ).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(2, 0)) if meta: tk.Label( inner, text=f"{meta} • {now}", bg=ACTION_BG, fg=ACTION_META_FG, font=FONT_TIMESTAMP, anchor="w", ).pack(fill=tk.X, anchor=tk.W, pady=(2, 0)) # ------------------------------------------------------------------ # Bulle paused_need_help interactive (J3.5) # ------------------------------------------------------------------ def _add_paused_bubble(self, payload: Dict[str, Any]) -> None: """Ajouter une bulle paused interactive (asset démo : Léa demande de l'aide).""" if self._root is None: return self._root.after(0, lambda: self._render_paused_bubble(payload)) def _render_paused_bubble(self, payload: Dict[str, Any]) -> None: tk = self._tk if getattr(self, "_msg_frame", None) is None: return replay_id = str(payload.get("replay_id", "") or "") workflow = payload.get("workflow", "?") reason = payload.get("reason") or "Action incertaine — j'ai besoin de votre validation." completed = payload.get("completed", 0) total = payload.get("total", "?") now = datetime.now().strftime("%H:%M") container = tk.Frame(self._msg_frame, bg=BG_COLOR) container.pack(fill=tk.X, padx=MARGIN, pady=6) inner = tk.Frame( container, bg=PAUSED_BG, padx=14, pady=12, highlightbackground=PAUSED_BORDER, highlightthickness=2, ) inner.pack(anchor=tk.W, padx=(0, 50), fill=tk.X) tk.Label( inner, text=f"⏸ Pause supervisée • {now}", bg=PAUSED_BG, fg=PAUSED_FG, font=("Segoe UI", 12, "bold"), anchor="w", ).pack(fill=tk.X, anchor=tk.W) tk.Label( inner, text=reason, bg=PAUSED_BG, fg=PAUSED_FG, font=FONT_MSG, wraplength=MSG_WRAP_WIDTH - 30, anchor="w", justify=tk.LEFT, ).pack(fill=tk.X, anchor=tk.W, pady=(6, 0)) tk.Label( inner, text=f"{workflow} — étape {completed}/{total}", bg=PAUSED_BG, fg=TIMESTAMP_FG, font=FONT_TIMESTAMP, anchor="w", ).pack(fill=tk.X, anchor=tk.W, pady=(4, 8)) btn_frame = tk.Frame(inner, bg=PAUSED_BG) btn_frame.pack(fill=tk.X, anchor=tk.W) btn_resume = tk.Button( btn_frame, text="Continuer", bg=PAUSED_BTN_RESUME_BG, fg="white", font=FONT_QUICK_BTN, padx=14, pady=4, bd=0, cursor="hand2", activebackground=PAUSED_BTN_RESUME_HOVER, activeforeground="white", command=lambda: self._on_paused_resume(replay_id), ) btn_resume.pack(side=tk.LEFT, padx=(0, 8)) btn_abort = tk.Button( btn_frame, text="Annuler", bg=PAUSED_BTN_ABORT_BG, fg="white", font=FONT_QUICK_BTN, padx=14, pady=4, bd=0, cursor="hand2", activebackground=PAUSED_BTN_ABORT_HOVER, activeforeground="white", command=lambda: self._on_paused_abort(replay_id), ) btn_abort.pack(side=tk.LEFT) self._active_paused_bubble = { "container": container, "inner": inner, "btn_resume": btn_resume, "btn_abort": btn_abort, "replay_id": replay_id, } def _close_active_paused_bubble(self, reason: str) -> None: if self._active_paused_bubble is None or self._root is None: return self._root.after(0, lambda: self._do_close_paused_bubble(reason)) def _do_close_paused_bubble(self, reason: str) -> None: bubble = self._active_paused_bubble if bubble is None: return try: bubble["btn_resume"].config(state="disabled") bubble["btn_abort"].config(state="disabled") label_text = { "lea:resumed": "→ Reprise", "lea:done": "→ Terminé", }.get(reason, f"→ {reason}") self._tk.Label( bubble["inner"], text=label_text, bg=PAUSED_BG, fg=PAUSED_FG, font=FONT_TIMESTAMP, anchor="w", ).pack(fill="x", anchor="w", pady=(6, 0)) except Exception: logger.debug("close paused bubble silenced", exc_info=True) self._active_paused_bubble = None def _on_paused_resume(self, replay_id: str) -> None: if not replay_id or self._bus is None or not self._bus.connected: self._add_lea_message("⚠ Bus indisponible — impossible de relancer") return self._bus.resume_replay(replay_id) if self._active_paused_bubble: try: self._active_paused_bubble["btn_resume"].config(state="disabled") self._active_paused_bubble["btn_abort"].config(state="disabled") except Exception: pass def _on_paused_abort(self, replay_id: str) -> None: if self._bus is None or not self._bus.connected: self._add_lea_message("⚠ Bus indisponible — impossible d'annuler") return self._bus.abort_replay(replay_id) if self._active_paused_bubble: try: self._active_paused_bubble["btn_resume"].config(state="disabled") self._active_paused_bubble["btn_abort"].config(state="disabled") except Exception: pass # ====================================================================== # Ajout de messages dans la zone de chat # ====================================================================== def _add_lea_message(self, text: str) -> None: """Ajoute un message de Lea (cote gauche, fond bleu).""" if self._root is None: return self._root.after(0, lambda: self._render_message( sender="lea", text=text, bg=MSG_LEA_BG, fg=MSG_LEA_FG, prefix="\U0001f4ac ", )) def _add_user_message(self, text: str) -> None: """Ajoute un message utilisateur (cote droit, fond vert).""" if self._root is None: return self._root.after(0, lambda: self._render_message( sender="user", text=text, bg=MSG_USER_BG, fg=MSG_USER_FG, prefix="", )) def _add_system_message(self, text: str) -> None: """Ajoute un message systeme (centre, gris).""" if self._root is None: return self._root.after(0, lambda: self._render_message( sender="system", text=text, bg=BG_COLOR, fg=TIMESTAMP_FG, prefix="", )) def _render_message( self, sender: str, text: str, bg: str, fg: str, prefix: str ) -> None: """Rendu d'un message dans la zone de chat (thread tkinter).""" tk = self._tk if self._msg_frame is None: return now = datetime.now().strftime("%H:%M") # Conteneur du message — espacement vertical accru msg_container = tk.Frame(self._msg_frame, bg=BG_COLOR) msg_container.pack(fill=tk.X, padx=MARGIN, pady=4) if sender == "system": # Message systeme centre label = tk.Label( msg_container, text=text, bg=bg, fg=fg, font=FONT_SYSTEM, wraplength=MSG_WRAP_WIDTH, ) label.pack(anchor=tk.CENTER) elif sender == "user": # Message utilisateur a droite — bulle avec bordure subtile inner = tk.Frame( msg_container, bg=bg, padx=12, pady=8, highlightbackground=MSG_BORDER_COLOR, highlightthickness=1, ) inner.pack(anchor=tk.E, padx=(50, 0)) # Horodatage tk.Label( inner, text=f"Vous \u2022 {now}", bg=bg, fg=TIMESTAMP_FG, font=FONT_TIMESTAMP, anchor=tk.E, ).pack(fill=tk.X, anchor=tk.E) # Texte — police agrandie tk.Label( inner, text=text, bg=bg, fg=fg, font=FONT_MSG, wraplength=MSG_WRAP_WIDTH - 40, justify=tk.LEFT, anchor=tk.W, ).pack(fill=tk.X, anchor=tk.W, pady=(3, 0)) else: # Message Lea a gauche — bulle avec bordure subtile inner = tk.Frame( msg_container, bg=bg, padx=12, pady=8, highlightbackground=MSG_BORDER_COLOR, highlightthickness=1, ) inner.pack(anchor=tk.W, padx=(0, 50)) # Horodatage tk.Label( inner, text=f"L\u00e9a \u2022 {now}", bg=bg, fg=TIMESTAMP_FG, font=FONT_TIMESTAMP, anchor=tk.W, ).pack(fill=tk.X, anchor=tk.W) # Texte — police agrandie tk.Label( inner, text=f"{prefix}{text}", bg=bg, fg=fg, font=FONT_MSG, wraplength=MSG_WRAP_WIDTH - 40, justify=tk.LEFT, anchor=tk.W, ).pack(fill=tk.X, anchor=tk.W, pady=(3, 0)) # Stocker dans l'historique self._messages.append({"sender": sender, "text": text, "time": now}) # Scroll vers le bas self._scroll_to_bottom() def _scroll_to_bottom(self) -> None: """Scrolle la zone de messages vers le bas.""" if self._canvas is not None: self._canvas.update_idletasks() self._canvas.yview_moveto(1.0) def _show_typing_indicator(self) -> None: """Affiche un indicateur 'Lea reflechit...'.""" if self._root is None: return self._root.after(0, self._do_show_typing) def _do_show_typing(self) -> None: """Affiche l'indicateur dans le thread tkinter.""" tk = self._tk if self._msg_frame is None: return self._typing_frame = tk.Frame(self._msg_frame, bg=BG_COLOR) self._typing_frame.pack(fill=tk.X, padx=MARGIN, pady=4) inner = tk.Frame( self._typing_frame, bg=MSG_LEA_BG, padx=12, pady=8, highlightbackground=MSG_BORDER_COLOR, highlightthickness=1, ) inner.pack(anchor=tk.W, padx=(0, 50)) tk.Label( inner, text="L\u00e9a r\u00e9fl\u00e9chit...", bg=MSG_LEA_BG, fg=TIMESTAMP_FG, font=FONT_MSG_ITALIC, ).pack(anchor=tk.W) self._scroll_to_bottom() def _remove_typing_indicator(self) -> None: """Retire l'indicateur de frappe.""" if self._root is None: return self._root.after(0, self._do_remove_typing) def _do_remove_typing(self) -> None: """Retire l'indicateur dans le thread tkinter.""" if hasattr(self, "_typing_frame") and self._typing_frame is not None: try: self._typing_frame.destroy() except Exception: pass self._typing_frame = None # ====================================================================== # Envoi de messages # ====================================================================== def _on_send(self) -> None: """Envoie le message saisi par l'utilisateur.""" text = self._input_var.get().strip() if not text or text == "Votre message...": return # Vider le champ self._input_var.set("") self._input_entry.configure(fg=INPUT_FG) # Afficher le message utilisateur self._add_user_message(text) # Envoyer au serveur dans un thread threading.Thread( target=self._send_to_server, args=(text,), daemon=True, ).start() def _send_to_server(self, message: str) -> None: """Envoie le message au serveur et affiche la reponse (thread reseau).""" self._show_typing_indicator() try: if self._server_client is None: self._remove_typing_indicator() self._add_lea_message( "Je ne suis pas encore connect\u00e9e au serveur.\n" "V\u00e9rifiez la connexion r\u00e9seau." ) return response = self._server_client.send_chat_message(message) self._remove_typing_indicator() if response is None: self._add_lea_message( "\u26a0 Connexion perdue. Impossible de joindre le serveur." ) return # Extraire le message de la reponse resp = response.get("response", {}) if isinstance(resp, dict): text = resp.get("message", str(resp)) else: text = str(resp) if text: self._add_lea_message(text) # Traiter les suggestions suggestions = [] if isinstance(resp, dict): suggestions = resp.get("suggestions", []) if suggestions: suggestions_text = "\n".join( f"\u2022 {s}" for s in suggestions ) self._add_system_message(f"Suggestions :\n{suggestions_text}") except Exception as e: self._remove_typing_indicator() logger.error("Erreur envoi message : %s", e) self._add_lea_message( f"\u26a0 Erreur : {e}" ) # ====================================================================== # Actions rapides # ====================================================================== def _on_quick_record(self) -> None: """Bouton Apprenez-moi — lance une session d'enregistrement.""" # Verifier si un enregistrement est deja en cours if self._shared_state is not None and self._shared_state.is_recording: self._add_lea_message( "Un enregistrement est d\u00e9j\u00e0 en cours.\n" "Cliquez \u00ab Arr\u00eater \u00bb pour terminer d'abord." ) return can_record = ( self._shared_state is not None or self._on_start_callback is not None ) if can_record: self._add_system_message("Pr\u00e9paration de l'apprentissage...") # Demander le nom dans un thread (dialogue tkinter) threading.Thread(target=self._do_quick_record, daemon=True).start() else: self._add_lea_message( "L'apprentissage n'est pas disponible pour le moment. " "Essayez depuis le menu de la barre des t\u00e2ches." ) def _do_quick_record(self) -> None: """Demande le consentement puis le nom de la tache et lance l'enregistrement. Notification prealable obligatoire (Articles 13/14, Reglement IA) : l'utilisateur doit etre informe de ce qui sera capture AVANT le demarrage. """ import tkinter as tk from tkinter import simpledialog, messagebox # --- Consentement prealable (Articles 13/14, Reglement IA) --- consent_root = tk.Tk() consent_root.withdraw() consent_root.attributes('-topmost', True) consent = messagebox.askyesno( "Enregistrement — Information", "\u26a0\ufe0f L'enregistrement va capturer votre \u00e9cran, " "vos clics et vos frappes clavier pour apprendre cette t\u00e2che.\n\n" "Les donn\u00e9es sensibles seront automatiquement flout\u00e9es.\n\n" "Voulez-vous continuer ?", parent=consent_root, ) consent_root.destroy() if not consent: self._add_lea_message("Enregistrement annul\u00e9.") return # --- Dialogue de saisie du nom --- tmp_root = tk.Tk() tmp_root.withdraw() tmp_root.attributes('-topmost', True) name = simpledialog.askstring( "Nouvelle t\u00e2che", "D\u00e9crivez cette t\u00e2che en quelques mots :", initialvalue="", parent=tmp_root, ) tmp_root.destroy() if name and name.strip(): name = name.strip() self._add_lea_message( f"C'est parti ! Montrez-moi comment faire \u00ab {name} \u00bb." ) try: # Utiliser l'etat partage si disponible (synchronise le systray) if self._shared_state is not None: self._shared_state.start_recording(name) elif self._on_start_callback is not None: self._on_start_callback(name) except Exception as e: self._add_lea_message(f"Oups, un probl\u00e8me : {e}") def _on_quick_tasks(self) -> None: """Bouton Lancer — demande ce que L\u00e9a sait faire.""" self._add_user_message("Qu'est-ce que vous savez faire ?") threading.Thread( target=self._send_to_server, args=("qu'est-ce que tu sais faire ?",), daemon=True, ).start() def _on_quick_import(self) -> None: """Bouton Donn\u00e9es — affiche les tables / imports.""" self._add_user_message("Montrez-moi les donn\u00e9es") threading.Thread( target=self._send_to_server, args=("montre les tables",), daemon=True, ).start() def _on_quick_stop(self) -> None: """Bouton Arr\u00eater — arr\u00eate l'enregistrement ou le replay en cours. Verifie l'etat partage et agit en consequence : 1. Si enregistrement en cours → arrete via shared_state 2. Si replay en cours → arrete via shared_state 3. Sinon → informe l'utilisateur qu'il n'y a rien a arreter """ if self._shared_state is not None: if self._shared_state.is_recording: name = self._shared_state.recording_name count = self._shared_state.actions_count self._shared_state.stop_recording() self._add_lea_message( f"Enregistrement arr\u00eat\u00e9.\n" f"J'ai m\u00e9moris\u00e9 {count} actions pour " f"\u00ab {name} \u00bb. C'est not\u00e9 !" ) return if self._shared_state.is_replay_active: self._shared_state.set_replay_active(False) self._add_lea_message("Replay arr\u00eat\u00e9.") return # Rien a arreter self._add_lea_message( "Il n'y a rien en cours \u00e0 arr\u00eater.\n" "Cliquez \u00ab Apprenez-moi \u00bb pour d\u00e9marrer." ) def _on_quick_help(self) -> None: """Bouton Aide — demande l'aide.""" self._add_user_message("Aide") threading.Thread( target=self._send_to_server, args=("aide",), daemon=True, ).start() # ====================================================================== # Envoi de fichier # ====================================================================== def _on_attach_file(self) -> None: """Ouvre un dialogue pour selectionner un fichier a envoyer.""" from tkinter import filedialog filepath = filedialog.askopenfilename( title="Choisir un fichier", filetypes=[ ("Tableurs", "*.xlsx *.xls *.csv"), ("PDF", "*.pdf"), ("Tous", "*.*"), ], parent=self._root, ) if filepath: self._add_user_message(f"\U0001f4ce {os.path.basename(filepath)}") threading.Thread( target=self._upload_file, args=(filepath,), daemon=True, ).start() def _upload_file(self, filepath: str) -> None: """Envoie le fichier au serveur et affiche la reponse (thread reseau).""" self._show_typing_indicator() try: if self._server_client is None: self._remove_typing_indicator() self._add_lea_message( "Je ne suis pas encore connect\u00e9e au serveur.\n" "V\u00e9rifiez la connexion r\u00e9seau." ) return import requests # Determiner l'URL de base du serveur chat host = self._server_host port = self._chat_port url = f"http://{host}:{port}/api/chat/upload" filename = os.path.basename(filepath) with open(filepath, "rb") as f: files = {"file": (filename, f)} resp = requests.post(url, files=files, timeout=60) self._remove_typing_indicator() if resp.ok: data = resp.json() # Construire un message de reponse lisible msg_parts = [] if data.get("filename"): msg_parts.append(f"Fichier **{data['filename']}** re\u00e7u !") preview = data.get("preview", {}) if preview: rows = preview.get("total_rows", "?") cols = preview.get("columns", []) cols_str = ", ".join(cols[:8]) if len(cols) > 8: cols_str += f"... (+{len(cols) - 8})" msg_parts.append(f"{rows} lignes, colonnes : {cols_str}") table_name = data.get("table_name", "") if table_name: msg_parts.append( f"Je cr\u00e9e la table '{table_name}' ?" ) if data.get("message"): msg_parts.append(str(data["message"])) message = "\n".join(msg_parts) if msg_parts else "Fichier envoy\u00e9 !" self._add_lea_message(message) else: error = "Erreur serveur" try: error = resp.json().get("error", error) except Exception: pass self._add_lea_message(f"\u26a0 {error}") except Exception as e: self._remove_typing_indicator() logger.error("Erreur upload fichier : %s", e) self._add_lea_message(f"\u26a0 Erreur d'envoi : {e}") # ====================================================================== # Verification de la connexion # ====================================================================== def _check_connection_status(self) -> None: """Verifie l'etat de connexion et met a jour l'indicateur.""" if self._destroyed or self._root is None: return try: connected = False if self._server_client is not None: connected = getattr(self._server_client, "connected", False) if connected: self._status_dot.configure(fg=STATUS_CONNECTED) self._status_label.configure(text="Connect\u00e9e") else: self._status_dot.configure(fg=STATUS_DISCONNECTED) self._status_label.configure(text="D\u00e9connect\u00e9e") except Exception: pass # Reverifier toutes les 10 secondes if not self._destroyed and self._root is not None: try: self._root.after(10000, self._check_connection_status) except Exception: pass