predictive_maintenance.py

#!/usr/bin/env python3
"""
Predictive Maintenance Module for RPA Systems
- Analyzes automation logs to predict potential failures
- Uses machine learning for anomaly detection and trend analysis
- Provides recommendations for maintenance and optimization
"""

import os
import sys
import json
import glob
import logging
import argparse
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional, Union, Tuple

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

class PredictiveMaintenance:
    def __init__(self, logs_dir: str = None, output_dir: str = None):
        """
        Initialize the Predictive Maintenance module
        
        Args:
            logs_dir: Directory containing RPA logs
            output_dir: Directory for output files
        """
        self.logs_dir = logs_dir or "/Users/yacinebenhamou/Desktop/logs"
        self.output_dir = output_dir or "/Users/yacinebenhamou/Desktop/output/maintenance"
        
        # Create output directory
        os.makedirs(self.output_dir, exist_ok=True)
        
        # Initialize ML models
        self.models = {}
        
        logger.info(f"Initialized Predictive Maintenance module")
        logger.info(f"Logs directory: {self.logs_dir}")
        logger.info(f"Output directory: {self.output_dir}")
    
    def collect_logs(self, days: int = 7) -> pd.DataFrame:
        """
        Collect and parse logs from the last N days
        
        Args:
            days: Number of days to look back
            
        Returns:
            DataFrame containing parsed log data
        """
        try:
            logger.info(f"Collecting logs from the last {days} days")
            
            # Calculate the cutoff date
            cutoff_date = datetime.now() - timedelta(days=days)
            
            # Find all log files
            log_files = []
            for ext in ["*.log", "*.txt"]:
                log_files.extend(glob.glob(os.path.join(self.logs_dir, ext)))
                log_files.extend(glob.glob(os.path.join(self.logs_dir, "**", ext), recursive=True))
            
            # Filter log files by modification time
            recent_log_files = []
            for log_file in log_files:
                try:
                    mtime = datetime.fromtimestamp(os.path.getmtime(log_file))
                    if mtime >= cutoff_date:
                        recent_log_files.append(log_file)
                except Exception as e:
                    logger.warning(f"Error checking file {log_file}: {str(e)}")
            
            logger.info(f"Found {len(recent_log_files)} log files from the last {days} days")
            
            # Parse log files
            log_data = []
            for log_file in recent_log_files:
                try:
                    with open(log_file, "r") as f:
                        for line in f:
                            try:
                                # Parse log line
                                parsed = self._parse_log_line(line)
                                if parsed:
                                    parsed["source"] = os.path.basename(log_file)
                                    log_data.append(parsed)
                            except Exception as e:
                                logger.debug(f"Error parsing line in {log_file}: {str(e)}")
                except Exception as e:
                    logger.warning(f"Error reading file {log_file}: {str(e)}")
            
            # Convert to DataFrame
            if log_data:
                df = pd.DataFrame(log_data)
                logger.info(f"Collected {len(df)} log entries")
                return df
            else:
                logger.warning("No log data collected")
                return pd.DataFrame()
        except Exception as e:
            logger.error(f"Error collecting logs: {str(e)}")
            logger.exception("Exception details:")
            return pd.DataFrame()
    
    def _parse_log_line(self, line: str) -> Optional[Dict[str, Any]]:
        """
        Parse a log line into a structured format
        
        Args:
            line: Log line to parse
            
        Returns:
            Dictionary with parsed log data or None if parsing failed
        """
        try:
            # Check if line is empty
            if not line.strip():
                return None
            
            # Try to parse timestamp
            timestamp = None
            level = "INFO"
            component = "unknown"
            message = line.strip()
            
            # Common timestamp format: 2025-04-29 01:10:40,071
            if " - " in line:
                parts = line.split(" - ", 3)
                if len(parts) >= 2:
                    try:
                        timestamp = datetime.strptime(parts[0].strip(), "%Y-%m-%d %H:%M:%S,%f")
                    except ValueError:
                        try:
                            timestamp = datetime.strptime(parts[0].strip(), "%Y-%m-%d %H:%M:%S")
                        except ValueError:
                            timestamp = None
                    
                    if len(parts) >= 3:
                        component = parts[1].strip()
                        level = parts[2].strip()
                        
                        if len(parts) >= 4:
                            message = parts[3].strip()
                        else:
                            message = ""
            
            # If timestamp parsing failed, try to find a timestamp in the line
            if not timestamp:
                import re
                timestamp_match = re.search(r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}', line)
                if timestamp_match:
                    try:
                        timestamp_str = timestamp_match.group(0)
                        if 'T' in timestamp_str:
                            timestamp = datetime.strptime(timestamp_str, "%Y-%m-%dT%H:%M:%S")
                        else:
                            timestamp = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S")
                    except ValueError:
                        timestamp = None
            
            # If still no timestamp, use current time
            if not timestamp:
                timestamp = datetime.now()
            
            # Determine log level
            for lvl in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]:
                if lvl in line.upper():
                    level = lvl
                    break
            
            # Extract component name
            if "RPA" in line or "rpa" in line:
                component = "RPA"
            elif "OCR" in line or "ocr" in line:
                component = "OCR"
            elif "PLAYWRIGHT" in line or "playwright" in line:
                component = "PLAYWRIGHT"
            elif "SELENIUM" in line or "selenium" in line:
                component = "SELENIUM"
            elif "ORCHESTRATOR" in line or "orchestrator" in line:
                component = "ORCHESTRATOR"
            
            # Check for error indicators
            is_error = level in ["ERROR", "CRITICAL"] or "error" in line.lower() or "exception" in line.lower() or "failed" in line.lower()
            
            # Check for performance indicators
            duration_match = re.search(r'(took|duration|elapsed)[:\s]+(\d+(\.\d+)?)[\s]*(ms|s|seconds|milliseconds)', line.lower())
            duration = None
            if duration_match:
                try:
                    duration_value = float(duration_match.group(2))
                    duration_unit = duration_match.group(4)
                    
                    # Convert to milliseconds
                    if duration_unit in ["s", "seconds"]:
                        duration = duration_value * 1000
                    else:
                        duration = duration_value
                except Exception:
                    duration = None
            
            return {
                "timestamp": timestamp,
                "level": level,
                "component": component,
                "message": message,
                "is_error": is_error,
                "duration": duration
            }
        except Exception as e:
            logger.debug(f"Error parsing log line: {str(e)}")
            return None
    
    def analyze_logs(self, df: pd.DataFrame) -> Dict[str, Any]:
        """
        Analyze log data to identify patterns and anomalies
        
        Args:
            df: DataFrame containing log data
            
        Returns:
            Dictionary with analysis results
        """
        try:
            if df.empty:
                logger.warning("No log data to analyze")
                return {"error": "No log data to analyze"}
            
            logger.info(f"Analyzing {len(df)} log entries")
            
            # Initialize results
            results = {
                "error_rate": 0.0,
                "error_trend": "stable",
                "performance_trend": "stable",
                "component_health": {},
                "anomalies": [],
                "recommendations": []
            }
            
            # Calculate error rate
            error_count = df["is_error"].sum()
            total_count = len(df)
            error_rate = error_count / total_count if total_count > 0 else 0
            results["error_rate"] = error_rate
            
            # Analyze errors by component
            component_errors = df[df["is_error"]].groupby("component").size()
            component_total = df.groupby("component").size()
            
            for component in component_total.index:
                error_count = component_errors.get(component, 0)
                total_count = component_total.get(component, 0)
                component_error_rate = error_count / total_count if total_count > 0 else 0
                
                health_score = 1.0 - component_error_rate
                health_status = "good"
                if health_score < 0.9:
                    health_status = "warning"
                if health_score < 0.7:
                    health_status = "critical"
                
                results["component_health"][component] = {
                    "error_rate": component_error_rate,
                    "health_score": health_score,
                    "status": health_status,
                    "log_count": int(total_count),
                    "error_count": int(error_count)
                }
            
            # Analyze performance (if duration data available)
            if "duration" in df.columns and not df["duration"].isna().all():
                performance_data = df[~df["duration"].isna()]
                
                if not performance_data.empty:
                    # Calculate performance metrics by component
                    for component in performance_data["component"].unique():
                        component_perf = performance_data[performance_data["component"] == component]
                        
                        if not component_perf.empty:
                            avg_duration = component_perf["duration"].mean()
                            max_duration = component_perf["duration"].max()
                            
                            if component in results["component_health"]:
                                results["component_health"][component]["avg_duration_ms"] = avg_duration
                                results["component_health"][component]["max_duration_ms"] = max_duration
            
            # Generate recommendations
            for component, health in results["component_health"].items():
                if health["status"] == "critical":
                    results["recommendations"].append({
                        "component": component,
                        "priority": "high",
                        "message": f"Critical error rate in {component} component ({health['error_rate']:.1%}). Immediate investigation required."
                    })
                elif health["status"] == "warning":
                    results["recommendations"].append({
                        "component": component,
                        "priority": "medium",
                        "message": f"Elevated error rate in {component} component ({health['error_rate']:.1%}). Review recent changes and logs."
                    })
                
                if health.get("avg_duration_ms", 0) > 5000:  # More than 5 seconds
                    results["recommendations"].append({
                        "component": component,
                        "priority": "medium",
                        "message": f"Slow performance in {component} component (avg: {health['avg_duration_ms']:.0f}ms). Consider optimization."
                    })
            
            # Overall system health
            if error_rate > 0.3:  # More than 30% errors
                results["system_health"] = "critical"
                results["recommendations"].append({
                    "component": "system",
                    "priority": "high",
                    "message": f"System-wide high error rate ({error_rate:.1%}). Consider rolling back recent changes."
                })
            elif error_rate > 0.1:  # More than 10% errors
                results["system_health"] = "warning"
                results["recommendations"].append({
                    "component": "system",
                    "priority": "medium",
                    "message": f"System-wide elevated error rate ({error_rate:.1%}). Monitor closely for degradation."
                })
            else:
                results["system_health"] = "good"
            
            logger.info(f"Analysis complete. System health: {results['system_health']}")
            return results
        except Exception as e:
            logger.error(f"Error analyzing logs: {str(e)}")
            logger.exception("Exception details:")
            return {"error": str(e)}
    
    def generate_report(self, analysis_results: Dict[str, Any]) -> str:
        """
        Generate an HTML report from analysis results
        
        Args:
            analysis_results: Results from log analysis
            
        Returns:
            Path to the generated HTML report
        """
        try:
            if "error" in analysis_results:
                logger.warning(f"Cannot generate report: {analysis_results['error']}")
                return ""
            
            # Generate timestamp
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            report_file = os.path.join(self.output_dir, f"maintenance_report_{timestamp}.html")
            
            # Generate HTML
            html = f"""
            <!DOCTYPE html>
            <html>
            <head>
                <title>RPA System Predictive Maintenance Report</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; }}
                    .good {{ color: green; }}
                    .warning {{ color: orange; }}
                    .critical {{ color: red; }}
                    .component {{ margin-bottom: 15px; border: 1px solid #ddd; padding: 15px; border-radius: 5px; }}
                    .component-good {{ background-color: #f0fff0; }}
                    .component-warning {{ background-color: #fffaf0; }}
                    .component-critical {{ background-color: #fff0f0; }}
                    .recommendation {{ margin-bottom: 10px; padding: 10px; border-radius: 5px; }}
                    .priority-high {{ background-color: #ffebee; border-left: 5px solid #f44336; }}
                    .priority-medium {{ background-color: #fff8e1; border-left: 5px solid #ffc107; }}
                    .priority-low {{ background-color: #e8f5e9; border-left: 5px solid #4caf50; }}
                </style>
            </head>
            <body>
                <h1>RPA System Predictive Maintenance Report</h1>
                <p><strong>Generated:</strong> {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</p>
                
                <div class="summary">
                    <h2>System Health Summary</h2>
                    <p><strong>Overall Health:</strong> <span class="{analysis_results['system_health']}">{analysis_results['system_health'].upper()}</span></p>
                    <p><strong>Error Rate:</strong> {analysis_results['error_rate']:.1%}</p>
                </div>
                
                <h2>Component Health</h2>
            """
            
            # Add component health sections
            for component, health in analysis_results["component_health"].items():
                html += f"""
                <div class="component component-{health['status']}">
                    <h3>{component}</h3>
                    <p><strong>Health Status:</strong> <span class="{health['status']}">{health['status'].upper()}</span></p>
                    <p><strong>Error Rate:</strong> {health['error_rate']:.1%} ({health['error_count']} errors out of {health['log_count']} logs)</p>
                """
                
                if "avg_duration_ms" in health:
                    html += f"""
                    <p><strong>Average Duration:</strong> {health['avg_duration_ms']:.0f}ms</p>
                    <p><strong>Maximum Duration:</strong> {health['max_duration_ms']:.0f}ms</p>
                    """
                
                html += "</div>\n"
            
            # Add recommendations
            if analysis_results["recommendations"]:
                html += f"""
                <h2>Recommendations</h2>
                """
                
                for rec in analysis_results["recommendations"]:
                    html += f"""
                    <div class="recommendation priority-{rec['priority']}">
                        <p><strong>{rec['component']}:</strong> {rec['message']}</p>
                        <p><strong>Priority:</strong> {rec['priority'].upper()}</p>
                    </div>
                    """
            
            # Close HTML
            html += f"""
            </body>
            </html>
            """
            
            # Write to file
            with open(report_file, "w") as f:
                f.write(html)
            
            logger.info(f"Generated maintenance report: {report_file}")
            return report_file
        except Exception as e:
            logger.error(f"Error generating report: {str(e)}")
            logger.exception("Exception details:")
            return ""
    
    def run_maintenance_check(self, days: int = 7) -> Dict[str, Any]:
        """
        Run a complete maintenance check
        
        Args:
            days: Number of days to analyze
            
        Returns:
            Dictionary with results and report path
        """
        try:
            logger.info(f"Running maintenance check for the last {days} days")
            
            # Collect logs
            df = self.collect_logs(days=days)
            
            if df.empty:
                return {
                    "success": False,
                    "error": "No log data collected"
                }
            
            # Analyze logs
            analysis_results = self.analyze_logs(df)
            
            if "error" in analysis_results:
                return {
                    "success": False,
                    "error": analysis_results["error"]
                }
            
            # Generate report
            report_file = self.generate_report(analysis_results)
            
            return {
                "success": True,
                "system_health": analysis_results["system_health"],
                "error_rate": analysis_results["error_rate"],
                "component_count": len(analysis_results["component_health"]),
                "recommendation_count": len(analysis_results["recommendations"]),
                "report_file": report_file
            }
        except Exception as e:
            logger.error(f"Error running maintenance check: {str(e)}")
            logger.exception("Exception details:")
            return {
                "success": False,
                "error": str(e)
            }

