""" Contrat de Données VWB - Evidence d'Exécution Auteur : Dom, Alice, Kiro - 09 janvier 2026 Ce module définit les contrats pour les preuves d'exécution (Evidence) des actions VisionOnly dans le Visual Workflow Builder. Classes : - VWBEvidenceType : Types d'evidence possibles - VWBEvidence : Contrat principal pour les preuves d'exécution """ from enum import Enum from dataclasses import dataclass, asdict from typing import Dict, Any, Optional, List from datetime import datetime import json import base64 from pathlib import Path class VWBEvidenceType(Enum): """Types d'evidence possibles dans le VWB.""" # Evidence visuelles SCREENSHOT_BEFORE = "screenshot_before" # Capture avant action SCREENSHOT_AFTER = "screenshot_after" # Capture après action SCREENSHOT_ERROR = "screenshot_error" # Capture lors d'erreur ELEMENT_HIGHLIGHT = "element_highlight" # Élément surligné # Evidence d'interaction CLICK_EVIDENCE = "click_evidence" # Preuve de clic TYPE_EVIDENCE = "type_evidence" # Preuve de saisie WAIT_EVIDENCE = "wait_evidence" # Preuve d'attente # Evidence de validation VALIDATION_SUCCESS = "validation_success" # Validation réussie VALIDATION_FAILURE = "validation_failure" # Validation échouée # Evidence système SYSTEM_STATE = "system_state" # État système PERFORMANCE_METRICS = "performance_metrics" # Métriques de performance # Evidence de débogage DEBUG_INFO = "debug_info" # Informations de débogage LOG_ENTRY = "log_entry" # Entrée de log @dataclass class VWBEvidence: """ Contrat de données pour les preuves d'exécution VWB. Cette classe encapsule toutes les informations nécessaires pour documenter et tracer l'exécution des actions VisionOnly dans le Visual Workflow Builder. """ # Identification de l'evidence evidence_id: str evidence_type: VWBEvidenceType # Contexte d'exécution action_id: str step_id: str # Informations temporelles timestamp: datetime # Contenu de l'evidence title: str description: str # Données structurées data: Dict[str, Any] # Liens vers autres evidence related_evidence_ids: List[str] # Paramètres optionnels avec valeurs par défaut workflow_id: Optional[str] = None execution_time_ms: Optional[float] = None screenshot_base64: Optional[str] = None screenshot_width: Optional[int] = None screenshot_height: Optional[int] = None highlight_box: Optional[Dict[str, int]] = None # {x, y, width, height} success: bool = True confidence_score: Optional[float] = None user_id: Optional[str] = None session_id: Optional[str] = None parent_evidence_id: Optional[str] = None def __post_init__(self): """Validation et initialisation post-création.""" if not self.evidence_id: self.evidence_id = f"ev_{self.action_id}_{int(self.timestamp.timestamp())}" if not self.data: self.data = {} if not self.related_evidence_ids: self.related_evidence_ids = [] # Validation des dimensions screenshot if self.screenshot_base64: if not self.screenshot_width or not self.screenshot_height: self._extract_screenshot_dimensions() def _extract_screenshot_dimensions(self): """Extrait les dimensions du screenshot depuis les données base64.""" try: if self.screenshot_base64: from PIL import Image import io # Décoder le base64 image_data = base64.b64decode(self.screenshot_base64) image = Image.open(io.BytesIO(image_data)) self.screenshot_width = image.width self.screenshot_height = image.height except Exception: # En cas d'erreur, utiliser des valeurs par défaut self.screenshot_width = 1920 self.screenshot_height = 1080 def to_dict(self) -> Dict[str, Any]: """Convertit l'evidence en dictionnaire pour sérialisation JSON.""" data = asdict(self) # Convertir l'enum en string data['evidence_type'] = self.evidence_type.value # Convertir le timestamp en ISO string data['timestamp'] = self.timestamp.isoformat() return data @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'VWBEvidence': """Crée une instance depuis un dictionnaire.""" # Convertir le string en enum data['evidence_type'] = VWBEvidenceType(data['evidence_type']) # Convertir le timestamp if isinstance(data['timestamp'], str): data['timestamp'] = datetime.fromisoformat(data['timestamp']) return cls(**data) def to_json(self) -> str: """Sérialise l'evidence en JSON.""" return json.dumps(self.to_dict(), ensure_ascii=False, indent=2) @classmethod def from_json(cls, json_str: str) -> 'VWBEvidence': """Désérialise depuis JSON.""" data = json.loads(json_str) return cls.from_dict(data) def has_screenshot(self) -> bool: """Vérifie si l'evidence contient un screenshot.""" return self.screenshot_base64 is not None and len(self.screenshot_base64) > 0 def has_highlight(self) -> bool: """Vérifie si l'evidence contient une zone surlignée.""" return ( self.highlight_box is not None and all(key in self.highlight_box for key in ['x', 'y', 'width', 'height']) ) def get_screenshot_data_url(self) -> Optional[str]: """Retourne l'URL data du screenshot pour affichage web.""" if not self.has_screenshot(): return None # Déterminer le format (PNG par défaut) format_prefix = "data:image/png;base64," if self.screenshot_base64.startswith('/9j/'): # JPEG magic bytes en base64 format_prefix = "data:image/jpeg;base64," return f"{format_prefix}{self.screenshot_base64}" def save_screenshot(self, file_path: str) -> bool: """Sauvegarde le screenshot sur disque.""" if not self.has_screenshot(): return False try: image_data = base64.b64decode(self.screenshot_base64) with open(file_path, 'wb') as f: f.write(image_data) return True except Exception: return False def add_data(self, key: str, value: Any) -> None: """Ajoute une donnée à l'evidence.""" self.data[key] = value def get_data(self, key: str, default: Any = None) -> Any: """Récupère une donnée de l'evidence.""" return self.data.get(key, default) def set_highlight_box(self, x: int, y: int, width: int, height: int) -> None: """Définit la zone de surbrillance.""" self.highlight_box = { 'x': max(0, x), 'y': max(0, y), 'width': max(1, width), 'height': max(1, height) } def add_related_evidence(self, evidence_id: str) -> None: """Ajoute une evidence liée.""" if evidence_id not in self.related_evidence_ids: self.related_evidence_ids.append(evidence_id) def get_file_size_mb(self) -> float: """Calcule la taille approximative en MB.""" size_bytes = 0 # Taille du screenshot if self.screenshot_base64: size_bytes += len(self.screenshot_base64.encode('utf-8')) # Taille des données JSON size_bytes += len(json.dumps(self.data).encode('utf-8')) return size_bytes / (1024 * 1024) def is_visual_evidence(self) -> bool: """Vérifie si c'est une evidence visuelle.""" visual_types = { VWBEvidenceType.SCREENSHOT_BEFORE, VWBEvidenceType.SCREENSHOT_AFTER, VWBEvidenceType.SCREENSHOT_ERROR, VWBEvidenceType.ELEMENT_HIGHLIGHT } return self.evidence_type in visual_types def is_interaction_evidence(self) -> bool: """Vérifie si c'est une evidence d'interaction.""" interaction_types = { VWBEvidenceType.CLICK_EVIDENCE, VWBEvidenceType.TYPE_EVIDENCE, VWBEvidenceType.WAIT_EVIDENCE } return self.evidence_type in interaction_types def get_summary(self) -> Dict[str, Any]: """Retourne un résumé de l'evidence pour affichage.""" return { 'evidence_id': self.evidence_id, 'type': self.evidence_type.value, 'title': self.title, 'timestamp': self.timestamp.isoformat(), 'success': self.success, 'has_screenshot': self.has_screenshot(), 'has_highlight': self.has_highlight(), 'file_size_mb': round(self.get_file_size_mb(), 2), 'confidence_score': self.confidence_score } def __str__(self) -> str: """Représentation string de l'evidence.""" status = "✅" if self.success else "❌" return f"{status} VWBEvidence({self.evidence_type.value}): {self.title}" def __repr__(self) -> str: """Représentation détaillée de l'evidence.""" return ( f"VWBEvidence(" f"evidence_id='{self.evidence_id}', " f"evidence_type={self.evidence_type.value}, " f"action_id='{self.action_id}', " f"success={self.success}, " f"has_screenshot={self.has_screenshot()}" f")" ) def create_screenshot_evidence( action_id: str, step_id: str, screenshot_base64: str, evidence_type: VWBEvidenceType = VWBEvidenceType.SCREENSHOT_BEFORE, title: str = "Capture d'écran", **kwargs ) -> VWBEvidence: """ Fonction utilitaire pour créer rapidement une evidence de screenshot. Args: action_id: ID de l'action step_id: ID de l'étape screenshot_base64: Screenshot en base64 evidence_type: Type d'evidence title: Titre de l'evidence **kwargs: Paramètres additionnels Returns: Instance de VWBEvidence """ return VWBEvidence( evidence_id=kwargs.get('evidence_id', ''), evidence_type=evidence_type, action_id=action_id, step_id=step_id, workflow_id=kwargs.get('workflow_id'), timestamp=kwargs.get('timestamp', datetime.now()), execution_time_ms=kwargs.get('execution_time_ms'), title=title, description=kwargs.get('description', title), screenshot_base64=screenshot_base64, screenshot_width=kwargs.get('screenshot_width'), screenshot_height=kwargs.get('screenshot_height'), highlight_box=kwargs.get('highlight_box'), data=kwargs.get('data', {}), success=kwargs.get('success', True), confidence_score=kwargs.get('confidence_score'), user_id=kwargs.get('user_id'), session_id=kwargs.get('session_id'), related_evidence_ids=kwargs.get('related_evidence_ids', []), parent_evidence_id=kwargs.get('parent_evidence_id') ) def create_interaction_evidence( action_id: str, step_id: str, evidence_type: VWBEvidenceType, title: str, interaction_data: Dict[str, Any], **kwargs ) -> VWBEvidence: """ Fonction utilitaire pour créer une evidence d'interaction. Args: action_id: ID de l'action step_id: ID de l'étape evidence_type: Type d'evidence d'interaction title: Titre de l'evidence interaction_data: Données de l'interaction **kwargs: Paramètres additionnels Returns: Instance de VWBEvidence """ return VWBEvidence( evidence_id=kwargs.get('evidence_id', ''), evidence_type=evidence_type, action_id=action_id, step_id=step_id, workflow_id=kwargs.get('workflow_id'), timestamp=kwargs.get('timestamp', datetime.now()), execution_time_ms=kwargs.get('execution_time_ms'), title=title, description=kwargs.get('description', title), screenshot_base64=kwargs.get('screenshot_base64'), screenshot_width=kwargs.get('screenshot_width'), screenshot_height=kwargs.get('screenshot_height'), highlight_box=kwargs.get('highlight_box'), data=interaction_data, success=kwargs.get('success', True), confidence_score=kwargs.get('confidence_score'), user_id=kwargs.get('user_id'), session_id=kwargs.get('session_id'), related_evidence_ids=kwargs.get('related_evidence_ids', []), parent_evidence_id=kwargs.get('parent_evidence_id') )