rpa_vision_v3/fix_backend_vwb_errors.py

#!/usr/bin/env python3
"""
Correction des Erreurs Backend Visual Workflow Builder

Auteur : Dom, Alice, Kiro - 08 janvier 2026

Ce script identifie et corrige automatiquement les erreurs communes
du backend Visual Workflow Builder.
"""

import os
import sys
import json
import shutil
from pathlib import Path
from typing import List, Dict, Any

def print_section(title: str):
    """Affiche une section avec formatage."""
    print(f"\n{'='*60}")
    print(f" {title}")
    print(f"{'='*60}")

def print_subsection(title: str):
    """Affiche une sous-section."""
    print(f"\n{'-'*40}")
    print(f" {title}")
    print(f"{'-'*40}")

def fix_missing_api_modules():
    """Crée les modules API manquants avec des implémentations minimales."""
    print_subsection("Correction des Modules API Manquants")
    
    backend_path = Path("visual_workflow_builder/backend")
    api_path = backend_path / "api"
    
    # Créer le répertoire api s'il n'existe pas
    api_path.mkdir(exist_ok=True)
    
    # Créer __init__.py
    init_file = api_path / "__init__.py"
    if not init_file.exists():
        init_file.write_text('"""API Package"""')
        print("✅ Créé api/__init__.py")
    
    # Modules API essentiels à créer
    api_modules = {
        "errors.py": '''"""
Gestionnaire d'erreurs pour l'API

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import jsonify
from typing import Any, Dict

class APIError(Exception):
    """Erreur API de base."""
    def __init__(self, message: str, status_code: int = 400):
        super().__init__(message)
        self.message = message
        self.status_code = status_code

class ValidationError(APIError):
    """Erreur de validation."""
    pass

class NotFoundError(APIError):
    """Erreur ressource non trouvée."""
    def __init__(self, message: str):
        super().__init__(message, 404)

class BadRequestError(APIError):
    """Erreur requête invalide."""
    pass

def error_response(status_code: int, message: str) -> tuple:
    """Retourne une réponse d'erreur formatée."""
    return jsonify({
        'error': message,
        'status_code': status_code
    }), status_code
''',
        
        "validation.py": '''"""
Validation des données pour l'API

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from typing import Dict, Any, List

def validate_workflow_data(data: Dict[str, Any]) -> List[str]:
    """Valide les données d'un workflow."""
    errors = []
    
    if not isinstance(data, dict):
        errors.append("Les données doivent être un objet JSON")
        return errors
    
    # Validation du nom
    name = data.get('name')
    if not name or not str(name).strip():
        errors.append("Le nom est requis")
    
    # Validation created_by
    created_by = data.get('created_by')
    if not created_by:
        errors.append("Le champ 'created_by' est requis")
    
    return errors

def validate_update_data(data: Dict[str, Any]) -> List[str]:
    """Valide les données de mise à jour d'un workflow."""
    errors = []
    
    if not isinstance(data, dict):
        errors.append("Les données doivent être un objet JSON")
        return errors
    
    # Si le nom est fourni, il doit être valide
    if 'name' in data:
        name = data['name']
        if not name or not str(name).strip():
            errors.append("Le nom ne peut pas être vide")
    
    return errors
''',
        
        "screen_capture.py": '''"""
API de capture d'écran (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

screen_capture_bp = Blueprint('screen_capture', __name__)

@screen_capture_bp.route('/capture', methods=['POST'])
def capture_screen():
    """Capture d'écran (placeholder)."""
    return jsonify({
        'message': 'Capture d\\'écran non implémentée dans la version allégée',
        'status': 'not_implemented'
    }), 501

@screen_capture_bp.route('/status', methods=['GET'])
def capture_status():
    """Statut du service de capture."""
    return jsonify({
        'status': 'disabled',
        'message': 'Service de capture désactivé en mode allégé'
    })
''',
        
        "real_demo.py": '''"""
API de démonstration réelle (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

real_demo_bp = Blueprint('real_demo', __name__)

@real_demo_bp.route('/demo/start', methods=['POST'])
def start_demo():
    """Démarre une démonstration (placeholder)."""
    return jsonify({
        'message': 'Démonstration non implémentée dans la version allégée',
        'status': 'not_implemented'
    }), 501

@real_demo_bp.route('/demo/status', methods=['GET'])
def demo_status():
    """Statut de la démonstration."""
    return jsonify({
        'status': 'disabled',
        'message': 'Démonstration désactivée en mode allégé'
    })
''',
        
        "node_types.py": '''"""
API des types de nœuds

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

node_types_bp = Blueprint('node_types', __name__)

# Types de nœuds de base
NODE_TYPES = [
    {
        'id': 'start',
        'name': 'Début',
        'description': 'Point de départ du workflow',
        'category': 'control',
        'inputs': [],
        'outputs': ['next']
    },
    {
        'id': 'end',
        'name': 'Fin',
        'description': 'Point de fin du workflow',
        'category': 'control',
        'inputs': ['previous'],
        'outputs': []
    },
    {
        'id': 'action',
        'name': 'Action',
        'description': 'Action générique',
        'category': 'action',
        'inputs': ['previous'],
        'outputs': ['next', 'error']
    },
    {
        'id': 'condition',
        'name': 'Condition',
        'description': 'Branchement conditionnel',
        'category': 'logic',
        'inputs': ['previous'],
        'outputs': ['true', 'false']
    }
]

@node_types_bp.route('/', methods=['GET'])
def list_node_types():
    """Liste tous les types de nœuds disponibles."""
    return jsonify(NODE_TYPES)

@node_types_bp.route('/<node_type_id>', methods=['GET'])
def get_node_type(node_type_id: str):
    """Récupère un type de nœud spécifique."""
    for node_type in NODE_TYPES:
        if node_type['id'] == node_type_id:
            return jsonify(node_type)
    
    return jsonify({'error': 'Type de nœud non trouvé'}), 404
'''
    }
    
    # Créer les modules manquants
    for filename, content in api_modules.items():
        file_path = api_path / filename
        if not file_path.exists():
            file_path.write_text(content)
            print(f"✅ Créé api/{filename}")
        else:
            print(f"⚠️  api/{filename} existe déjà")

