Feat: Humanizer anti-détection pour environnements Citrix/VDI
- Module humanizer.py avec simulation comportement humain - Courbes de Bézier pour mouvements souris - Décalage gaussien pour positions de clic - Frappe avec rythme variable et micro-erreurs - 4 profils: fast, normal, slow, stealth - Intégré dans click_anchor et type_text (humanize=True par défaut) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Dom
@@ -0,0 +1,414 @@
|
||||
"""
|
||||
Action VWB - Clic sur Ancre Visuelle
|
||||
|
||||
Auteur : Dom, Alice, Kiro - 09 janvier 2026
|
||||
|
||||
Cette action permet de cliquer sur un élément UI identifié par une ancre visuelle
|
||||
dans le Visual Workflow Builder.
|
||||
|
||||
Classes :
|
||||
- VWBClickAnchorAction : Action de clic sur ancre visuelle
|
||||
"""
|
||||
|
||||
from typing import Dict, Any, List, Optional, Tuple
|
||||
from datetime import datetime
|
||||
import time
|
||||
|
||||
# Import des modules de base
|
||||
from ..base_action import BaseVWBAction, VWBActionResult, VWBActionStatus
|
||||
from ...contracts.error import VWBErrorType, VWBErrorSeverity, create_vwb_error
|
||||
from ...contracts.evidence import VWBEvidenceType, create_interaction_evidence
|
||||
from ...contracts.visual_anchor import VWBVisualAnchor
|
||||
|
||||
|
||||
class VWBClickAnchorAction(BaseVWBAction):
|
||||
"""
|
||||
Action de clic sur ancre visuelle VWB.
|
||||
|
||||
Cette action localise un élément UI à partir d'une ancre visuelle
|
||||
et effectue un clic dessus. Elle utilise le ScreenCapturer VWB existant
|
||||
avec l'Option A (thread-safe) pour la capture d'écran.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
action_id: str,
|
||||
parameters: Dict[str, Any],
|
||||
screen_capturer=None
|
||||
):
|
||||
"""
|
||||
Initialise l'action de clic sur ancre.
|
||||
|
||||
Args:
|
||||
action_id: Identifiant unique de l'action
|
||||
parameters: Paramètres incluant l'ancre visuelle
|
||||
screen_capturer: Instance du ScreenCapturer (Option A thread-safe)
|
||||
"""
|
||||
super().__init__(
|
||||
action_id=action_id,
|
||||
name="Clic sur Ancre Visuelle",
|
||||
description="Clique sur un élément UI identifié par une ancre visuelle",
|
||||
parameters=parameters,
|
||||
screen_capturer=screen_capturer
|
||||
)
|
||||
|
||||
# Paramètres spécifiques au clic
|
||||
self.visual_anchor: Optional[VWBVisualAnchor] = parameters.get('visual_anchor')
|
||||
self.click_type = parameters.get('click_type', 'left') # left, right, double
|
||||
self.click_offset_x = parameters.get('click_offset_x', 0)
|
||||
self.click_offset_y = parameters.get('click_offset_y', 0)
|
||||
self.wait_after_click_ms = parameters.get('wait_after_click_ms', 500)
|
||||
|
||||
# Configuration de matching
|
||||
self.confidence_threshold = parameters.get('confidence_threshold', 0.8)
|
||||
self.search_timeout_ms = parameters.get('search_timeout_ms', 5000)
|
||||
|
||||
# Humanisation (anti-détection Citrix/VDI)
|
||||
self.humanize = parameters.get('humanize', True)
|
||||
self.humanize_profile = parameters.get('humanize_profile', 'normal')
|
||||
|
||||
def validate_parameters(self) -> List[str]:
|
||||
"""Valide les paramètres de l'action de clic."""
|
||||
errors = []
|
||||
|
||||
# Vérifier l'ancre visuelle
|
||||
if not self.visual_anchor:
|
||||
errors.append("Ancre visuelle requise")
|
||||
elif not isinstance(self.visual_anchor, VWBVisualAnchor):
|
||||
errors.append("Ancre visuelle invalide")
|
||||
elif not self.visual_anchor.is_active:
|
||||
errors.append("Ancre visuelle inactive")
|
||||
|
||||
# Vérifier le type de clic
|
||||
if self.click_type not in ['left', 'right', 'double']:
|
||||
errors.append(f"Type de clic invalide: {self.click_type}")
|
||||
|
||||
# Vérifier les offsets
|
||||
if not isinstance(self.click_offset_x, (int, float)):
|
||||
errors.append("Offset X doit être un nombre")
|
||||
if not isinstance(self.click_offset_y, (int, float)):
|
||||
errors.append("Offset Y doit être un nombre")
|
||||
|
||||
# Vérifier le seuil de confiance
|
||||
if not (0.0 <= self.confidence_threshold <= 1.0):
|
||||
errors.append("Seuil de confiance doit être entre 0.0 et 1.0")
|
||||
|
||||
# Vérifier le ScreenCapturer
|
||||
if not self.screen_capturer:
|
||||
errors.append("ScreenCapturer requis pour la capture d'écran")
|
||||
|
||||
return errors
|
||||
|
||||
def execute_core(self, step_id: str) -> VWBActionResult:
|
||||
"""
|
||||
Exécute l'action de clic sur ancre visuelle.
|
||||
|
||||
Args:
|
||||
step_id: Identifiant de l'étape
|
||||
|
||||
Returns:
|
||||
Résultat d'exécution
|
||||
"""
|
||||
start_time = datetime.now()
|
||||
|
||||
try:
|
||||
# Étape 1: Capturer l'écran actuel
|
||||
print(f"🔍 Recherche de l'ancre visuelle: {self.visual_anchor.name}")
|
||||
current_screenshot = self._capture_current_screen()
|
||||
if current_screenshot is None:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.SCREEN_CAPTURE_FAILED,
|
||||
message="Impossible de capturer l'écran actuel"
|
||||
)
|
||||
|
||||
# Étape 2: Localiser l'ancre visuelle
|
||||
match_result = self._find_visual_anchor(current_screenshot)
|
||||
if not match_result['found']:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.ELEMENT_NOT_FOUND,
|
||||
message=f"Ancre visuelle '{self.visual_anchor.name}' non trouvée",
|
||||
technical_details=match_result
|
||||
)
|
||||
|
||||
# Étape 3: Calculer les coordonnées de clic
|
||||
click_coordinates = self._calculate_click_coordinates(match_result)
|
||||
print(f"🎯 Coordonnées de clic calculées: {click_coordinates}")
|
||||
|
||||
# Étape 4: Effectuer le clic
|
||||
click_success = self._perform_click(click_coordinates)
|
||||
if not click_success:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.CLICK_FAILED,
|
||||
message="Échec du clic sur l'élément",
|
||||
technical_details={'coordinates': click_coordinates}
|
||||
)
|
||||
|
||||
# Étape 5: Attendre après le clic
|
||||
if self.wait_after_click_ms > 0:
|
||||
time.sleep(self.wait_after_click_ms / 1000.0)
|
||||
|
||||
# Étape 6: Mettre à jour les statistiques de l'ancre
|
||||
end_time = datetime.now()
|
||||
execution_time = (end_time - start_time).total_seconds() * 1000
|
||||
self.visual_anchor.update_usage_stats(execution_time, True)
|
||||
|
||||
# Créer l'evidence d'interaction
|
||||
interaction_evidence = create_interaction_evidence(
|
||||
action_id=self.action_id,
|
||||
step_id=step_id,
|
||||
evidence_type=VWBEvidenceType.CLICK_EVIDENCE,
|
||||
title=f"Clic sur {self.visual_anchor.name}",
|
||||
interaction_data={
|
||||
'anchor_id': self.visual_anchor.anchor_id,
|
||||
'anchor_name': self.visual_anchor.name,
|
||||
'click_coordinates': click_coordinates,
|
||||
'click_type': self.click_type,
|
||||
'confidence_score': match_result.get('confidence', 0.0),
|
||||
'search_time_ms': match_result.get('search_time_ms', 0),
|
||||
'match_box': match_result.get('match_box')
|
||||
},
|
||||
confidence_score=match_result.get('confidence', 0.0)
|
||||
)
|
||||
|
||||
# Créer le résultat de succès
|
||||
result = VWBActionResult(
|
||||
action_id=self.action_id,
|
||||
step_id=step_id,
|
||||
status=VWBActionStatus.SUCCESS,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
execution_time_ms=execution_time,
|
||||
output_data={
|
||||
'click_coordinates': click_coordinates,
|
||||
'click_type': self.click_type,
|
||||
'anchor_confidence': match_result.get('confidence', 0.0),
|
||||
'anchor_found': True
|
||||
},
|
||||
evidence_list=[interaction_evidence]
|
||||
)
|
||||
|
||||
print(f"✅ Clic réussi sur {self.visual_anchor.name} en {execution_time:.1f}ms")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.SYSTEM_ERROR,
|
||||
message=f"Erreur lors du clic: {str(e)}",
|
||||
technical_details={'exception': str(e)}
|
||||
)
|
||||
|
||||
def _capture_current_screen(self) -> Optional[Dict[str, Any]]:
|
||||
"""Capture l'écran actuel avec métadonnées."""
|
||||
try:
|
||||
# Utiliser la méthode ultra stable (Option A)
|
||||
img_array = self.screen_capturer.capture()
|
||||
if img_array is None:
|
||||
return None
|
||||
|
||||
from PIL import Image
|
||||
import base64
|
||||
import io
|
||||
|
||||
# Convertir en PIL Image
|
||||
pil_image = Image.fromarray(img_array)
|
||||
|
||||
# Convertir en base64 pour stockage
|
||||
buffer = io.BytesIO()
|
||||
pil_image.save(buffer, format='PNG', optimize=True)
|
||||
screenshot_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
|
||||
|
||||
return {
|
||||
'image_array': img_array,
|
||||
'pil_image': pil_image,
|
||||
'screenshot_base64': screenshot_base64,
|
||||
'width': pil_image.width,
|
||||
'height': pil_image.height,
|
||||
'timestamp': datetime.now().isoformat()
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Erreur capture d'écran: {e}")
|
||||
return None
|
||||
|
||||
def _find_visual_anchor(self, screenshot_data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Localise l'ancre visuelle dans le screenshot.
|
||||
|
||||
Args:
|
||||
screenshot_data: Données du screenshot
|
||||
|
||||
Returns:
|
||||
Résultat de la recherche avec coordonnées si trouvé
|
||||
"""
|
||||
search_start = time.time()
|
||||
|
||||
try:
|
||||
# Simulation de recherche d'ancre visuelle
|
||||
# Dans une implémentation complète, ceci utiliserait:
|
||||
# - Template matching pour ancres image
|
||||
# - OCR pour ancres texte
|
||||
# - Embeddings CLIP pour recherche sémantique
|
||||
|
||||
pil_image = screenshot_data['pil_image']
|
||||
|
||||
# Pour cette implémentation de base, simuler une recherche réussie
|
||||
# au centre de l'écran avec une confiance élevée
|
||||
center_x = pil_image.width // 2
|
||||
center_y = pil_image.height // 2
|
||||
|
||||
# Simuler un délai de recherche réaliste
|
||||
search_delay = min(self.search_timeout_ms / 1000.0, 0.5)
|
||||
time.sleep(search_delay)
|
||||
|
||||
search_time_ms = (time.time() - search_start) * 1000
|
||||
|
||||
# Vérifier si l'ancre a une bounding box définie
|
||||
if self.visual_anchor.has_bounding_box():
|
||||
# Utiliser la zone de recherche adaptée à la résolution
|
||||
search_area = self.visual_anchor.get_search_area(
|
||||
pil_image.width,
|
||||
pil_image.height
|
||||
)
|
||||
if search_area:
|
||||
center_x = search_area['x'] + search_area['width'] // 2
|
||||
center_y = search_area['y'] + search_area['height'] // 2
|
||||
|
||||
# Simuler une confiance basée sur le seuil configuré
|
||||
confidence = min(self.confidence_threshold + 0.1, 0.95)
|
||||
|
||||
return {
|
||||
'found': confidence >= self.confidence_threshold,
|
||||
'confidence': confidence,
|
||||
'match_box': {
|
||||
'x': center_x - 50,
|
||||
'y': center_y - 25,
|
||||
'width': 100,
|
||||
'height': 50
|
||||
},
|
||||
'center_coordinates': {'x': center_x, 'y': center_y},
|
||||
'search_time_ms': search_time_ms,
|
||||
'method': 'simulated_template_matching'
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
search_time_ms = (time.time() - search_start) * 1000
|
||||
return {
|
||||
'found': False,
|
||||
'error': str(e),
|
||||
'search_time_ms': search_time_ms
|
||||
}
|
||||
|
||||
def _calculate_click_coordinates(self, match_result: Dict[str, Any]) -> Dict[str, int]:
|
||||
"""
|
||||
Calcule les coordonnées finales de clic avec offsets.
|
||||
|
||||
Args:
|
||||
match_result: Résultat de la recherche d'ancre
|
||||
|
||||
Returns:
|
||||
Coordonnées de clic finales
|
||||
"""
|
||||
center_coords = match_result['center_coordinates']
|
||||
|
||||
final_x = int(center_coords['x'] + self.click_offset_x)
|
||||
final_y = int(center_coords['y'] + self.click_offset_y)
|
||||
|
||||
return {
|
||||
'x': final_x,
|
||||
'y': final_y,
|
||||
'offset_x': self.click_offset_x,
|
||||
'offset_y': self.click_offset_y
|
||||
}
|
||||
|
||||
def _perform_click(self, coordinates: Dict[str, int]) -> bool:
|
||||
"""
|
||||
Effectue le clic aux coordonnées spécifiées.
|
||||
|
||||
Args:
|
||||
coordinates: Coordonnées de clic
|
||||
|
||||
Returns:
|
||||
True si le clic a réussi
|
||||
"""
|
||||
try:
|
||||
x, y = coordinates['x'], coordinates['y']
|
||||
|
||||
# Utiliser le Humanizer si activé (anti-détection Citrix/VDI)
|
||||
if self.humanize:
|
||||
try:
|
||||
from ...utils.humanizer import Humanizer, HumanProfile
|
||||
|
||||
# Sélectionner le profil
|
||||
profile_map = {
|
||||
'fast': HumanProfile.FAST,
|
||||
'normal': HumanProfile.NORMAL,
|
||||
'slow': HumanProfile.SLOW,
|
||||
'stealth': HumanProfile.STEALTH,
|
||||
}
|
||||
profile = profile_map.get(self.humanize_profile, HumanProfile.NORMAL)
|
||||
|
||||
humanizer = Humanizer(profile=profile)
|
||||
|
||||
if self.click_type == 'left':
|
||||
real_x, real_y = humanizer.click(x, y, button='left')
|
||||
elif self.click_type == 'right':
|
||||
real_x, real_y = humanizer.click(x, y, button='right')
|
||||
elif self.click_type == 'double':
|
||||
real_x, real_y = humanizer.double_click(x, y)
|
||||
|
||||
print(f"🖱️ Clic humanisé {self.click_type} à ({real_x}, {real_y}) [cible: {x}, {y}]")
|
||||
return True
|
||||
|
||||
except ImportError as e:
|
||||
print(f"⚠️ Humanizer non disponible: {e}, fallback pyautogui direct")
|
||||
|
||||
# Fallback: pyautogui direct (sans humanisation)
|
||||
try:
|
||||
import pyautogui
|
||||
|
||||
if self.click_type == 'left':
|
||||
pyautogui.click(x, y)
|
||||
elif self.click_type == 'right':
|
||||
pyautogui.rightClick(x, y)
|
||||
elif self.click_type == 'double':
|
||||
pyautogui.doubleClick(x, y)
|
||||
|
||||
print(f"🖱️ Clic {self.click_type} effectué à ({x}, {y})")
|
||||
return True
|
||||
|
||||
except ImportError:
|
||||
print("⚠️ pyautogui non disponible - simulation du clic")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Erreur lors du clic: {e}")
|
||||
return False
|
||||
|
||||
def get_action_info(self) -> Dict[str, Any]:
|
||||
"""Retourne les informations de l'action pour l'interface."""
|
||||
return {
|
||||
'action_id': self.action_id,
|
||||
'name': self.name,
|
||||
'description': self.description,
|
||||
'type': 'click_anchor',
|
||||
'parameters': {
|
||||
'anchor_name': self.visual_anchor.name if self.visual_anchor else 'Non définie',
|
||||
'click_type': self.click_type,
|
||||
'confidence_threshold': self.confidence_threshold,
|
||||
'click_offset': {
|
||||
'x': self.click_offset_x,
|
||||
'y': self.click_offset_y
|
||||
},
|
||||
'wait_after_click_ms': self.wait_after_click_ms
|
||||
},
|
||||
'status': self.current_status.value,
|
||||
'anchor_reliable': self.visual_anchor.is_reliable() if self.visual_anchor else False
|
||||
}
|
||||
452
visual_workflow_builder/backend/actions/vision_ui/type_text.py
Normal file
452
visual_workflow_builder/backend/actions/vision_ui/type_text.py
Normal file
@@ -0,0 +1,452 @@
|
||||
"""
|
||||
Action VWB - Saisie de Texte
|
||||
|
||||
Auteur : Dom, Alice, Kiro - 09 janvier 2026
|
||||
|
||||
Cette action permet de saisir du texte dans un champ identifié par une ancre visuelle
|
||||
dans le Visual Workflow Builder.
|
||||
|
||||
Classes :
|
||||
- VWBTypeTextAction : Action de saisie de texte
|
||||
"""
|
||||
|
||||
from typing import Dict, Any, List, Optional
|
||||
from datetime import datetime
|
||||
import time
|
||||
|
||||
# Import des modules de base
|
||||
from ..base_action import BaseVWBAction, VWBActionResult, VWBActionStatus
|
||||
from ...contracts.error import VWBErrorType, VWBErrorSeverity, create_vwb_error
|
||||
from ...contracts.evidence import VWBEvidenceType, create_interaction_evidence
|
||||
from ...contracts.visual_anchor import VWBVisualAnchor
|
||||
|
||||
|
||||
class VWBTypeTextAction(BaseVWBAction):
|
||||
"""
|
||||
Action de saisie de texte VWB.
|
||||
|
||||
Cette action localise un champ de saisie à partir d'une ancre visuelle
|
||||
et y saisit le texte spécifié. Elle peut optionnellement cliquer sur le champ
|
||||
avant la saisie et valider après.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
action_id: str,
|
||||
parameters: Dict[str, Any],
|
||||
screen_capturer=None
|
||||
):
|
||||
"""
|
||||
Initialise l'action de saisie de texte.
|
||||
|
||||
Args:
|
||||
action_id: Identifiant unique de l'action
|
||||
parameters: Paramètres incluant l'ancre et le texte
|
||||
screen_capturer: Instance du ScreenCapturer (Option A thread-safe)
|
||||
"""
|
||||
super().__init__(
|
||||
action_id=action_id,
|
||||
name="Saisie de Texte",
|
||||
description="Saisit du texte dans un champ identifié par une ancre visuelle",
|
||||
parameters=parameters,
|
||||
screen_capturer=screen_capturer
|
||||
)
|
||||
|
||||
# Paramètres spécifiques à la saisie
|
||||
self.visual_anchor: Optional[VWBVisualAnchor] = parameters.get('visual_anchor')
|
||||
self.text_to_type = parameters.get('text_to_type', '')
|
||||
self.clear_field_first = parameters.get('clear_field_first', True)
|
||||
self.click_before_typing = parameters.get('click_before_typing', True)
|
||||
self.press_enter_after = parameters.get('press_enter_after', False)
|
||||
self.typing_speed_ms = parameters.get('typing_speed_ms', 50) # Délai entre caractères
|
||||
self.wait_after_typing_ms = parameters.get('wait_after_typing_ms', 500)
|
||||
|
||||
# Configuration de matching
|
||||
self.confidence_threshold = parameters.get('confidence_threshold', 0.8)
|
||||
self.search_timeout_ms = parameters.get('search_timeout_ms', 5000)
|
||||
|
||||
# Humanisation (anti-détection Citrix/VDI)
|
||||
self.humanize = parameters.get('humanize', True)
|
||||
self.humanize_profile = parameters.get('humanize_profile', 'normal')
|
||||
|
||||
def validate_parameters(self) -> List[str]:
|
||||
"""Valide les paramètres de l'action de saisie."""
|
||||
errors = []
|
||||
|
||||
# Vérifier l'ancre visuelle
|
||||
if not self.visual_anchor:
|
||||
errors.append("Ancre visuelle requise")
|
||||
elif not isinstance(self.visual_anchor, VWBVisualAnchor):
|
||||
errors.append("Ancre visuelle invalide")
|
||||
elif not self.visual_anchor.is_active:
|
||||
errors.append("Ancre visuelle inactive")
|
||||
|
||||
# Vérifier le texte à saisir
|
||||
if not isinstance(self.text_to_type, str):
|
||||
errors.append("Le texte à saisir doit être une chaîne")
|
||||
|
||||
# Vérifier les paramètres booléens
|
||||
if not isinstance(self.clear_field_first, bool):
|
||||
errors.append("clear_field_first doit être un booléen")
|
||||
if not isinstance(self.click_before_typing, bool):
|
||||
errors.append("click_before_typing doit être un booléen")
|
||||
if not isinstance(self.press_enter_after, bool):
|
||||
errors.append("press_enter_after doit être un booléen")
|
||||
|
||||
# Vérifier les délais
|
||||
if not isinstance(self.typing_speed_ms, (int, float)) or self.typing_speed_ms < 0:
|
||||
errors.append("typing_speed_ms doit être un nombre positif")
|
||||
if not isinstance(self.wait_after_typing_ms, (int, float)) or self.wait_after_typing_ms < 0:
|
||||
errors.append("wait_after_typing_ms doit être un nombre positif")
|
||||
|
||||
# Vérifier le seuil de confiance
|
||||
if not (0.0 <= self.confidence_threshold <= 1.0):
|
||||
errors.append("Seuil de confiance doit être entre 0.0 et 1.0")
|
||||
|
||||
# Vérifier le ScreenCapturer
|
||||
if not self.screen_capturer:
|
||||
errors.append("ScreenCapturer requis pour la capture d'écran")
|
||||
|
||||
return errors
|
||||
|
||||
def execute_core(self, step_id: str) -> VWBActionResult:
|
||||
"""
|
||||
Exécute l'action de saisie de texte.
|
||||
|
||||
Args:
|
||||
step_id: Identifiant de l'étape
|
||||
|
||||
Returns:
|
||||
Résultat d'exécution
|
||||
"""
|
||||
start_time = datetime.now()
|
||||
|
||||
try:
|
||||
# Étape 1: Capturer l'écran actuel
|
||||
print(f"🔍 Recherche du champ de saisie: {self.visual_anchor.name}")
|
||||
current_screenshot = self._capture_current_screen()
|
||||
if current_screenshot is None:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.SCREEN_CAPTURE_FAILED,
|
||||
message="Impossible de capturer l'écran actuel"
|
||||
)
|
||||
|
||||
# Étape 2: Localiser le champ de saisie
|
||||
match_result = self._find_input_field(current_screenshot)
|
||||
if not match_result['found']:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.ELEMENT_NOT_FOUND,
|
||||
message=f"Champ de saisie '{self.visual_anchor.name}' non trouvé",
|
||||
technical_details=match_result
|
||||
)
|
||||
|
||||
# Étape 3: Cliquer sur le champ si nécessaire
|
||||
if self.click_before_typing:
|
||||
click_coordinates = self._calculate_click_coordinates(match_result)
|
||||
click_success = self._perform_click(click_coordinates)
|
||||
if not click_success:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.CLICK_FAILED,
|
||||
message="Impossible de cliquer sur le champ de saisie",
|
||||
technical_details={'coordinates': click_coordinates}
|
||||
)
|
||||
|
||||
# Attendre que le champ soit actif
|
||||
time.sleep(0.2)
|
||||
|
||||
# Étape 4: Vider le champ si nécessaire
|
||||
if self.clear_field_first:
|
||||
self._clear_field()
|
||||
|
||||
# Étape 5: Saisir le texte
|
||||
typing_success = self._type_text()
|
||||
if not typing_success:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.TYPE_TEXT_FAILED,
|
||||
message="Échec de la saisie de texte",
|
||||
technical_details={'text_length': len(self.text_to_type)}
|
||||
)
|
||||
|
||||
# Étape 6: Appuyer sur Entrée si nécessaire
|
||||
if self.press_enter_after:
|
||||
self._press_enter()
|
||||
|
||||
# Étape 7: Attendre après la saisie
|
||||
if self.wait_after_typing_ms > 0:
|
||||
time.sleep(self.wait_after_typing_ms / 1000.0)
|
||||
|
||||
# Étape 8: Mettre à jour les statistiques de l'ancre
|
||||
end_time = datetime.now()
|
||||
execution_time = (end_time - start_time).total_seconds() * 1000
|
||||
self.visual_anchor.update_usage_stats(execution_time, True)
|
||||
|
||||
# Créer l'evidence d'interaction
|
||||
interaction_evidence = create_interaction_evidence(
|
||||
action_id=self.action_id,
|
||||
step_id=step_id,
|
||||
evidence_type=VWBEvidenceType.TYPE_EVIDENCE,
|
||||
title=f"Saisie dans {self.visual_anchor.name}",
|
||||
interaction_data={
|
||||
'anchor_id': self.visual_anchor.anchor_id,
|
||||
'anchor_name': self.visual_anchor.name,
|
||||
'text_typed': self.text_to_type,
|
||||
'text_length': len(self.text_to_type),
|
||||
'cleared_first': self.clear_field_first,
|
||||
'clicked_before': self.click_before_typing,
|
||||
'pressed_enter': self.press_enter_after,
|
||||
'confidence_score': match_result.get('confidence', 0.0),
|
||||
'search_time_ms': match_result.get('search_time_ms', 0),
|
||||
'match_box': match_result.get('match_box')
|
||||
},
|
||||
confidence_score=match_result.get('confidence', 0.0)
|
||||
)
|
||||
|
||||
# Créer le résultat de succès
|
||||
result = VWBActionResult(
|
||||
action_id=self.action_id,
|
||||
step_id=step_id,
|
||||
status=VWBActionStatus.SUCCESS,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
execution_time_ms=execution_time,
|
||||
output_data={
|
||||
'text_typed': self.text_to_type,
|
||||
'text_length': len(self.text_to_type),
|
||||
'field_cleared': self.clear_field_first,
|
||||
'enter_pressed': self.press_enter_after,
|
||||
'anchor_confidence': match_result.get('confidence', 0.0),
|
||||
'field_found': True
|
||||
},
|
||||
evidence_list=[interaction_evidence]
|
||||
)
|
||||
|
||||
print(f"✅ Saisie réussie dans {self.visual_anchor.name} en {execution_time:.1f}ms")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
return self._create_error_result(
|
||||
step_id=step_id,
|
||||
start_time=start_time,
|
||||
error_type=VWBErrorType.SYSTEM_ERROR,
|
||||
message=f"Erreur lors de la saisie: {str(e)}",
|
||||
technical_details={'exception': str(e)}
|
||||
)
|
||||
|
||||
def _capture_current_screen(self) -> Optional[Dict[str, Any]]:
|
||||
"""Capture l'écran actuel avec métadonnées."""
|
||||
try:
|
||||
# Utiliser la méthode ultra stable (Option A)
|
||||
img_array = self.screen_capturer.capture()
|
||||
if img_array is None:
|
||||
return None
|
||||
|
||||
from PIL import Image
|
||||
import base64
|
||||
import io
|
||||
|
||||
# Convertir en PIL Image
|
||||
pil_image = Image.fromarray(img_array)
|
||||
|
||||
# Convertir en base64 pour stockage
|
||||
buffer = io.BytesIO()
|
||||
pil_image.save(buffer, format='PNG', optimize=True)
|
||||
screenshot_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
|
||||
|
||||
return {
|
||||
'image_array': img_array,
|
||||
'pil_image': pil_image,
|
||||
'screenshot_base64': screenshot_base64,
|
||||
'width': pil_image.width,
|
||||
'height': pil_image.height,
|
||||
'timestamp': datetime.now().isoformat()
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Erreur capture d'écran: {e}")
|
||||
return None
|
||||
|
||||
def _find_input_field(self, screenshot_data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Localise le champ de saisie dans le screenshot.
|
||||
|
||||
Args:
|
||||
screenshot_data: Données du screenshot
|
||||
|
||||
Returns:
|
||||
Résultat de la recherche avec coordonnées si trouvé
|
||||
"""
|
||||
search_start = time.time()
|
||||
|
||||
try:
|
||||
pil_image = screenshot_data['pil_image']
|
||||
|
||||
# Simuler un délai de recherche réaliste
|
||||
search_delay = min(self.search_timeout_ms / 1000.0, 0.3)
|
||||
time.sleep(search_delay)
|
||||
|
||||
search_time_ms = (time.time() - search_start) * 1000
|
||||
|
||||
# Simuler la localisation du champ
|
||||
center_x = pil_image.width // 2
|
||||
center_y = pil_image.height // 2
|
||||
|
||||
# Vérifier si l'ancre a une bounding box définie
|
||||
if self.visual_anchor.has_bounding_box():
|
||||
search_area = self.visual_anchor.get_search_area(
|
||||
pil_image.width,
|
||||
pil_image.height
|
||||
)
|
||||
if search_area:
|
||||
center_x = search_area['x'] + search_area['width'] // 2
|
||||
center_y = search_area['y'] + search_area['height'] // 2
|
||||
|
||||
# Simuler une confiance basée sur le seuil configuré
|
||||
confidence = min(self.confidence_threshold + 0.05, 0.92)
|
||||
|
||||
return {
|
||||
'found': confidence >= self.confidence_threshold,
|
||||
'confidence': confidence,
|
||||
'match_box': {
|
||||
'x': center_x - 75,
|
||||
'y': center_y - 15,
|
||||
'width': 150,
|
||||
'height': 30
|
||||
},
|
||||
'center_coordinates': {'x': center_x, 'y': center_y},
|
||||
'search_time_ms': search_time_ms,
|
||||
'method': 'simulated_input_field_detection'
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
search_time_ms = (time.time() - search_start) * 1000
|
||||
return {
|
||||
'found': False,
|
||||
'error': str(e),
|
||||
'search_time_ms': search_time_ms
|
||||
}
|
||||
|
||||
def _calculate_click_coordinates(self, match_result: Dict[str, Any]) -> Dict[str, int]:
|
||||
"""Calcule les coordonnées de clic sur le champ."""
|
||||
center_coords = match_result['center_coordinates']
|
||||
|
||||
return {
|
||||
'x': int(center_coords['x']),
|
||||
'y': int(center_coords['y'])
|
||||
}
|
||||
|
||||
def _perform_click(self, coordinates: Dict[str, int]) -> bool:
|
||||
"""Effectue un clic sur le champ de saisie."""
|
||||
try:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.click(coordinates['x'], coordinates['y'])
|
||||
print(f"🖱️ Clic sur le champ à ({coordinates['x']}, {coordinates['y']})")
|
||||
return True
|
||||
except ImportError:
|
||||
print("⚠️ pyautogui non disponible - simulation du clic")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"❌ Erreur lors du clic: {e}")
|
||||
return False
|
||||
|
||||
def _clear_field(self):
|
||||
"""Vide le champ de saisie."""
|
||||
try:
|
||||
try:
|
||||
import pyautogui
|
||||
# Sélectionner tout le texte et le supprimer
|
||||
pyautogui.hotkey('ctrl', 'a')
|
||||
time.sleep(0.1)
|
||||
pyautogui.press('delete')
|
||||
print("🧹 Champ vidé")
|
||||
except ImportError:
|
||||
print("⚠️ pyautogui non disponible - simulation du vidage")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Erreur lors du vidage: {e}")
|
||||
|
||||
def _type_text(self) -> bool:
|
||||
"""Saisit le texte avec comportement humain."""
|
||||
try:
|
||||
# Utiliser le Humanizer si activé (anti-détection Citrix/VDI)
|
||||
if self.humanize:
|
||||
try:
|
||||
from ...utils.humanizer import Humanizer, HumanProfile
|
||||
|
||||
profile_map = {
|
||||
'fast': HumanProfile.FAST,
|
||||
'normal': HumanProfile.NORMAL,
|
||||
'slow': HumanProfile.SLOW,
|
||||
'stealth': HumanProfile.STEALTH,
|
||||
}
|
||||
profile = profile_map.get(self.humanize_profile, HumanProfile.NORMAL)
|
||||
|
||||
humanizer = Humanizer(profile=profile)
|
||||
typed = humanizer.type_text(self.text_to_type)
|
||||
|
||||
print(f"⌨️ Texte humanisé: '{typed}' ({len(typed)} caractères)")
|
||||
return True
|
||||
|
||||
except ImportError as e:
|
||||
print(f"⚠️ Humanizer non disponible: {e}, fallback pyautogui direct")
|
||||
|
||||
# Fallback: pyautogui direct
|
||||
try:
|
||||
import pyautogui
|
||||
|
||||
for char in self.text_to_type:
|
||||
pyautogui.write(char)
|
||||
if self.typing_speed_ms > 0:
|
||||
time.sleep(self.typing_speed_ms / 1000.0)
|
||||
|
||||
print(f"⌨️ Texte saisi: '{self.text_to_type}' ({len(self.text_to_type)} caractères)")
|
||||
return True
|
||||
|
||||
except ImportError:
|
||||
print(f"⚠️ pyautogui non disponible - simulation: '{self.text_to_type}'")
|
||||
time.sleep(len(self.text_to_type) * self.typing_speed_ms / 1000.0)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Erreur lors de la saisie: {e}")
|
||||
return False
|
||||
|
||||
def _press_enter(self):
|
||||
"""Appuie sur la touche Entrée."""
|
||||
try:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.press('enter')
|
||||
print("⏎ Touche Entrée pressée")
|
||||
except ImportError:
|
||||
print("⚠️ pyautogui non disponible - simulation de la touche Entrée")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Erreur lors de l'appui sur Entrée: {e}")
|
||||
|
||||
def get_action_info(self) -> Dict[str, Any]:
|
||||
"""Retourne les informations de l'action pour l'interface."""
|
||||
return {
|
||||
'action_id': self.action_id,
|
||||
'name': self.name,
|
||||
'description': self.description,
|
||||
'type': 'type_text',
|
||||
'parameters': {
|
||||
'anchor_name': self.visual_anchor.name if self.visual_anchor else 'Non définie',
|
||||
'text_to_type': self.text_to_type,
|
||||
'text_length': len(self.text_to_type),
|
||||
'clear_field_first': self.clear_field_first,
|
||||
'click_before_typing': self.click_before_typing,
|
||||
'press_enter_after': self.press_enter_after,
|
||||
'typing_speed_ms': self.typing_speed_ms,
|
||||
'confidence_threshold': self.confidence_threshold
|
||||
},
|
||||
'status': self.current_status.value,
|
||||
'anchor_reliable': self.visual_anchor.is_reliable() if self.visual_anchor else False
|
||||
}
|
||||
Reference in New Issue
Block a user