nl_rpa_interface.py

#!/usr/bin/env python3
"""
Natural Language RPA Interface
- Provides a natural language interface to the integrated RPA system
- Allows controlling automation workflows through simple text commands
- Uses local LLM for NL understanding and command generation
"""

import os
import sys
import json
import time
import logging
import argparse
import subprocess
from datetime import datetime
from typing import Dict, List, Any, Optional, Union

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("logs/nl_rpa_interface.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger('nl-rpa-interface')

class NLRPAInterface:
    def __init__(self, integrated_rpa_path: str = None):
        """
        Initialize the Natural Language RPA Interface
        
        Args:
            integrated_rpa_path: Path to integrated RPA automation script
        """
        self.integrated_rpa_path = integrated_rpa_path or "/Users/yacinebenhamou/Desktop/integrated_rpa_automation.py"
        self.workflows_dir = "/Users/yacinebenhamou/Desktop/output/integrated"
        self.nl_examples = self._load_examples()
        
        # Create output directories
        os.makedirs("output/nl_interface", exist_ok=True)
        os.makedirs("logs", exist_ok=True)
        
        # Check if integrated RPA script exists
        if not os.path.exists(self.integrated_rpa_path):
            logger.error(f"Integrated RPA script not found: {self.integrated_rpa_path}")
            raise FileNotFoundError(f"Integrated RPA script not found: {self.integrated_rpa_path}")
        
        logger.info(f"Initialized Natural Language RPA Interface")
        logger.info(f"Integrated RPA Path: {self.integrated_rpa_path}")
    
    def _load_examples(self) -> List[Dict[str, str]]:
        """
        Load example natural language commands and their mappings
        
        Returns:
            List of example commands and their mappings
        """
        return [
            {
                "nl_command": "Run OCR detection on the desktop",
                "workflow": "ocr_detection",
                "system": "kgg",
                "params": {"target": "desktop"}
            },
            {
                "nl_command": "Automate browser using Playwright",
                "workflow": "playwright",
                "system": "optimusprime",
                "params": {"headless": "false"}
            },
            {
                "nl_command": "Run UI navigation on system preferences",
                "workflow": "ui_navigation",
                "system": "kgg",
                "params": {"target_app": "system_preferences"}
            },
            {
                "nl_command": "Run predictive analytics on previous data",
                "workflow": "predictive_analytics",
                "system": "kgg",
                "params": {"data_source": "previous_steps"}
            },
            {
                "nl_command": "Run Selenium tests with Chrome",
                "workflow": "selenium",
                "system": "optimusprime",
                "params": {"browser": "chrome"}
            },
            {
                "nl_command": "Run the end-to-end workflow",
                "workflow": "orchestrator",
                "system": "optimusprime",
                "params": {}
            }
        ]
    
    def process_nl_command(self, nl_command: str) -> Dict[str, Any]:
        """
        Process a natural language command and convert it to a workflow configuration
        
        Args:
            nl_command: Natural language command
            
        Returns:
            Workflow configuration
        """
        try:
            logger.info(f"Processing natural language command: {nl_command}")
            
            # Try to match with examples first
            for example in self.nl_examples:
                if example["nl_command"].lower() in nl_command.lower():
                    logger.info(f"Matched example: {example['nl_command']}")
                    return {
                        "system": example["system"],
                        "workflow": example["workflow"],
                        "params": example["params"]
                    }
            
            # If no exact match, use LLM to parse the command
            return self._parse_with_llm(nl_command)
        except Exception as e:
            logger.error(f"Error processing NL command: {str(e)}")
            logger.exception("Exception details:")
            return {}
    
    def _parse_with_llm(self, nl_command: str) -> Dict[str, Any]:
        """
        Parse a natural language command using a local LLM
        
        Args:
            nl_command: Natural language command
            
        Returns:
            Parsed command as a workflow configuration
        """
        try:
            # Check if OLLAMA is available
            ollama_available = self._check_ollama()
            
            if ollama_available:
                # Use OLLAMA for local LLM processing
                return self._parse_with_ollama(nl_command)
            else:
                # Fallback to rule-based parsing
                return self._rule_based_parsing(nl_command)
        except Exception as e:
            logger.error(f"Error parsing with LLM: {str(e)}")
            logger.exception("Exception details:")
            return self._rule_based_parsing(nl_command)
    
    def _check_ollama(self) -> bool:
        """
        Check if OLLAMA is available
        
        Returns:
            True if OLLAMA is available, False otherwise
        """
        try:
            result = subprocess.run(["ollama", "list"], capture_output=True, text=True, timeout=5)
            return result.returncode == 0
        except Exception:
            return False
    
    def _parse_with_ollama(self, nl_command: str) -> Dict[str, Any]:
        """
        Parse a natural language command using OLLAMA
        
        Args:
            nl_command: Natural language command
            
        Returns:
            Parsed command as a workflow configuration
        """
        try:
            # Prepare the prompt for OLLAMA
            prompt = f"""
            You are an AI assistant that converts natural language commands into structured workflow configurations.
            Please convert the following command into a JSON object with the following structure:
            {{"system": "kgg" or "optimusprime", "workflow": "workflow_name", "params": {{"param1": "value1", ...}}}}
            
            Available systems and workflows:
            - KGG: ocr_detection, ui_navigation, predictive_analytics, advanced_analytics
            - OptimusPrime: playwright, selenium, ocr, orchestrator
            
            Command: {nl_command}
            
            JSON output:
            """
            
            # Run OLLAMA with the prompt
            result = subprocess.run(["ollama", "run", "mistral", prompt], capture_output=True, text=True, timeout=10)
            
            # Parse the output
            output = result.stdout.strip()
            
            # Extract JSON from the output
            json_start = output.find('{')
            json_end = output.rfind('}')
            
            if json_start >= 0 and json_end >= 0:
                json_str = output[json_start:json_end+1]
                parsed = json.loads(json_str)
                logger.info(f"Parsed with OLLAMA: {parsed}")
                return parsed
            else:
                logger.warning(f"Could not extract JSON from OLLAMA output: {output}")
                return self._rule_based_parsing(nl_command)
        except Exception as e:
            logger.error(f"Error parsing with OLLAMA: {str(e)}")
            logger.exception("Exception details:")
            return self._rule_based_parsing(nl_command)
    
    def _rule_based_parsing(self, nl_command: str) -> Dict[str, Any]:
        """
        Parse a natural language command using rule-based approach
        
        Args:
            nl_command: Natural language command
            
        Returns:
            Parsed command as a workflow configuration
        """
        nl_command = nl_command.lower()
        
        # Default to OCR detection if nothing else matches
        result = {
            "system": "kgg",
            "workflow": "ocr_detection",
            "params": {}
        }
        
        # Check for KGG workflows
        if "ocr" in nl_command:
            result["system"] = "kgg"
            result["workflow"] = "ocr_detection"
            if "desktop" in nl_command:
                result["params"]["target"] = "desktop"
        elif "ui" in nl_command or "navigation" in nl_command:
            result["system"] = "kgg"
            result["workflow"] = "ui_navigation"
            if "system" in nl_command and "preferences" in nl_command:
                result["params"]["target_app"] = "system_preferences"
        elif "predict" in nl_command or "analytics" in nl_command:
            result["system"] = "kgg"
            result["workflow"] = "predictive_analytics"
            if "previous" in nl_command:
                result["params"]["data_source"] = "previous_steps"
        
        # Check for OptimusPrime workflows
        elif "playwright" in nl_command or "browser" in nl_command:
            result["system"] = "optimusprime"
            result["workflow"] = "playwright"
            if "headless" in nl_command:
                result["params"]["headless"] = "true"
            else:
                result["params"]["headless"] = "false"
        elif "selenium" in nl_command or "test" in nl_command:
            result["system"] = "optimusprime"
            result["workflow"] = "selenium"
            if "chrome" in nl_command:
                result["params"]["browser"] = "chrome"
            elif "firefox" in nl_command:
                result["params"]["browser"] = "firefox"
        elif "end-to-end" in nl_command or "orchestrator" in nl_command:
            result["system"] = "optimusprime"
            result["workflow"] = "orchestrator"
        
        logger.info(f"Rule-based parsing result: {result}")
        return result
    
    def create_workflow_config(self, nl_command: str) -> Dict[str, Any]:
        """
        Create a workflow configuration from a natural language command
        
        Args:
            nl_command: Natural language command
            
        Returns:
            Workflow configuration
        """
        try:
            # Process the natural language command
            parsed = self.process_nl_command(nl_command)
            
            if not parsed:
                logger.warning(f"Could not parse command: {nl_command}")
                return {}
            
            # Create a workflow configuration
            workflow_id = f"nl_workflow_{int(time.time())}"
            workflow_name = f"NL Workflow: {nl_command[:50]}"
            
            # Create a single-step workflow
            workflow_config = {
                "id": workflow_id,
                "name": workflow_name,
                "description": f"Workflow created from natural language command: {nl_command}",
                "version": "1.0.0",
                "created_at": datetime.now().isoformat(),
                "steps": [
                    {
                        "id": "step1",
                        "name": f"{parsed['system'].capitalize()} {parsed['workflow']}",
                        "system": parsed["system"],
                        "workflow": parsed["workflow"],
                        "params": parsed["params"],
                        "stop_on_failure": False
                    }
                ],
                "notification": {
                    "on_completion": True,
                    "on_failure": True,
                    "email": False
                },
                "reporting": {
                    "generate_html": True,
                    "include_screenshots": True,
                    "include_logs": True,
                    "include_analytics": True
                }
            }
            
            # Add workflow file for OptimusPrime if needed
            if parsed["system"] == "optimusprime":
                if parsed["workflow"] == "playwright":
                    workflow_config["steps"][0]["workflow_file"] = "samples/advanced_playwright_workflow.json"
                elif parsed["workflow"] == "selenium":
                    workflow_config["steps"][0]["workflow_file"] = "samples/enhanced_selenium_sample.side"
                elif parsed["workflow"] == "ocr":
                    workflow_config["steps"][0]["workflow_file"] = "samples/ocr_workflow_sample.json"
                elif parsed["workflow"] == "orchestrator":
                    workflow_config["steps"][0]["workflow_file"] = "samples/advanced_workflow_sample.json"
            
            logger.info(f"Created workflow configuration: {workflow_config}")
            return workflow_config
        except Exception as e:
            logger.error(f"Error creating workflow config: {str(e)}")
            logger.exception("Exception details:")
            return {}
    
    def run_workflow(self, workflow_config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Run a workflow using the integrated RPA automation
        
        Args:
            workflow_config: Workflow configuration
            
        Returns:
            Execution results
        """
        try:
            # Save workflow configuration to a temporary file
            workflow_file = f"output/nl_interface/{workflow_config['id']}.json"
            
            with open(workflow_file, "w") as f:
                json.dump(workflow_config, f, indent=2)
            
            logger.info(f"Saved workflow configuration to {workflow_file}")
            
            # Run the integrated RPA automation
            cmd = [
                "bash", "-c",
                f"cd /Users/yacinebenhamou/Desktop && ./run_integrated_rpa.sh --workflow {workflow_file}"
            ]
            
            logger.info(f"Running command: {cmd[-1]}")
            result = subprocess.run(cmd, capture_output=True, text=True)
            
            # Parse output
            output = result.stdout
            error = result.stderr
            success = result.returncode == 0
            
            logger.info(f"Workflow execution completed with status: {success}")
            
            # Find the latest report
            latest_report = None
            try:
                report_cmd = ["bash", "-c", "ls -t /Users/yacinebenhamou/Desktop/output/integrated/*_report.html 2>/dev/null | head -1"]
                report_result = subprocess.run(report_cmd, capture_output=True, text=True)
                latest_report = report_result.stdout.strip()
            except Exception:
                pass
            
            return {
                "success": success,
                "output": output,
                "error": error,
                "workflow": workflow_config["id"],
                "report": latest_report
            }
        except Exception as e:
            logger.error(f"Error running workflow: {str(e)}")
            logger.exception("Exception details:")
            return {
                "success": False,
                "error": str(e),
                "workflow": workflow_config.get("id", "unknown")
            }
    
    def process_command(self, nl_command: str) -> Dict[str, Any]:
        """
        Process a natural language command and run the corresponding workflow
        
        Args:
            nl_command: Natural language command
            
        Returns:
            Execution results
        """
        try:
            logger.info(f"Processing command: {nl_command}")
            
            # Create workflow configuration
            workflow_config = self.create_workflow_config(nl_command)
            
            if not workflow_config:
                return {
                    "success": False,
                    "error": f"Could not create workflow configuration for command: {nl_command}"
                }
            
            # Run workflow
            return self.run_workflow(workflow_config)
        except Exception as e:
            logger.error(f"Error processing command: {str(e)}")
            logger.exception("Exception details:")
            return {
                "success": False,
                "error": str(e)
            }

def main():
    # Parse command line arguments
    parser = argparse.ArgumentParser(description="Natural Language RPA Interface")
    parser.add_argument("--command", help="Natural language command")
    parser.add_argument("--interactive", action="store_true", help="Run in interactive mode")
    parser.add_argument("--log-level", help="Logging level", choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], default="INFO")
    args = parser.parse_args()
    
    # Set up logging level
    numeric_level = getattr(logging, args.log_level.upper(), None)
    if not isinstance(numeric_level, int):
        raise ValueError(f"Invalid log level: {args.log_level}")
    logging.getLogger('nl-rpa-interface').setLevel(numeric_level)
    
    # Initialize interface
    interface = NLRPAInterface()
    
    # Process command or run in interactive mode
    if args.command:
        # Process single command
        result = interface.process_command(args.command)
        print(json.dumps(result, indent=2))
        return 0 if result["success"] else 1
    elif args.interactive:
        # Run in interactive mode
        print("Natural Language RPA Interface")
        print("Type 'exit' or 'quit' to exit")
        print("Type 'help' for example commands")
        print()
        
        while True:
            try:
                command = input("Enter command: ")
                
                if command.lower() in ["exit", "quit"]:
                    break
                elif command.lower() == "help":
                    print("\nExample commands:")
                    for example in interface.nl_examples:
                        print(f"- {example['nl_command']}")
                    print()
                    continue
                
                print("Processing command...")
                result = interface.process_command(command)
                
                if result["success"]:
                    print("\nCommand executed successfully!")
                    if "report" in result and result["report"]:
                        print(f"Report: {result['report']}")
                else:
                    print("\nCommand execution failed!")
                    print(f"Error: {result.get('error', 'Unknown error')}")
                
                print()
            except KeyboardInterrupt:
                print("\nExiting...")
                break
            except Exception as e:
                print(f"\nError: {str(e)}")
        
        return 0
    else:
        # Show help
        parser.print_help()
        return 0

if __name__ == "__main__":
    sys.exit(main())