def fix_services_directory():
    """Crée le répertoire services avec les modules nécessaires."""
    print_subsection("Correction du Répertoire Services")
    
    backend_path = Path("visual_workflow_builder/backend")
    services_path = backend_path / "services"
    
    # Créer le répertoire services
    services_path.mkdir(exist_ok=True)
    
    # Créer __init__.py
    init_file = services_path / "__init__.py"
    if not init_file.exists():
        init_file.write_text('"""Services Package"""')
        print("✅ Créé services/__init__.py")
    
    # Modules services essentiels
    services_modules = {
        "converter.py": '''"""
Convertisseur de workflows visuels vers graphes

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from typing import Any, Dict, List
from models import VisualWorkflow

class ConversionError(Exception):
    """Erreur de conversion."""
    pass

class WorkflowGraph:
    """Représentation graphe d'un workflow."""
    
    def __init__(self, workflow_id: str):
        self.workflow_id = workflow_id
        self.nodes = []
        self.edges = []
        self.metadata = {}
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            'workflow_id': self.workflow_id,
            'nodes': self.nodes,
            'edges': self.edges,
            'metadata': self.metadata
        }

class VisualToGraphConverter:
    """Convertit un workflow visuel en graphe."""
    
    def __init__(self):
        self.warnings = []
    
    def convert(self, workflow: VisualWorkflow) -> WorkflowGraph:
        """Convertit un workflow visuel en graphe."""
        graph = WorkflowGraph(workflow.id)
        
        # Conversion basique
        graph.nodes = [node.to_dict() for node in workflow.nodes]
        graph.edges = [edge.to_dict() for edge in workflow.edges]
        graph.metadata = {
            'name': workflow.name,
            'description': workflow.description,
            'variables': [var.to_dict() for var in workflow.variables]
        }
        
        return graph
    
    def get_warnings(self) -> List[str]:
        """Retourne les avertissements de conversion."""
        return self.warnings

def convert_visual_to_graph(workflow: VisualWorkflow) -> WorkflowGraph:
    """Fonction utilitaire de conversion."""
    converter = VisualToGraphConverter()
    return converter.convert(workflow)
''',
        
        "execution_integration.py": '''"""
Intégration avec le système d'exécution

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from enum import Enum
from typing import Dict, Any, List, Optional
from datetime import datetime

class ExecutionStatus(Enum):
    """Statuts d'exécution."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

class ExecutionResult:
    """Résultat d'exécution."""
    
    def __init__(self, execution_id: str, status: ExecutionStatus):
        self.execution_id = execution_id
        self.status = status
        self.started_at = datetime.now().isoformat()
        self.completed_at = None
        self.error_message = None
        self.result_data = {}
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            'execution_id': self.execution_id,
            'status': self.status.value,
            'started_at': self.started_at,
            'completed_at': self.completed_at,
            'error_message': self.error_message,
            'result_data': self.result_data
        }

class MockExecutor:
    """Exécuteur simulé pour la version allégée."""
    
    def __init__(self):
        self.executions: Dict[str, ExecutionResult] = {}
    
    def execute_workflow(self, workflow_id: str, variables: Dict[str, Any]) -> str:
        """Lance l'exécution d'un workflow."""
        execution_id = f"exec_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        result = ExecutionResult(execution_id, ExecutionStatus.PENDING)
        self.executions[execution_id] = result
        
        # Simulation: marquer comme complété immédiatement
        result.status = ExecutionStatus.COMPLETED
        result.completed_at = datetime.now().isoformat()
        result.result_data = {
            'workflow_id': workflow_id,
            'variables': variables,
            'message': 'Exécution simulée (mode allégé)'
        }
        
        return execution_id
    
    def get_execution_status(self, execution_id: str) -> Optional[ExecutionResult]:
        """Récupère le statut d'une exécution."""
        return self.executions.get(execution_id)
    
    def cancel_execution(self, execution_id: str) -> bool:
        """Annule une exécution."""
        if execution_id in self.executions:
            result = self.executions[execution_id]
            if result.status in [ExecutionStatus.PENDING, ExecutionStatus.RUNNING]:
                result.status = ExecutionStatus.CANCELLED
                result.completed_at = datetime.now().isoformat()
                return True
        return False
    
    def list_executions(self, workflow_id: Optional[str] = None) -> List[Dict[str, Any]]:
        """Liste les exécutions."""
        executions = []
        for result in self.executions.values():
            if workflow_id is None or result.result_data.get('workflow_id') == workflow_id:
                executions.append(result.to_dict())
        return executions

# Instance globale
_executor = MockExecutor()

def get_executor() -> MockExecutor:
    """Retourne l'instance de l'exécuteur."""
    return _executor
'''
    }
    
    # Créer les modules services
    for filename, content in services_modules.items():
        file_path = services_path / filename
        if not file_path.exists():
            file_path.write_text(content)
            print(f"✅ Créé services/{filename}")
        else:
            print(f"⚠️  services/{filename} existe déjà")