def main():
    # Parse command line arguments
    parser = argparse.ArgumentParser(description="Predictive Maintenance for RPA Systems")
    parser.add_argument("--days", type=int, default=7, help="Number of days to analyze")
    parser.add_argument("--logs-dir", help="Directory containing log files")
    parser.add_argument("--output-dir", help="Directory for output files")
    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('predictive-maintenance').setLevel(numeric_level)
    
    # Initialize maintenance module
    maintenance = PredictiveMaintenance(
        logs_dir=args.logs_dir,
        output_dir=args.output_dir
    )
    
    # Run maintenance check
    result = maintenance.run_maintenance_check(days=args.days)
    
    if result["success"]:
        print(f"\nMaintenance check completed successfully!")
        print(f"System Health: {result['system_health'].upper()}")
        print(f"Error Rate: {result['error_rate']:.1%}")
        print(f"Components Analyzed: {result['component_count']}")
        print(f"Recommendations: {result['recommendation_count']}")
        
        if result["report_file"]:
            print(f"\nReport generated: {result['report_file']}")
            
            # Try to open the report
            try:
                import subprocess
                subprocess.run(["open", result["report_file"]])
            except Exception:
                pass
        
        return 0
    else:
        print(f"\nMaintenance check failed: {result.get('error', 'Unknown error')}")
        return 1

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