rpa_vision_v3/agent_v0/agent_v1/ui/paused_toast.py

# agent_v1/ui/paused_toast.py
"""
Toast Tkinter custom pour la pause supervisée (« Léa a besoin de votre aide »).

Démo GHT 8 mai 2026 — Fallback robuste 100 % autonome quand :
  - plyer.notification est silencieux sous Windows 11 (Focus Assist, balloon tips
    bloqués par la stratégie système),
  - la ChatWindow Léa V1 est `withdraw()`-cachée par défaut (Dom ne la voit pas),
  - aucune autre UI ne peut garantir que Dom verra physiquement le message.

Stratégie :
  - Toplevel topmost overrideredirect en haut à droite de l'écran principal,
  - fond bleu Léa, titre + message, auto-close après TOAST_DURATION_S,
  - thread-safe : peut être appelé depuis n'importe quel thread (le polling
    replay tourne dans un daemon thread, pas le thread principal),
  - aucune dépendance externe (juste tkinter stdlib),
  - rate limit interne pour éviter le flood (1 toast / 3s minimum).

Si un Tk root existe déjà dans le process (ChatWindow), on attache le Toplevel
à ce root via `root.after(0, ...)` — c'est l'idiome thread-safe officiel de
tkinter. Sinon on crée un Tk() dédié dans un daemon thread.
"""
from __future__ import annotations

import logging
import threading
import time
from typing import Any, Optional

logger = logging.getLogger(__name__)

# Couleurs cohérentes avec le thème Léa (cf. chat_window.py)
TOAST_BG = "#2563EB"          # Bleu Léa (HEADER_BG)
TOAST_FG = "#FFFFFF"
TOAST_TITLE_BG = "#1E40AF"    # Bleu plus foncé pour le bandeau titre
TOAST_BORDER = "#1E3A8A"

TOAST_WIDTH = 380
TOAST_PAD_X = 18
TOAST_PAD_Y = 14
TOAST_DURATION_MS = 15000
TOAST_RATE_LIMIT_S = 3.0

_lock = threading.Lock()
_last_shown_at: float = 0.0
_last_message: str = ""


def _resolve_existing_root() -> Optional[Any]:
    """Tente de récupérer le Tk root déjà créé par la ChatWindow.

    On évite tk._default_root (deprecated) et on remonte plutôt via les
    threads existants : la ChatWindow garde une référence dans son instance
    mais n'expose rien de global. On se rabat donc sur la création d'un Tk
    indépendant si on n'a rien — c'est sûr, tkinter supporte plusieurs Tk()
    concurrents tant qu'ils sont chacun dans leur propre thread.
    """
    try:
        import tkinter as tk
        # tk._default_root est interne mais c'est le moyen le plus simple
        # de partager un mainloop existant. Si ChatWindow tourne, ce sera
        # son root.
        root = getattr(tk, "_default_root", None)
        if root is not None:
            # Vérifier qu'il est encore vivant
            try:
                root.winfo_exists()
                return root
            except Exception:
                return None
        return None
    except Exception:
        return None


