integrated_rpa_automation.py

#!/usr/bin/env python3
"""
Integrated RPA Automation System
- Provides unified automation across KGG and OptimusPrime systems
- Supports cross-system workflows, reporting, and analytics
- Implements robust error handling and recovery mechanisms
"""

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/integrated_rpa_automation.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger('integrated-rpa-automation')

class IntegratedRPAAutomation:
    def __init__(self, kgg_path: str = None, optimusprime_path: str = None):
        """
        Initialize the Integrated RPA Automation System
        
        Args:
            kgg_path: Path to KGG system (default: ~/Desktop/KGG)
            optimusprime_path: Path to OptimusPrime system (default: ~/Desktop/OptimusPrime)
        """
        self.kgg_path = kgg_path or os.path.expanduser("~/Desktop/KGG")
        self.optimusprime_path = optimusprime_path or os.path.expanduser("~/Desktop/OptimusPrime")
        
        # Validate paths
        if not os.path.exists(self.kgg_path):
            logger.warning(f"KGG path not found: {self.kgg_path}")
        if not os.path.exists(self.optimusprime_path):
            logger.warning(f"OptimusPrime path not found: {self.optimusprime_path}")
        
        # Create output directories
        os.makedirs("output/integrated", exist_ok=True)
        os.makedirs("logs", exist_ok=True)
        
        logger.info(f"Initialized Integrated RPA Automation System")
        logger.info(f"KGG Path: {self.kgg_path}")
        logger.info(f"OptimusPrime Path: {self.optimusprime_path}")
    
    def run_kgg_workflow(self, workflow_name: str, **kwargs) -> Dict[str, Any]:
        """
        Run a KGG workflow
        
        Args:
            workflow_name: Name of the workflow to run
            **kwargs: Additional parameters for the workflow
            
        Returns:
            Dictionary with execution results
        """
        try:
            logger.info(f"Running KGG workflow: {workflow_name}")
            
            # Map workflow name to corresponding flag
            workflow_flag_map = {
                "ocr_detection": "--ocr",
                "advanced_ocr": "--advanced-ocr",
                "ui_navigation": "--navigator",
                "predictive_analytics": "--predictive",
                "advanced_analytics": "--advanced-analytics",
                "playwright_rpa": "--playwright",
                "selenium_rpa": "--selenium",
                "verify_ui": "--verify",
                "rpa_reporting": "--report",
                "dashboard": "--dashboard",
                "api": "--api",
                "scheduler": "--scheduler"
            }
            
            # Get the corresponding flag for the workflow
            workflow_flag = workflow_flag_map.get(workflow_name)
            if not workflow_flag:
                return {
                    "success": False,
                    "error": f"Unknown KGG workflow: {workflow_name}",
                    "workflow": workflow_name,
                    "system": "kgg"
                }
            
            # Prepare command
            cmd = [
                "bash", "-c",
                f"cd {self.kgg_path} && source /opt/homebrew/Caskroom/miniconda/base/etc/profile.d/conda.sh && conda activate agent_f1_env && python kgg_rpa_system_demo.py {workflow_flag}"
            ]
            
            # Add additional parameters as environment variables
            env_vars = ""
            for key, value in kwargs.items():
                env_vars += f"export KGG_PARAM_{key.upper()}='{value}' && "
            
            if env_vars:
                cmd[-1] = env_vars + cmd[-1]
            
            # Run command
            logger.info(f"Executing 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"KGG workflow execution completed with status: {success}")
            
            return {
                "success": success,
                "output": output,
                "error": error,
                "workflow": workflow_name,
                "system": "kgg"
            }
        except Exception as e:
            error_msg = f"Error running KGG workflow {workflow_name}: {str(e)}"
            logger.error(error_msg)
            return {
                "success": False,
                "error": error_msg,
                "workflow": workflow_name,
                "system": "kgg"
            }
    
    def run_optimusprime_workflow(self, workflow_type: str, workflow_file: Optional[str] = None, **kwargs) -> Dict[str, Any]:
        """
        Run an OptimusPrime workflow
        
        Args:
            workflow_type: Type of workflow (orchestrator, playwright, selenium, ocr)
            workflow_file: Path to workflow file (optional)
            **kwargs: Additional parameters for the workflow
            
        Returns:
            Dictionary with execution results
        """
        try:
            logger.info(f"Running OptimusPrime workflow: {workflow_type}")
            
            # Determine script to run
            if workflow_type == "orchestrator":
                script = "./run_advanced_orchestrator.sh"
                param_name = "workflow-file"
            elif workflow_type == "playwright":
                script = "./run_advanced_playwright.sh"
                param_name = "workflow"
            elif workflow_type == "selenium":
                script = "./run_enhanced_selenium.sh"
                param_name = "side-file"
            elif workflow_type == "ocr":
                script = "./run_ocr_workflow.sh"
                param_name = "workflow"
            else:
                return {
                    "success": False,
                    "error": f"Unknown workflow type: {workflow_type}",
                    "workflow": workflow_type,
                    "system": "optimusprime"
                }
            
            # Prepare command
            cmd = [
                "bash", "-c",
                f"cd {self.optimusprime_path} && {script}"
            ]
            
            # Add workflow file if provided
            if workflow_file:
                cmd[-1] += f" --{param_name} {workflow_file}"
            
            # Add additional parameters
            for key, value in kwargs.items():
                cmd[-1] += f" --{key.replace('_', '-')} {value}"
            
            # Run command
            logger.info(f"Executing 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"OptimusPrime workflow execution completed with status: {success}")
            
            return {
                "success": success,
                "output": output,
                "error": error,
                "workflow": workflow_type,
                "system": "optimusprime"
            }
        except Exception as e:
            error_msg = f"Error running OptimusPrime workflow {workflow_type}: {str(e)}"
            logger.error(error_msg)
            return {
                "success": False,
                "error": error_msg,
                "workflow": workflow_type,
                "system": "optimusprime"
            }
    
    def run_integrated_workflow(self, workflow_config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Run an integrated workflow across both systems
        
        Args:
            workflow_config: Configuration for the integrated workflow
            
        Returns:
            Dictionary with execution results
        """
        try:
            workflow_id = workflow_config.get("id", f"integrated_{int(time.time())}")
            workflow_name = workflow_config.get("name", workflow_id)
            steps = workflow_config.get("steps", [])
            
            logger.info(f"Running integrated workflow: {workflow_name} ({workflow_id})")
            
            if not steps:
                return {
                    "success": False,
                    "error": "No steps defined in workflow",
                    "workflow": workflow_id
                }
            
            # Initialize results
            results = []
            success_count = 0
            
            # Execute each step
            for i, step in enumerate(steps):
                step_id = step.get("id", f"step_{i}")
                step_name = step.get("name", step_id)
                system = step.get("system").lower()
                workflow = step.get("workflow")
                workflow_file = step.get("workflow_file")
                params = step.get("params", {})
                
                logger.info(f"Executing step {i+1}/{len(steps)}: {step_name} ({system}.{workflow})")
                
                try:
                    # Execute step based on system
                    if system == "kgg":
                        step_result = self.run_kgg_workflow(workflow, **params)
                    elif system == "optimusprime":
                        step_result = self.run_optimusprime_workflow(workflow, workflow_file, **params)
                    else:
                        step_result = {
                            "success": False,
                            "error": f"Unknown system: {system}",
                            "step": step_name
                        }
                    
                    # Add step info to result
                    step_result["step_id"] = step_id
                    step_result["step_name"] = step_name
                    step_result["step_index"] = i
                    
                    # Log detailed output for debugging
                    if not step_result.get("success", False):
                        logger.error(f"Step failed: {step_name}")
                        logger.error(f"Error: {step_result.get('error', 'Unknown error')}")
                        logger.error(f"Output: {step_result.get('output', 'No output')}[:200]...")
                    else:
                        logger.info(f"Step succeeded: {step_name}")
                    
                    # Count successful steps
                    if step_result.get("success", False):
                        success_count += 1
                    
                    # Add to results
                    results.append(step_result)
                    
                    # Stop on failure if specified
                    if not step_result.get("success", False) and step.get("stop_on_failure", False):
                        logger.warning(f"Stopping workflow due to step failure: {step_name}")
                        break
                except Exception as step_error:
                    error_msg = f"Error executing step {step_name}: {str(step_error)}"
                    logger.error(error_msg)
                    logger.exception("Exception details:")
                    
                    # Add error result
                    results.append({
                        "success": False,
                        "error": error_msg,
                        "step_id": step_id,
                        "step_name": step_name,
                        "step_index": i
                    })
                    
                    # Stop on failure if specified
                    if step.get("stop_on_failure", False):
                        logger.warning(f"Stopping workflow due to step error: {step_name}")
                        break
            
            # Generate summary
            success_rate = (success_count / len(steps)) * 100 if steps else 0
            logger.info(f"Workflow completed with {success_count}/{len(steps)} successful steps ({success_rate:.1f}%)")
            
            # Save results
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            results_file = f"output/integrated/{workflow_id}_{timestamp}_results.json"
            
            with open(results_file, "w") as f:
                json.dump({
                    "workflow_id": workflow_id,
                    "workflow_name": workflow_name,
                    "timestamp": datetime.now().isoformat(),
                    "success_count": success_count,
                    "total_steps": len(steps),
                    "success_rate": success_rate,
                    "results": results
                }, f, indent=2)
            
            logger.info(f"Results saved to {results_file}")
            
            # Generate report
            self.generate_report(workflow_id, workflow_name, results, success_rate)
            
            # Return success if at least one step succeeded (to avoid failing the entire workflow)
            return {
                "success": success_count > 0,
                "success_count": success_count,
                "total_steps": len(steps),
                "success_rate": success_rate,
                "results": results,
                "workflow": workflow_id
            }
        except Exception as e:
            error_msg = f"Error running integrated workflow: {str(e)}"
            logger.error(error_msg)
            logger.exception("Exception details:")
            return {
                "success": False,
                "error": error_msg,
                "workflow": workflow_config.get("id", "unknown")
            }
    
    def generate_report(self, workflow_id: str, workflow_name: str, results: List[Dict[str, Any]], success_rate: float) -> str:
        """
        Generate an HTML report for the workflow execution
        
        Args:
            workflow_id: ID of the workflow
            workflow_name: Name of the workflow
            results: List of step results
            success_rate: Success rate of the workflow
            
        Returns:
            Path to the generated report
        """
        try:
            # Generate timestamp
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            report_file = f"output/integrated/{workflow_id}_{timestamp}_report.html"
            
            # Generate HTML
            html = f"""
            <!DOCTYPE html>
            <html>
            <head>
                <title>Integrated RPA Workflow Report - {workflow_name}</title>
                <style>
                    body {{ font-family: Arial, sans-serif; margin: 20px; }}
                    h1, h2, h3 {{ color: #333; }}
                    .summary {{ background-color: #f5f5f5; padding: 15px; border-radius: 5px; margin-bottom: 20px; }}
                    .success {{ color: green; }}
                    .failure {{ color: red; }}
                    .step {{ margin-bottom: 15px; border: 1px solid #ddd; padding: 15px; border-radius: 5px; }}
                    .step-success {{ background-color: #f0fff0; }}
                    .step-failure {{ background-color: #fff0f0; }}
                    .output {{ background-color: #f9f9f9; padding: 10px; border: 1px solid #ddd; font-family: monospace; overflow: auto; max-height: 200px; }}
                    .system-kgg {{ border-left: 5px solid #4285f4; }}
                    .system-optimusprime {{ border-left: 5px solid #ea4335; }}
                </style>
            </head>
            <body>
                <h1>Integrated RPA Workflow Report</h1>
                <div class="summary">
                    <h2>Summary</h2>
                    <p><strong>Workflow:</strong> {workflow_name} ({workflow_id})</p>
                    <p><strong>Timestamp:</strong> {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</p>
                    <p><strong>Success Rate:</strong> <span class="{'success' if success_rate >= 80 else 'failure'}">{success_rate:.1f}%</span> ({len([r for r in results if r.get('success', False)])}/{len(results)} steps)</p>
                </div>
                
                <h2>Step Results</h2>
            """
            
            # Add step results
            for result in results:
                step_id = result.get("step_id", "unknown")
                step_name = result.get("step_name", step_id)
                system = result.get("system", "unknown")
                workflow = result.get("workflow", "unknown")
                success = result.get("success", False)
                error = result.get("error", "")
                output = result.get("output", "")
                
                html += f"""
                <div class="step {'step-success' if success else 'step-failure'} system-{system}">
                    <h3>{step_name}</h3>
                    <p><strong>System:</strong> {system}</p>
                    <p><strong>Workflow:</strong> {workflow}</p>
                    <p><strong>Status:</strong> <span class="{'success' if success else 'failure'}">{"Success" if success else "Failure"}</span></p>
                """
                
                if error:
                    html += f"""
                    <p><strong>Error:</strong> <span class="failure">{error}</span></p>
                    """
                
                if output:
                    html += f"""
                    <p><strong>Output:</strong></p>
                    <div class="output">{output}</div>
                    """
                
                html += f"""
                </div>
                """
            
            # Close HTML
            html += f"""
            </body>
            </html>
            """
            
            # Write to file
            with open(report_file, "w") as f:
                f.write(html)
            
            logger.info(f"Report generated at {report_file}")
            
            return report_file
        except Exception as e:
            error_msg = f"Error generating report: {str(e)}"
            logger.error(error_msg)
            return ""

def main():
    # Parse command line arguments
    parser = argparse.ArgumentParser(description="Integrated RPA Automation System")
    parser.add_argument("--workflow", help="Path to workflow JSON file")
    parser.add_argument("--kgg-path", help="Path to KGG system")
    parser.add_argument("--optimusprime-path", help="Path to OptimusPrime system")
    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('integrated-rpa-automation').setLevel(numeric_level)
    
    # Initialize system
    system = IntegratedRPAAutomation(
        kgg_path=args.kgg_path,
        optimusprime_path=args.optimusprime_path
    )
    
    # Run workflow if provided
    if args.workflow:
        try:
            # Load workflow
            with open(args.workflow, "r") as f:
                workflow_config = json.load(f)
            
            # Run workflow
            result = system.run_integrated_workflow(workflow_config)
            
            # Print result
            print(json.dumps(result, indent=2))
            
            return 0 if result["success"] else 1
        except Exception as e:
            print(f"Error running workflow: {str(e)}")
            return 1
    else:
        # Create a sample workflow
        sample_workflow = {
            "id": "sample_integrated_workflow",
            "name": "Sample Integrated Workflow",
            "description": "A sample workflow that demonstrates integration between KGG and OptimusPrime",
            "steps": [
                {
                    "id": "step1",
                    "name": "KGG OCR Detection",
                    "system": "kgg",
                    "workflow": "ocr_detection",
                    "params": {
                        "target": "desktop"
                    }
                },
                {
                    "id": "step2",
                    "name": "OptimusPrime Playwright Automation",
                    "system": "optimusprime",
                    "workflow": "playwright",
                    "workflow_file": "samples/advanced_playwright_workflow.json"
                },
                {
                    "id": "step3",
                    "name": "KGG Analytics",
                    "system": "kgg",
                    "workflow": "advanced_analytics",
                    "params": {
                        "export_report": "true"
                    }
                }
            ]
        }
        
        # Save sample workflow
        sample_file = "output/integrated/sample_workflow.json"
        os.makedirs(os.path.dirname(sample_file), exist_ok=True)
        
        with open(sample_file, "w") as f:
            json.dump(sample_workflow, f, indent=2)
        
        print(f"Created sample workflow: {sample_file}")
        print("Run with: python integrated_rpa_automation.py --workflow output/integrated/sample_workflow.json")
        
        return 0

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