"""Data models for self-healing workflows.""" from dataclasses import dataclass, field from typing import Optional, Dict, Any from datetime import datetime @dataclass class RecoveryContext: """Context information for recovery attempts.""" original_action: str target_element: str failure_reason: str screenshot_path: str workflow_id: str node_id: str attempt_count: int max_attempts: int = 3 confidence_threshold: float = 0.7 metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'original_action': self.original_action, 'target_element': self.target_element, 'failure_reason': self.failure_reason, 'screenshot_path': self.screenshot_path, 'workflow_id': self.workflow_id, 'node_id': self.node_id, 'attempt_count': self.attempt_count, 'max_attempts': self.max_attempts, 'confidence_threshold': self.confidence_threshold, 'metadata': self.metadata } @dataclass class RecoveryResult: """Result of a recovery attempt.""" success: bool strategy_used: str new_element: Optional[str] = None confidence_score: float = 0.0 execution_time: float = 0.0 learned_pattern: Optional[Dict] = None requires_user_input: bool = False error_message: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'success': self.success, 'strategy_used': self.strategy_used, 'new_element': self.new_element, 'confidence_score': self.confidence_score, 'execution_time': self.execution_time, 'learned_pattern': self.learned_pattern, 'requires_user_input': self.requires_user_input, 'error_message': self.error_message } @dataclass class RecoveryPattern: """A learned recovery pattern.""" pattern_id: str original_failure: str recovery_strategy: str success_count: int failure_count: int confidence_score: float context_metadata: Dict[str, Any] created_at: datetime last_used: datetime @property def success_rate(self) -> float: """Calculate success rate.""" total = self.success_count + self.failure_count return self.success_count / total if total > 0 else 0.0 def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'pattern_id': self.pattern_id, 'original_failure': self.original_failure, 'recovery_strategy': self.recovery_strategy, 'success_count': self.success_count, 'failure_count': self.failure_count, 'confidence_score': self.confidence_score, 'context_metadata': self.context_metadata, 'created_at': self.created_at.isoformat(), 'last_used': self.last_used.isoformat() } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'RecoveryPattern': """Create from dictionary.""" return cls( pattern_id=data['pattern_id'], original_failure=data['original_failure'], recovery_strategy=data['recovery_strategy'], success_count=data['success_count'], failure_count=data['failure_count'], confidence_score=data['confidence_score'], context_metadata=data['context_metadata'], created_at=datetime.fromisoformat(data['created_at']), last_used=datetime.fromisoformat(data['last_used']) ) @dataclass class RecoverySuggestion: """A suggested recovery action.""" strategy: str confidence: float description: str estimated_time: float metadata: Dict[str, Any] = field(default_factory=dict)