def _build_toast(parent: Any, title: str, message: str) -> Any:
    """Construit le Toplevel toast (appelé dans le thread tkinter)."""
    import tkinter as tk

    top = tk.Toplevel(parent)
    top.withdraw()  # éviter le flash pendant la construction
    top.overrideredirect(True)  # pas de barre de titre
    top.attributes("-topmost", True)
    try:
        # Petit boost de visibilité Windows : alpha légèrement transparent
        top.attributes("-alpha", 0.97)
    except Exception:
        pass

    # Bordure visuelle (cadre extérieur foncé)
    outer = tk.Frame(top, bg=TOAST_BORDER, padx=2, pady=2)
    outer.pack(fill="both", expand=True)

    # Bandeau titre
    title_frame = tk.Frame(outer, bg=TOAST_TITLE_BG)
    title_frame.pack(fill="x")
    tk.Label(
        title_frame,
        text=f"  ⏸  {title}",
        bg=TOAST_TITLE_BG,
        fg=TOAST_FG,
        font=("Segoe UI", 12, "bold"),
        anchor="w",
        padx=10,
        pady=8,
    ).pack(fill="x")

    # Corps du message
    body_frame = tk.Frame(outer, bg=TOAST_BG)
    body_frame.pack(fill="both", expand=True)
    tk.Label(
        body_frame,
        text=message,
        bg=TOAST_BG,
        fg=TOAST_FG,
        font=("Segoe UI", 11),
        wraplength=TOAST_WIDTH - 40,
        justify="left",
        anchor="w",
        padx=TOAST_PAD_X,
        pady=TOAST_PAD_Y,
    ).pack(fill="both", expand=True)

    # Pied de page : "Cliquez pour fermer"
    footer = tk.Label(
        outer,
        text="Cliquez pour fermer",
        bg=TOAST_BG,
        fg="#BFDBFE",
        font=("Segoe UI", 9, "italic"),
        anchor="e",
        padx=10,
        pady=4,
    )
    footer.pack(fill="x", side="bottom")

    # Position : haut-droite de l'écran principal
    top.update_idletasks()
    height = top.winfo_reqheight()
    screen_w = top.winfo_screenwidth()
    x = screen_w - TOAST_WIDTH - 16
    y = 16
    top.geometry(f"{TOAST_WIDTH}x{height}+{x}+{y}")

    # Click anywhere to close
    def _close(_=None):
        try:
            top.destroy()
        except Exception:
            pass

    top.bind("<Button-1>", _close)
    for child in (outer, title_frame, body_frame, footer):
        try:
            child.bind("<Button-1>", _close)
        except Exception:
            pass

    # Afficher + boost focus brut pour passer devant Focus Assist
    top.deiconify()
    top.lift()
    try:
        top.focus_force()
    except Exception:
        pass

    # Re-pin topmost après 100 ms (Windows désactive parfois -topmost
    # quand le focus est pris par une autre app)
    def _repin():
        try:
            top.attributes("-topmost", True)
            top.lift()
        except Exception:
            pass

    try:
        top.after(100, _repin)
        top.after(500, _repin)
        top.after(2000, _repin)
    except Exception:
        pass

    # Auto-close
    try:
        top.after(TOAST_DURATION_MS, _close)
    except Exception:
        pass

    return top


def _show_in_dedicated_thread(title: str, message: str) -> None:
    """Crée un Tk() indépendant dans un daemon thread.

    Utilisé en fallback quand aucun Tk root n'existe. Le thread vit le
    temps du toast (~15s) puis se termine proprement.
    """
    def _run():
        try:
            # DPI awareness (Windows haute résolution)
            try:
                import ctypes
                ctypes.windll.shcore.SetProcessDpiAwareness(1)
            except Exception:
                pass

            import tkinter as tk

            root = tk.Tk()
            root.withdraw()
            try:
                dpi = root.winfo_fpixels("1i")
                root.tk.call("tk", "scaling", dpi / 72.0)
            except Exception:
                pass

            top = _build_toast(root, title, message)

            # Quitter mainloop quand le toast est détruit
            def _watch():
                try:
                    if not top.winfo_exists():
                        root.quit()
                        return
                except Exception:
                    root.quit()
                    return
                root.after(200, _watch)

            root.after(200, _watch)
            root.mainloop()
            try:
                root.destroy()
            except Exception:
                pass
        except Exception:
            logger.debug("paused_toast dedicated thread failed", exc_info=True)

    t = threading.Thread(target=_run, daemon=True, name="paused-toast-tk")
    t.start()


def show_paused_toast(
    title: str = "Léa a besoin de votre aide",
    message: str = "",
) -> bool:
    """Affiche un toast paused topmost.

    Thread-safe, rate-limité, sans dépendance externe. Retourne True si le
    toast a été déclenché, False s'il a été ignoré (rate limit ou erreur).
    """
    global _last_shown_at, _last_message

    if not message:
        message = "Action en attente de votre validation."

    # Rate limit basique : éviter qu'un poll en boucle ouvre 50 toasts
    now = time.monotonic()
    with _lock:
        same_message = (message == _last_message)
        elapsed = now - _last_shown_at
        if same_message and elapsed < TOAST_RATE_LIMIT_S:
            logger.debug(
                "paused_toast rate-limited (%.1fs since last identical)", elapsed
            )
            return False
        _last_shown_at = now
        _last_message = message

    # Tentative 1 : utiliser le Tk root existant (ChatWindow) via after()
    root = _resolve_existing_root()
    if root is not None:
        try:
            root.after(0, lambda: _build_toast(root, title, message))
            logger.info("paused_toast scheduled on existing Tk root")
            return True
        except Exception:
            logger.debug("paused_toast existing-root path failed", exc_info=True)

    # Tentative 2 : créer un Tk() dans un daemon thread
    try:
        _show_in_dedicated_thread(title, message)
        logger.info("paused_toast scheduled in dedicated thread")
        return True
    except Exception:
        logger.error("paused_toast dedicated-thread path failed", exc_info=True)
        return False


__all__ = ["show_paused_toast"]