# tray_ui.py """ Interface tray (icône de zone de notification) pour agent_v0. Menu : - Start/Stop session (clic gauche ou entrée par défaut) - Start session - Stop session - Open sessions folder - Open logs - Quit Gère : - création / arrêt de RawSession - démarrage / arrêt des capteurs souris & clavier - capture des screenshots (full ou crop) - hover (infobulles) via hover_idle - scroll (molette) - sauvegarde JSON + ZIP + upload - copie optionnelle vers un chemin réseau """ from __future__ import annotations import logging import os import platform import socket import sys import threading import shutil from typing import Optional, List import pystray from pystray import Menu, MenuItem from PIL import Image, ImageDraw from config import SESSIONS_ROOT, EXIT_AFTER_SESSION, LOGS_DIR from raw_session import RawSession from screen_capturer import ScreenCapturer from event_captor import EventCaptor from key_captor import KeyCaptor from window_info import get_active_window_info from storage import create_session_zip from uploader import upload_session_zip from user_config import load_user_config logger = logging.getLogger("agent_v0.tray") def _open_in_file_manager(path: str) -> None: """Ouvre un dossier dans le gestionnaire de fichiers en fonction de l'OS.""" try: if sys.platform.startswith("linux"): os.makedirs(path, exist_ok=True) os.system(f'xdg-open "{path}" &') elif sys.platform.startswith("win"): os.makedirs(path, exist_ok=True) os.startfile(path) # type: ignore[attr-defined] elif sys.platform == "darwin": os.makedirs(path, exist_ok=True) os.system(f'open "{path}" &') else: logger.warning("Plateforme non supportée pour l'ouverture de dossier: %s", sys.platform) except Exception as e: logger.exception("Erreur lors de l'ouverture du dossier %s : %s", path, e) class TrayApp: """ Application basée sur une icône de tray. """ def __init__(self) -> None: self._lock = threading.Lock() self.session: Optional[RawSession] = None self.capturer: Optional[ScreenCapturer] = None self.event_captor: Optional[EventCaptor] = None self.key_captor: Optional[KeyCaptor] = None self.session_active: bool = False # Config utilisateur self.config = load_user_config() logger.info("Config chargée : %s", self.config) self.icon = pystray.Icon( "agent_v0", self._create_icon_image(active=False), "agent_v0", self._create_menu(), ) # --- Icône & menu --- def _create_icon_image(self, active: bool) -> Image.Image: size = 64 img = Image.new("RGBA", (size, size), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) color = (0, 200, 0, 255) if active else (120, 120, 120, 255) radius = 24 center = (size // 2, size // 2) bbox = [ center[0] - radius, center[1] - radius, center[0] + radius, center[1] + radius, ] draw.ellipse(bbox, fill=color) draw.ellipse(bbox, outline=(0, 0, 0, 255), width=2) return img def _create_menu(self) -> Menu: """ Menu tray. - "Start/Stop session" est l'action par défaut (clic gauche). """ return Menu( MenuItem("Start/Stop session", self.on_toggle_session, default=True), MenuItem("Start session", self.on_start_session), MenuItem("Stop session", self.on_stop_session), MenuItem("Open sessions folder", self.on_open_sessions), MenuItem("Open logs", self.on_open_logs), MenuItem("Quit", self.on_quit), ) def _set_icon_active(self, active: bool) -> None: self.icon.icon = self._create_icon_image(active) # --- Lifecycle --- def run(self) -> None: logger.info("Démarrage de TrayApp...") self.icon.run() # --- Actions menu --- def on_toggle_session(self, icon: pystray.Icon, item: MenuItem) -> None: with self._lock: if self.session_active: logger.info("Toggle: arrêt de la session.") self._stop_session_internal() else: logger.info("Toggle: démarrage d'une session.") self._start_session_internal() def on_start_session(self, icon: pystray.Icon, item: MenuItem) -> None: with self._lock: if self.session_active: logger.info("Session déjà active, Start ignoré.") return logger.info("Démarrage d'une nouvelle session (menu Start)...") self._start_session_internal() def on_stop_session(self, icon: pystray.Icon, item: MenuItem) -> None: with self._lock: if not self.session_active: logger.info("Aucune session active, Stop ignoré.") return logger.info("Arrêt de la session (menu Stop)...") self._stop_session_internal() def on_open_sessions(self, icon: pystray.Icon, item: MenuItem) -> None: logger.info("Ouverture du dossier sessions...") _open_in_file_manager(os.path.abspath(SESSIONS_ROOT)) def on_open_logs(self, icon: pystray.Icon, item: MenuItem) -> None: logger.info("Ouverture du dossier logs...") _open_in_file_manager(str(LOGS_DIR)) def on_quit(self, icon: pystray.Icon, item: MenuItem) -> None: logger.info("Quit demandé depuis le menu tray.") with self._lock: if self.session_active: logger.info("Arrêt de la session avant de quitter...") self._stop_session_internal() self.icon.stop() # --- Gestion de la session --- def _start_session_internal(self) -> None: cfg = self.config # Infos de base sur la machine plat = platform.system().lower() hostname = socket.gethostname() # Détection de la résolution de l'écran principal width, height = ScreenCapturer.detect_primary_resolution() screen_resolution = [width, height] display_scale = 1.0 # TODO: affiner si besoin logger.info( "Nouvelle session : platform=%s, hostname=%s, res=%dx%d, mode=%s", plat, hostname, screen_resolution[0], screen_resolution[1], cfg.get("mode", "enriched"), ) session = RawSession.create( user_id=cfg.get("user_id", "demo_user"), user_label=cfg.get("user_label", cfg.get("user_id", "demo_user")), customer=cfg.get("customer", ""), training_label=cfg.get("training_label", ""), notes=cfg.get("notes", ""), platform=plat, hostname=hostname, screen_resolution=screen_resolution, display_scale=display_scale, ) screenshot_mode = cfg.get("screenshot_mode", "crop") crop_w = int(cfg.get("screenshot_crop_width", 900)) crop_h = int(cfg.get("screenshot_crop_height", 700)) self.session = session self.capturer = ScreenCapturer( session=session, screenshot_mode=screenshot_mode, crop_width=crop_w, crop_height=crop_h, ) capture_hover = bool(cfg.get("capture_hover", True)) capture_scroll = bool(cfg.get("capture_scroll", True)) hover_min_idle_ms = int(cfg.get("hover_min_idle_ms", 700)) # Callbacks def on_mouse_click(button_str: str, x: int, y: int) -> None: self._record_click_with_screenshot(button_str, x, y) def on_key_combo(keys: List[str]) -> None: self._record_key_combo_with_screenshot(keys) def on_scroll(x: int, y: int, dx: int, dy: int) -> None: self._record_scroll_with_screenshot(x, y, dx, dy) def on_hover(x: int, y: int, idle_ms: int) -> None: self._record_hover_with_screenshot(x, y, idle_ms) # Capteurs self.event_captor = EventCaptor( on_mouse_click=on_mouse_click, on_scroll=on_scroll if capture_scroll else None, on_hover=on_hover if capture_hover else None, hover_min_idle_ms=hover_min_idle_ms, ) self.key_captor = KeyCaptor(on_key_combo=on_key_combo) self.event_captor.start() self.key_captor.start() self.session_active = True self._set_icon_active(True) logger.info("Session %s démarrée.", session.session_id) def _stop_session_internal(self) -> None: if not self.session_active or self.session is None: return # Stop capteurs if self.event_captor is not None: self.event_captor.stop() self.event_captor = None if self.key_captor is not None: self.key_captor.stop() self.key_captor = None session = self.session self.session = None self.capturer = None self.session_active = False self._set_icon_active(False) session.close() # Sauvegarde JSON json_path = "" try: json_path = session.save_json() logger.info("Session JSON sauvegardée : %s", json_path) except Exception as e: logger.exception("Erreur lors de la sauvegarde JSON : %s", e) # Création ZIP zip_path = "" try: zip_path = create_session_zip(session) logger.info("ZIP de session créé : %s", zip_path) except Exception as e: logger.exception("Erreur lors de la création du ZIP : %s", e) # Upload if zip_path: try: ok = upload_session_zip(zip_path, session.session_id) if ok: logger.info("Upload du ZIP réussi.") else: logger.warning("Upload du ZIP échoué ou désactivé.") except Exception as e: logger.exception("Erreur lors de l'upload du ZIP : %s", e) # Copie réseau si configurée net_path = self.config.get("network_save_path") or "" if net_path and zip_path: try: dest_dir = os.path.join(net_path, session.session_id) os.makedirs(dest_dir, exist_ok=True) if json_path and os.path.isfile(json_path): shutil.copy2(json_path, os.path.join(dest_dir, os.path.basename(json_path))) shutil.copy2(zip_path, os.path.join(dest_dir, os.path.basename(zip_path))) logger.info("Session copiée sur le chemin réseau : %s", dest_dir) except Exception as e: logger.exception("Erreur lors de la copie vers le chemin réseau : %s", e) logger.info("Session terminée. SESSIONS_ROOT = %s", os.path.abspath(SESSIONS_ROOT)) if EXIT_AFTER_SESSION: logger.info("EXIT_AFTER_SESSION=True → arrêt de l'application.") self.icon.stop() # --- Enregistrement des événements --- def _record_click_with_screenshot(self, button_str: str, x: int, y: int) -> None: if not (self.session_active and self.session and self.capturer): logger.debug("Clic ignoré (pas de session active).") return screenshot_id, _ = self.capturer.capture((x, y)) win_info = get_active_window_info() window_title = win_info["title"] app_name = win_info["app_name"] self.session.add_mouse_click_event( button=button_str, pos=[x, y], window_title=window_title, app_name=app_name, screenshot_id=screenshot_id, ) logger.info( "Clic %s à (%d, %d) → screenshot=%s, window='%s', app='%s'", button_str, x, y, screenshot_id, window_title, app_name, ) def _record_key_combo_with_screenshot(self, keys: List[str]) -> None: if not (self.session_active and self.session and self.capturer): logger.debug("Combo clavier ignoré (pas de session active).") return # Pour les combos clavier, on ne sait pas toujours où est la souris : # on peut donc capturer en mode crop autours de la dernière position # connue OU plein écran si screenshot_mode = "full". screenshot_id, _ = self.capturer.capture(None) win_info = get_active_window_info() window_title = win_info["title"] app_name = win_info["app_name"] self.session.add_key_combo_event( keys=keys, window_title=window_title, app_name=app_name, screenshot_id=screenshot_id, ) logger.info( "Combo clavier %s → screenshot=%s, window='%s', app='%s'", keys, screenshot_id, window_title, app_name, ) def _record_scroll_with_screenshot(self, x: int, y: int, dx: int, dy: int) -> None: if not (self.session_active and self.session and self.capturer): logger.debug("Scroll ignoré (pas de session active).") return screenshot_id, _ = self.capturer.capture((x, y)) win_info = get_active_window_info() window_title = win_info["title"] app_name = win_info["app_name"] self.session.add_scroll_event( pos=[x, y], delta=[dx, dy], window_title=window_title, app_name=app_name, screenshot_id=screenshot_id, ) logger.info( "Scroll dx=%d dy=%d à (%d, %d) → screenshot=%s, window='%s', app='%s'", dx, dy, x, y, screenshot_id, window_title, app_name, ) def _record_hover_with_screenshot(self, x: int, y: int, idle_ms: int) -> None: if not (self.session_active and self.session and self.capturer): logger.debug("Hover ignoré (pas de session active).") return screenshot_id, _ = self.capturer.capture((x, y)) win_info = get_active_window_info() window_title = win_info["title"] app_name = win_info["app_name"] self.session.add_hover_idle_event( pos=[x, y], idle_ms=idle_ms, window_title=window_title, app_name=app_name, screenshot_id=screenshot_id, ) logger.info( "Hover à (%d, %d) idle=%dms → screenshot=%s, window='%s', app='%s'", x, y, idle_ms, screenshot_id, window_title, app_name, )