def fix_optional_blueprints():
    """Crée des versions minimales des blueprints optionnels."""
    print_subsection("Correction des Blueprints Optionnels")
    
    backend_path = Path("visual_workflow_builder/backend")
    api_path = backend_path / "api"
    
    optional_blueprints = {
        "self_healing.py": '''"""
API Self-Healing (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

self_healing_bp = Blueprint('self_healing', __name__)

@self_healing_bp.route('/status', methods=['GET'])
def self_healing_status():
    """Statut du self-healing."""
    return jsonify({
        'status': 'disabled',
        'message': 'Self-healing désactivé en mode allégé'
    })
''',
        
        "visual_targets.py": '''"""
API Visual Targets (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

visual_targets_bp = Blueprint('visual_targets', __name__)

def init_visual_target_manager(screen_capturer, ui_detector, fusion_engine):
    """Initialise le gestionnaire de cibles visuelles (placeholder)."""
    pass

@visual_targets_bp.route('/status', methods=['GET'])
def visual_targets_status():
    """Statut des cibles visuelles."""
    return jsonify({
        'status': 'disabled',
        'message': 'Cibles visuelles désactivées en mode allégé'
    })
''',
        
        "element_detection.py": '''"""
API Element Detection (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

element_detection_bp = Blueprint('element_detection', __name__)

def init_element_detection(ui_detector, screen_capturer):
    """Initialise la détection d'éléments (placeholder)."""
    pass

@element_detection_bp.route('/status', methods=['GET'])
def element_detection_status():
    """Statut de la détection d'éléments."""
    return jsonify({
        'status': 'disabled',
        'message': 'Détection d\\'éléments désactivée en mode allégé'
    })
''',
        
        "analytics.py": '''"""
API Analytics (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

analytics_bp = Blueprint('analytics', __name__)

@analytics_bp.route('/stats', methods=['GET'])
def get_stats():
    """Statistiques (placeholder)."""
    return jsonify({
        'workflows': 0,
        'executions': 0,
        'message': 'Analytics désactivées en mode allégé'
    })
''',
        
        "templates.py": '''"""
API Templates (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

templates_bp = Blueprint('templates', __name__)

@templates_bp.route('/', methods=['GET'])
def list_templates():
    """Liste des templates."""
    return jsonify([])

@templates_bp.route('/<template_id>', methods=['GET'])
def get_template(template_id):
    """Récupère un template."""
    return jsonify({'error': 'Template non trouvé'}), 404
''',
        
        "executions.py": '''"""
API Executions (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

executions_bp = Blueprint('executions', __name__)

@executions_bp.route('/', methods=['GET'])
def list_executions():
    """Liste des exécutions."""
    return jsonify([])

@executions_bp.route('/<execution_id>', methods=['GET'])
def get_execution(execution_id):
    """Récupère une exécution."""
    return jsonify({'error': 'Exécution non trouvée'}), 404
''',
        
        "import_export.py": '''"""
API Import/Export (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

from flask import Blueprint, jsonify

import_export_bp = Blueprint('import_export', __name__)

@import_export_bp.route('/export/<workflow_id>', methods=['GET'])
def export_workflow(workflow_id):
    """Export d'un workflow."""
    return jsonify({'error': 'Export non implémenté en mode allégé'}), 501

@import_export_bp.route('/import', methods=['POST'])
def import_workflow():
    """Import d'un workflow."""
    return jsonify({'error': 'Import non implémenté en mode allégé'}), 501
''',
        
        "websocket_handlers.py": '''"""
Gestionnaires WebSocket (version minimale)

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

# Placeholder pour les gestionnaires WebSocket
# En mode allégé, les WebSockets sont désactivés

def init_websocket_handlers():
    """Initialise les gestionnaires WebSocket (placeholder)."""
    pass
'''
    }
    
    # Créer les blueprints optionnels
    for filename, content in optional_blueprints.items():
        file_path = api_path / filename
        if not file_path.exists():
            file_path.write_text(content)
            print(f"✅ Créé api/{filename}")
        else:
            print(f"⚠️  api/{filename} existe déjà")

