858e6007f9
API = Source de vérité unique (SQLite + Flask) - Backend: API v3 avec session, workflow, capture, execute - Frontend: Vanilla TypeScript, pas de state local - Contrats stricts pour les actions RPA - Drag & drop pour réorganiser les étapes - Insertion d'étapes entre deux existantes - Bibliothèque de captures (sessionStorage) - Exécution avec coordonnées statiques (pyautogui) Fonctionne mais fragile (coordonnées fixes, pas de détection visuelle) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
307 lines
11 KiB
Python
307 lines
11 KiB
Python
"""
|
|
Modèles SQLAlchemy - RPA Vision v3
|
|
Source de vérité unique pour les workflows VWB
|
|
|
|
Auteur: Dom, Alice, Kiro - 23 janvier 2026
|
|
"""
|
|
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from datetime import datetime
|
|
from typing import Dict, Any, List, Optional
|
|
import json
|
|
|
|
# Instance SQLAlchemy partagée
|
|
db = SQLAlchemy()
|
|
|
|
|
|
class Workflow(db.Model):
|
|
"""Workflow VWB - Conteneur d'étapes ordonnées"""
|
|
__tablename__ = 'workflows'
|
|
|
|
id = db.Column(db.String(64), primary_key=True)
|
|
name = db.Column(db.String(255), nullable=False)
|
|
description = db.Column(db.Text, nullable=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
|
|
# Relations
|
|
steps = db.relationship('Step', backref='workflow', lazy='dynamic',
|
|
order_by='Step.order', cascade='all, delete-orphan')
|
|
executions = db.relationship('Execution', backref='workflow', lazy='dynamic',
|
|
cascade='all, delete-orphan')
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Sérialise le workflow complet"""
|
|
return {
|
|
'id': self.id,
|
|
'name': self.name,
|
|
'description': self.description,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
|
|
'steps': [step.to_dict() for step in self.steps.order_by(Step.order).all()],
|
|
'step_count': self.steps.count()
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f'<Workflow {self.id}: {self.name}>'
|
|
|
|
|
|
class Step(db.Model):
|
|
"""Étape d'un workflow - Action VWB avec type et paramètres"""
|
|
__tablename__ = 'steps'
|
|
|
|
id = db.Column(db.String(64), primary_key=True)
|
|
workflow_id = db.Column(db.String(64), db.ForeignKey('workflows.id'), nullable=False)
|
|
|
|
# Type d'action - SOURCE DE VÉRITÉ UNIQUE
|
|
action_type = db.Column(db.String(64), nullable=False)
|
|
|
|
# Ordre dans le workflow (0-indexed)
|
|
order = db.Column(db.Integer, nullable=False, default=0)
|
|
|
|
# Position sur le canvas (pour l'affichage)
|
|
position_x = db.Column(db.Float, default=0)
|
|
position_y = db.Column(db.Float, default=0)
|
|
|
|
# Paramètres de l'action (JSON)
|
|
parameters_json = db.Column(db.Text, default='{}')
|
|
|
|
# Référence vers l'ancre visuelle (si applicable)
|
|
anchor_id = db.Column(db.String(64), db.ForeignKey('visual_anchors.id'), nullable=True)
|
|
|
|
# Métadonnées
|
|
label = db.Column(db.String(255), nullable=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
# Relations
|
|
anchor = db.relationship('VisualAnchor', backref='steps')
|
|
|
|
@property
|
|
def parameters(self) -> Dict[str, Any]:
|
|
"""Retourne les paramètres comme dict"""
|
|
try:
|
|
return json.loads(self.parameters_json) if self.parameters_json else {}
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
@parameters.setter
|
|
def parameters(self, value: Dict[str, Any]):
|
|
"""Stocke les paramètres comme JSON"""
|
|
self.parameters_json = json.dumps(value) if value else '{}'
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Sérialise l'étape"""
|
|
return {
|
|
'id': self.id,
|
|
'workflow_id': self.workflow_id,
|
|
'action_type': self.action_type, # SOURCE DE VÉRITÉ
|
|
'order': self.order,
|
|
'position': {'x': self.position_x, 'y': self.position_y},
|
|
'parameters': self.parameters,
|
|
'anchor_id': self.anchor_id,
|
|
'anchor': self.anchor.to_dict() if self.anchor else None,
|
|
'label': self.label or self.action_type,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'updated_at': self.updated_at.isoformat() if self.updated_at else None
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f'<Step {self.id}: {self.action_type} (order={self.order})>'
|
|
|
|
|
|
class VisualAnchor(db.Model):
|
|
"""Ancre visuelle - Image de référence pour localiser un élément UI"""
|
|
__tablename__ = 'visual_anchors'
|
|
|
|
id = db.Column(db.String(64), primary_key=True)
|
|
|
|
# Image de référence (chemin fichier, pas base64 en DB)
|
|
image_path = db.Column(db.String(512), nullable=True)
|
|
thumbnail_path = db.Column(db.String(512), nullable=True)
|
|
|
|
# Bounding box de la sélection
|
|
bbox_x = db.Column(db.Float, nullable=True)
|
|
bbox_y = db.Column(db.Float, nullable=True)
|
|
bbox_width = db.Column(db.Float, nullable=True)
|
|
bbox_height = db.Column(db.Float, nullable=True)
|
|
|
|
# Résolution de l'écran lors de la capture
|
|
screen_width = db.Column(db.Integer, nullable=True)
|
|
screen_height = db.Column(db.Integer, nullable=True)
|
|
|
|
# Description pour l'utilisateur
|
|
description = db.Column(db.Text, nullable=True)
|
|
|
|
# Seuil de confiance pour la détection
|
|
confidence_threshold = db.Column(db.Float, default=0.8)
|
|
|
|
# Métadonnées
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
capture_method = db.Column(db.String(64), default='screen_capture')
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Sérialise l'ancre visuelle"""
|
|
return {
|
|
'id': self.id,
|
|
'image_url': f'/api/v3/anchor/{self.id}/image' if self.image_path else None,
|
|
'thumbnail_url': f'/api/v3/anchor/{self.id}/thumbnail' if self.thumbnail_path else None,
|
|
'bounding_box': {
|
|
'x': self.bbox_x,
|
|
'y': self.bbox_y,
|
|
'width': self.bbox_width,
|
|
'height': self.bbox_height
|
|
} if self.bbox_x is not None else None,
|
|
'screen_resolution': {
|
|
'width': self.screen_width,
|
|
'height': self.screen_height
|
|
} if self.screen_width else None,
|
|
'description': self.description,
|
|
'confidence_threshold': self.confidence_threshold,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f'<VisualAnchor {self.id}>'
|
|
|
|
|
|
class Execution(db.Model):
|
|
"""Exécution d'un workflow - Historique et état"""
|
|
__tablename__ = 'executions'
|
|
|
|
id = db.Column(db.String(64), primary_key=True)
|
|
workflow_id = db.Column(db.String(64), db.ForeignKey('workflows.id'), nullable=False)
|
|
|
|
# État de l'exécution
|
|
status = db.Column(db.String(32), default='pending') # pending, running, paused, completed, error, cancelled
|
|
|
|
# Timestamps
|
|
started_at = db.Column(db.DateTime, nullable=True)
|
|
ended_at = db.Column(db.DateTime, nullable=True)
|
|
|
|
# Progression
|
|
current_step_index = db.Column(db.Integer, default=0)
|
|
total_steps = db.Column(db.Integer, default=0)
|
|
|
|
# Résumé
|
|
completed_steps = db.Column(db.Integer, default=0)
|
|
failed_steps = db.Column(db.Integer, default=0)
|
|
|
|
# Erreur globale (si applicable)
|
|
error_message = db.Column(db.Text, nullable=True)
|
|
|
|
# Relations
|
|
step_results = db.relationship('ExecutionStep', backref='execution', lazy='dynamic',
|
|
cascade='all, delete-orphan')
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Sérialise l'exécution"""
|
|
return {
|
|
'id': self.id,
|
|
'workflow_id': self.workflow_id,
|
|
'status': self.status,
|
|
'started_at': self.started_at.isoformat() if self.started_at else None,
|
|
'ended_at': self.ended_at.isoformat() if self.ended_at else None,
|
|
'current_step_index': self.current_step_index,
|
|
'total_steps': self.total_steps,
|
|
'completed_steps': self.completed_steps,
|
|
'failed_steps': self.failed_steps,
|
|
'error_message': self.error_message,
|
|
'progress': (self.current_step_index / self.total_steps * 100) if self.total_steps > 0 else 0,
|
|
'step_results': [sr.to_dict() for sr in self.step_results.all()]
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f'<Execution {self.id}: {self.status}>'
|
|
|
|
|
|
class ExecutionStep(db.Model):
|
|
"""Résultat d'exécution d'une étape"""
|
|
__tablename__ = 'execution_steps'
|
|
|
|
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
|
|
execution_id = db.Column(db.String(64), db.ForeignKey('executions.id'), nullable=False)
|
|
step_id = db.Column(db.String(64), nullable=False)
|
|
|
|
# Résultat
|
|
status = db.Column(db.String(32), default='pending') # pending, running, success, error, skipped
|
|
|
|
# Timestamps
|
|
started_at = db.Column(db.DateTime, nullable=True)
|
|
ended_at = db.Column(db.DateTime, nullable=True)
|
|
duration_ms = db.Column(db.Integer, nullable=True)
|
|
|
|
# Erreur (si applicable)
|
|
error_message = db.Column(db.Text, nullable=True)
|
|
|
|
# Evidence (chemin vers screenshot)
|
|
evidence_path = db.Column(db.String(512), nullable=True)
|
|
|
|
# Output data (JSON)
|
|
output_json = db.Column(db.Text, default='{}')
|
|
|
|
@property
|
|
def output(self) -> Dict[str, Any]:
|
|
try:
|
|
return json.loads(self.output_json) if self.output_json else {}
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
@output.setter
|
|
def output(self, value: Dict[str, Any]):
|
|
self.output_json = json.dumps(value) if value else '{}'
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
'step_id': self.step_id,
|
|
'status': self.status,
|
|
'started_at': self.started_at.isoformat() if self.started_at else None,
|
|
'ended_at': self.ended_at.isoformat() if self.ended_at else None,
|
|
'duration_ms': self.duration_ms,
|
|
'error_message': self.error_message,
|
|
'evidence_url': f'/api/v3/evidence/{self.id}' if self.evidence_path else None,
|
|
'output': self.output
|
|
}
|
|
|
|
|
|
# Session active (en mémoire, pas en DB)
|
|
class SessionState:
|
|
"""État de la session utilisateur (en mémoire)"""
|
|
|
|
def __init__(self):
|
|
self.active_workflow_id: Optional[str] = None
|
|
self.selected_step_id: Optional[str] = None
|
|
self.active_execution_id: Optional[str] = None
|
|
self.last_capture: Optional[Dict[str, Any]] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
'active_workflow_id': self.active_workflow_id,
|
|
'selected_step_id': self.selected_step_id,
|
|
'active_execution_id': self.active_execution_id,
|
|
'has_capture': self.last_capture is not None
|
|
}
|
|
|
|
|
|
# Instance globale de session
|
|
_session_state = SessionState()
|
|
|
|
|
|
def get_session_state() -> SessionState:
|
|
"""Retourne l'état de session global"""
|
|
return _session_state
|
|
|
|
|
|
def init_db(app):
|
|
"""Initialise la base de données avec l'application Flask"""
|
|
db.init_app(app)
|
|
with app.app_context():
|
|
db.create_all()
|
|
print("✅ [DB] Base de données SQLite initialisée")
|
|
|
|
|
|
def get_db_session():
|
|
"""Retourne la session DB actuelle"""
|
|
return db.session
|