""" RawSession - Couche 0 : Capture Brute Enregistre fidèlement toutes les interactions utilisateur avec horodatage précis et contexte complet. C'est la fondation du système RPA Vision V3. """ from dataclasses import dataclass, field from datetime import datetime from typing import Dict, List, Optional, Any from pathlib import Path import json @dataclass class RawWindowContext: """ Contexte de fenêtre pour un événement (RawSession) Renommé de WindowContext pour éviter collision avec ScreenState.WindowContext Auteur: Dom, Alice Kiro - 15 décembre 2024 """ title: str app_name: str def to_dict(self) -> Dict[str, str]: return { "title": self.title, "app_name": self.app_name } @classmethod def from_dict(cls, data: Dict[str, str]) -> 'RawWindowContext': return cls( title=data["title"], app_name=data["app_name"] ) # Alias de compatibilité pour migration douce WindowContext = RawWindowContext @dataclass class Event: """ Événement utilisateur capturé Types supportés: - mouse_click, mouse_move, mouse_scroll - key_press, key_release, text_input - window_change, screen_change """ t: float # Timestamp relatif en secondes depuis début session type: str # Type d'événement window: RawWindowContext screenshot_id: Optional[str] = None data: Dict[str, Any] = field(default_factory=dict) # Données spécifiques au type def to_dict(self) -> Dict[str, Any]: result = { "t": self.t, "type": self.type, "window": self.window.to_dict(), } if self.screenshot_id: result["screenshot_id"] = self.screenshot_id # Ajouter les données spécifiques result.update(self.data) return result @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Event': # Extraire les champs de base t = data["t"] event_type = data["type"] window = RawWindowContext.from_dict(data["window"]) screenshot_id = data.get("screenshot_id") # Le reste va dans data event_data = {k: v for k, v in data.items() if k not in ["t", "type", "window", "screenshot_id"]} return cls( t=t, type=event_type, window=window, screenshot_id=screenshot_id, data=event_data ) @dataclass class Screenshot: """Référence à un screenshot capturé""" screenshot_id: str relative_path: str captured_at: str # ISO format timestamp def to_dict(self) -> Dict[str, str]: return { "screenshot_id": self.screenshot_id, "relative_path": self.relative_path, "captured_at": self.captured_at } @classmethod def from_dict(cls, data: Dict[str, str]) -> 'Screenshot': return cls( screenshot_id=data["screenshot_id"], relative_path=data["relative_path"], captured_at=data["captured_at"] ) @dataclass class RawSession: """ Session brute capturant tous les événements utilisateur Format: rawsession_v1 """ session_id: str agent_version: str environment: Dict[str, Any] user: Dict[str, str] context: Dict[str, str] started_at: datetime ended_at: Optional[datetime] = None events: List[Event] = field(default_factory=list) screenshots: List[Screenshot] = field(default_factory=list) schema_version: str = "rawsession_v1" def add_event(self, event: Event) -> None: """Ajouter un événement à la session""" self.events.append(event) def add_screenshot(self, screenshot: Screenshot) -> None: """Ajouter un screenshot à la session""" self.screenshots.append(screenshot) def to_json(self) -> Dict[str, Any]: """Sérialiser en JSON""" return { "schema_version": self.schema_version, "session_id": self.session_id, "agent_version": self.agent_version, "environment": self.environment, "user": self.user, "context": self.context, "started_at": self.started_at.isoformat(), "ended_at": self.ended_at.isoformat() if self.ended_at else None, "events": [event.to_dict() for event in self.events], "screenshots": [screenshot.to_dict() for screenshot in self.screenshots] } @classmethod def from_json(cls, data: Dict[str, Any]) -> 'RawSession': """Désérialiser depuis JSON""" # Valider schéma schema_version = data.get("schema_version") if schema_version != "rawsession_v1": raise ValueError( f"Unsupported schema version: {schema_version}. " f"Expected: rawsession_v1" ) # Parser dates started_at = datetime.fromisoformat(data["started_at"]) ended_at = datetime.fromisoformat(data["ended_at"]) if data.get("ended_at") else None # Parser events et screenshots events = [Event.from_dict(e) for e in data.get("events", [])] screenshots = [Screenshot.from_dict(s) for s in data.get("screenshots", [])] return cls( schema_version=schema_version, session_id=data["session_id"], agent_version=data["agent_version"], environment=data["environment"], user=data["user"], context=data["context"], started_at=started_at, ended_at=ended_at, events=events, screenshots=screenshots ) def save_to_file(self, filepath: Path) -> None: """Sauvegarder dans un fichier JSON""" filepath.parent.mkdir(parents=True, exist_ok=True) with open(filepath, 'w', encoding='utf-8') as f: json.dump(self.to_json(), f, indent=2, ensure_ascii=False) @classmethod def load_from_file(cls, filepath: Path) -> 'RawSession': """Charger depuis un fichier JSON""" with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) return cls.from_json(data)