def create_test_script():
    """Crée un script de test pour vérifier le backend."""
    print_subsection("Création du Script de Test")
    
    test_script = '''#!/usr/bin/env python3
"""
Test du Backend Visual Workflow Builder

Auteur : Dom, Alice, Kiro - 08 janvier 2026
"""

import requests
import json
import time
import sys

def test_backend(base_url="http://localhost:5002"):
    """Test le backend VWB."""
    print(f"🧪 Test du backend: {base_url}")
    
    # Test 1: Health check
    try:
        response = requests.get(f"{base_url}/health", timeout=5)
        if response.status_code == 200:
            print("✅ Health check OK")
            print(f"   Réponse: {response.json()}")
        else:
            print(f"❌ Health check échoué: {response.status_code}")
            return False
    except Exception as e:
        print(f"❌ Erreur health check: {e}")
        return False
    
    # Test 2: Liste des workflows
    try:
        response = requests.get(f"{base_url}/api/workflows", timeout=5)
        if response.status_code == 200:
            workflows = response.json()
            print(f"✅ Liste workflows OK ({len(workflows)} workflows)")
        else:
            print(f"❌ Liste workflows échoué: {response.status_code}")
            return False
    except Exception as e:
        print(f"❌ Erreur liste workflows: {e}")
        return False
    
    # Test 3: Création d'un workflow
    try:
        test_workflow = {
            "name": "Test Workflow",
            "description": "Workflow de test automatique",
            "created_by": "test_script"
        }
        
        response = requests.post(
            f"{base_url}/api/workflows",
            json=test_workflow,
            timeout=5
        )
        
        if response.status_code == 201:
            created_workflow = response.json()
            workflow_id = created_workflow['id']
            print(f"✅ Création workflow OK (ID: {workflow_id})")
            
            # Test 4: Récupération du workflow créé
            response = requests.get(f"{base_url}/api/workflows/{workflow_id}", timeout=5)
            if response.status_code == 200:
                print("✅ Récupération workflow OK")
                return True
            else:
                print(f"❌ Récupération workflow échoué: {response.status_code}")
                return False
        else:
            print(f"❌ Création workflow échoué: {response.status_code}")
            print(f"   Réponse: {response.text}")
            return False
    except Exception as e:
        print(f"❌ Erreur création workflow: {e}")
        return False

if __name__ == "__main__":
    print("🚀 Test automatique du backend VWB")
    print("=" * 40)
    
    # Attendre que le serveur soit prêt
    print("⏳ Attente du démarrage du serveur...")
    time.sleep(2)
    
    success = test_backend()
    
    if success:
        print("\\n✅ Tous les tests sont passés!")
        sys.exit(0)
    else:
        print("\\n❌ Certains tests ont échoué")
        sys.exit(1)
'''
    
    test_path = Path("visual_workflow_builder/backend/test_backend.py")
    test_path.write_text(test_script)
    print(f"✅ Script de test créé: {test_path}")

def main():
    """Fonction principale de correction."""
    print_section("CORRECTION DES ERREURS BACKEND VWB")
    print("Auteur : Dom, Alice, Kiro - 08 janvier 2026")
    
    # Vérifier que nous sommes dans le bon répertoire
    if not Path("visual_workflow_builder/backend").exists():
        print("❌ Erreur: Répertoire backend introuvable")
        print("   Exécutez ce script depuis la racine du projet")
        return
    
    # Corrections
    fix_missing_api_modules()
    fix_services_directory()
    fix_optional_blueprints()
    create_test_script()
    
    print_section("CORRECTIONS TERMINÉES")
    print("✅ Tous les modules manquants ont été créés")
    print("✅ Le backend devrait maintenant démarrer sans erreurs")
    print("")
    print("Pour tester le backend:")
    print("1. cd visual_workflow_builder/backend")
    print("2. ./start_fast.sh")
    print("3. Dans un autre terminal: python test_backend.py")

if __name__ == "__main__":
    main()