logs.py

"""
ETL Analyzer - AI-Powered Windows Event Trace Filter
====================================================

This tool analyzes Windows ETL files and extracts the most relevant information
for AI-assisted troubleshooting, reducing noise and context size.

Features:
- Parse Windows ETL files using Windows Performance Toolkit
- Intelligent event filtering based on severity and patterns
- Statistical analysis of event patterns
- Context-aware summarization
- Export filtered results for AI consumption

Requirements:
- Windows Performance Toolkit (WPT) installed
- Python packages: pandas, numpy, json, xml, re
"""

import os
import re
import json
import subprocess
import tempfile
import xml.etree.ElementTree as ET
from pathlib import Path
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional, Set
from dataclasses import dataclass, asdict
from collections import Counter, defaultdict
import pandas as pd
import numpy as np

@dataclass
class EventInfo:
    """Structured representation of an ETL event"""
    event_id: str
    provider: str
    level: str
    task: str
    opcode: str
    keyword: str
    message: str
    process_id: str
    thread_id: str
    relevance_score: float = 0.0
    category: str = "unknown"



class ETLAnalyzer:
    """Main ETL analysis class with AI-focused filtering"""

    def __init__(self, etl_file_path: str):
        self.etl_file_path = Path(etl_file_path)
        self.events: List[EventInfo] = []
        self.filtered_events: List[EventInfo] = []

        # Event filtering configurations
        self.high_priority_providers = {
            'Microsoft-Windows-Kernel-General',
            'Microsoft-Windows-Kernel-Process',
            'Microsoft-Windows-Application-Experience',
            'Microsoft-Windows-WinINet',
            'Microsoft-Windows-DNS-Client',
            'Microsoft-Windows-TCPIP',
            'Microsoft-Windows-NDIS-PacketCapture',
            'Microsoft-Windows-Security-Auditing',
            'Microsoft-Windows-DistributedCOM',
            'Microsoft-Windows-RPC',
            'Microsoft-Windows-WER-SystemErrorReporting'
        }

        self.critical_event_ids = {
            1074,  # System shutdown/restart
            6008,  # Unexpected shutdown
            41,    # System reboot without clean shutdown
            1001,  # Windows Error Reporting
            4624,  # Successful logon
            4625,  # Failed logon
            7034,  # Service crashed unexpectedly
            7031,  # Service terminated unexpectedly
            1000,  # Application error
            1002   # Application hang
        }

        self.noise_patterns = [
            r'.*QueryInterface.*',
            r'.*AddRef.*',
            r'.*Release.*',
            r'.*routine maintenance.*',
            r'.*successfully loaded.*',
            r'.*completed successfully.*',
            r'.*information.*verbose.*'
        ]

    def convert_etl_to_csv(self) -> Path:
        """Convert ETL file to CSV using Windows Performance Toolkit"""
        try:
            output_csv = self.etl_file_path.with_suffix('.csv')

            # Use wpa.exe or tracerpt.exe to convert ETL to readable format
            cmd = [
                'tracerpt.exe',
                str(self.etl_file_path),
                '-o', str(output_csv),
                '-of', 'CSV'
            ]

            print(f"Converting ETL file: {self.etl_file_path}")
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)

            if output_csv.exists():
                return output_csv
            else:
                raise FileNotFoundError("CSV conversion failed")

        except subprocess.CalledProcessError as e:
            print(f"ETL conversion failed: {e}")
            print("Make sure Windows Performance Toolkit is installed")
            raise
        except Exception as e:
            print(f"Error during ETL conversion: {e}")
            raise

    def parse_events_from_csv(self, csv_path: Path) -> List[EventInfo]:
      events = []
      
      try:
          # First, read the file to understand the structure
          with open(csv_path, 'r', encoding='utf-8-sig', errors='ignore') as f:
              lines = f.readlines()
          
          if len(lines) < 2:
              print("CSV file appears to be empty or has no data rows")
              return events
          
          # Parse header to understand column structure
          header_line = lines[0].strip()
          headers = [col.strip() for col in header_line.split(',')]
          
          # Find the index of 'User Data' column
          user_data_index = None
          for i, header in enumerate(headers):
              if 'User Data' in header:
                  user_data_index = i
                  break
          
          print(f"Found {len(headers)} columns in CSV header")
          print(f"User Data starts at column index: {user_data_index}")
          
          # Process each data line
          for line_num, line in enumerate(lines[1:], start=2):
              try:
                  line = line.strip()
                  if not line:
                      continue
                  
                  # Split the line by commas, but be careful with quoted values
                  import csv
                  from io import StringIO
                  
                  # Use csv reader to properly handle quoted fields
                  csv_reader = csv.reader(StringIO(line), quoting=csv.QUOTE_ALL)
                  row_data = next(csv_reader)
                  
                  # Ensure we have enough columns
                  if len(row_data) < len(headers):
                      # Pad with empty strings if needed
                      row_data.extend([''] * (len(headers) - len(row_data)))
                  
                  # Extract standard fields (before User Data)
                  standard_fields = row_data[:user_data_index] if user_data_index else row_data
                  
                  # Extract User Data fields (everything from User Data column onwards)
                  user_data_fields = row_data[user_data_index:] if user_data_index else []
                  
                  # Combine all user data fields into a single message
                  user_data_combined = ', '.join(str(field).strip('"') for field in user_data_fields if field.strip())
                  
                  # Create a dictionary for easier field access
                  row_dict = {}
                  for i, header in enumerate(headers[:user_data_index] if user_data_index else headers):
                      row_dict[header.strip()] = standard_fields[i] if i < len(standard_fields) else ''
                  
                  # Handle potential None values and type conversions safely
                  event_id = row_dict.get('Event ID', '0')
                  if not event_id or event_id.strip() == '':
                      event_id = '0'
                  
                  process_id = row_dict.get('PID', '0')
                  if not process_id or process_id.strip() == '':
                      process_id = '0'
                  
                  thread_id = row_dict.get('TID', '0')
                  if not thread_id or thread_id.strip() == '':
                      thread_id = '0'
                  
                  # Clean up field values (remove quotes and extra whitespace)
                  def clean_field(value):
                      if value is None:
                          return ''
                      return str(value).strip().strip('"').strip()
                  
                  event = EventInfo(
                      provider=clean_field(row_dict.get('Event Name', '')),
                      event_id=clean_field(event_id),
                      level=clean_field(row_dict.get('Level', 'Information')),
                      task=clean_field(row_dict.get('Task', '')),
                      opcode=clean_field(row_dict.get('Opcode', '0')),
                      keyword=clean_field(row_dict.get('Keyword', '')),
                      message=user_data_combined,  # All User Data fields combined
                      process_id=clean_field(process_id),
                      thread_id=clean_field(thread_id)
                  )
                  events.append(event)
                  
              except Exception as e:
                  print(f"Warning: Skipping malformed row {line_num}: {e}")
                  continue
                  
          print(f"Successfully parsed {len(events)} events from CSV")
          
      except Exception as e:
          print(f"Error parsing CSV: {e}")
          raise
          
      return events


    def calculate_relevance_score(self, event: EventInfo) -> float:
        """Calculate relevance score for an event (0-1 scale)"""
        score = 0.0

        # Base score by event level
        level_scores = {
            'Critical': 1.0,
            'Error': 0.9,
            'Warning': 0.7,
            'Information': 0.3,
            'Verbose': 0.1
        }
        score += level_scores.get(event.level, 0.3)

        # Boost for high-priority providers
        if event.provider in self.high_priority_providers:
            score += 0.3

        # Boost for critical event IDs
        if event.event_id in self.critical_event_ids:
            score += 0.4

        # Reduce score for noise patterns
        for pattern in self.noise_patterns:
            if re.search(pattern, event.message, re.IGNORECASE):
                score *= 0.5
                break

        # Boost for error-indicating keywords
        error_keywords = ['error', 'failed', 'exception', 'crash', 'timeout', 'denied', 'not found']
        for keyword in error_keywords:
            if keyword.lower() in event.message.lower():
                score += 0.2
                break

        return min(score, 1.0)

    def categorize_event(self, event: EventInfo) -> str:
        """Categorize events for better organization"""
        message_lower = event.message.lower()
        provider_lower = event.provider.lower()

        # Security events
        if 'security' in provider_lower or 'logon' in message_lower or 'authentication' in message_lower:
            return 'security'

        # Network events
        if any(net in provider_lower for net in ['tcpip', 'dns', 'wininet', 'network', 'transport', 'endpoint']):
            return 'network'

        # System events
        if any(sys in provider_lower for sys in ['kernel', 'system', 'boot']):
            return 'system'

        # Application events
        if any(app in provider_lower for app in ['application', 'app']):
            return 'application'

        # Performance events
        if any(perf in message_lower for perf in ['performance', 'slow', 'timeout', 'hang']):
            return 'performance'

        # Service events
        if 'service' in provider_lower or 'service' in message_lower:
            return 'service'

        return 'other'

    def analyze_events(self) -> Dict:
        """Perform comprehensive analysis of events"""
        if not self.events:
            return {}

        # Calculate relevance scores and categorize
        for event in self.events:
            event.relevance_score = self.calculate_relevance_score(event)
            event.category = self.categorize_event(event)

        # Statistical analysis
        analysis = {
            'total_events': len(self.events),
            'event_levels': Counter(event.level for event in self.events),
            'event_categories': Counter(event.category for event in self.events),
            'top_providers': Counter(event.provider for event in self.events).most_common(10),
            'critical_events': len([e for e in self.events if e.level in ['Critical', 'Error']]),
            'high_relevance_events': len([e for e in self.events if e.relevance_score > 0.7]),
            'top_error_patterns': self._analyze_error_patterns()
        }

        return analysis

    # def _get_time_range(self) -> Dict:
    #     """Get time range of events"""
    #     if not self.events:
    #         return {}

    #     timestamps = [event.timestamp for event in self.events if event.timestamp]
    #     if timestamps:
    #         return {
    #             'start': min(timestamps),
    #             'end': max(timestamps),
    #             'duration': f"{len(set(timestamps))} unique timestamps"
    #         }
    #     return {}

    def _analyze_error_patterns(self) -> List[Tuple[str, int]]:
        """Analyze common error patterns"""
        error_messages = [
            event.message for event in self.events
            if event.level in ['Critical', 'Error'] and event.message
        ]

        # Extract common error patterns
        patterns = defaultdict(int)
        for message in error_messages:
            # Simple pattern extraction - could be enhanced with ML
            words = re.findall(r'\b\w+\b', message.lower())
            for word in words:
                if len(word) > 4 and word not in ['error', 'failed', 'unable', 'could', 'dropped', 'drop', 'endpoint' 'network', 'transport']:
                    patterns[word] += 1

        return Counter(patterns).most_common(10)

    def filter_events(self,
                     min_relevance_score: float = 0.5,
                     max_events: int = 1000,
                     include_categories: Optional[Set[str]] = None,
                     time_window_hours: Optional[int] = None) -> List[EventInfo]:
        """Filter events based on relevance and criteria"""

        filtered = self.events.copy()

        # Filter by relevance score
        filtered = [e for e in filtered if e.relevance_score >= min_relevance_score]

        # Filter by categories if specified
        if include_categories:
            filtered = [e for e in filtered if e.category in include_categories]

        # Filter by time window if specified
        if time_window_hours and filtered:
            # This would need proper datetime parsing for production use
            pass

        # Sort by relevance score (descending) and take top N
        filtered.sort(key=lambda x: x.relevance_score, reverse=True)
        filtered = filtered[:max_events]

        self.filtered_events = filtered
        return filtered

    def generate_summary(self) -> Dict:
        """Generate AI-friendly summary of filtered events"""
        if not self.filtered_events:
            return {}

        # Group events by category and level
        summary = {
            'filtered_event_count': len(self.filtered_events),
            'critical_issues': [],
            'warnings': [],
            'patterns': [],
            'recommendations': []
        }

        # Extract critical issues
        critical_events = [e for e in self.filtered_events if e.level in ['Critical', 'Error']]
        for event in critical_events[:10]:  # Top 10 critical events
            summary['critical_issues'].append({
                'provider': event.provider,
                'event_id': event.event_id,
                'message': event.message[:200] + '...' if len(event.message) > 200 else event.message,
                'relevance_score': event.relevance_score
            })

        # Extract warnings
        warning_events = [e for e in self.filtered_events if e.level == 'Warning']
        for event in warning_events[:5]:
            summary['warnings'].append({
                'provider': event.provider,
                'message': event.message[:150] + '...' if len(event.message) > 150 else event.message,
                'relevance_score': event.relevance_score
            })

        # Identify patterns
        provider_counts = Counter(e.provider for e in self.filtered_events)
        summary['patterns'] = [
            f"High activity from {provider} ({count} events)"
            for provider, count in provider_counts.most_common(5)
            if count > 1
        ]

        # Generate recommendations
        if len(critical_events) > 0:
            summary['recommendations'].append("Investigate critical errors immediately")
        if len([e for e in self.filtered_events if e.category == 'performance']) > 5:
            summary['recommendations'].append("Review system performance issues")
        if len([e for e in self.filtered_events if e.category == 'security']) > 0:
            summary['recommendations'].append("Review security-related events")

        return summary

    def export_for_ai(self, output_path: str, format: str = 'json') -> None:
        """Export filtered events in AI-friendly format"""
        summary = self.generate_summary()
        analysis = self.analyze_events()

        export_data = {
            'metadata': {
                'source_file': str(self.etl_file_path),
                'analysis_timestamp': datetime.now().isoformat(),
                'total_events_processed': len(self.events),
                'filtered_events_exported': len(self.filtered_events)
            },
            'analysis': analysis,
            'summary': summary,
            'filtered_events': [asdict(event) for event in self.filtered_events]
        }

        output_file = Path(output_path)

        if format.lower() == 'json':
            with open(output_file.with_suffix('.json'), 'w', encoding='utf-8') as f:
                json.dump(export_data, f, indent=2, ensure_ascii=False)
        elif format.lower() == 'csv':
            df = pd.DataFrame([asdict(event) for event in self.filtered_events])
            df.to_csv(output_file.with_suffix('.csv'), index=False, encoding='utf-8')
        else:
            raise ValueError("Supported formats: 'json', 'csv'")

        print(f"Exported filtered events to: {output_file}")


def main():
    """Example usage of the ETL Analyzer"""
    
    # Example usage
    etl_file = input("Enter path to ETL file: ").strip('"')
    
    if not os.path.exists(etl_file):
        print(f"Error: ETL file not found: {etl_file}")
        return
        
    try:
        # Initialize analyzer
        analyzer = ETLAnalyzer(etl_file)
        
        # Convert ETL to CSV
        print("Converting ETL file...")
        csv_file = analyzer.convert_etl_to_csv()
        
        # Parse events
        print("Parsing events...")
        analyzer.events = analyzer.parse_events_from_csv(csv_file)
        
        # Analyze events
        print("Analyzing events...")
        analysis = analyzer.analyze_events()
        
        print(f"\nAnalysis Results:")
        print(f"Total events: {analysis.get('total_events', 0)}")
        print(f"Critical/Error events: {analysis.get('critical_events', 0)}")
        print(f"High relevance events: {analysis.get('high_relevance_events', 0)}")
        
        # Filter events
        print("\nFiltering events...")
        filtered = analyzer.filter_events(
            min_relevance_score=0.6,
            max_events=500
        )
        
        print(f"Filtered to {len(filtered)} most relevant events")
        
        # Export results
        output_path = etl_file.replace('.etl', '_filtered')
        analyzer.export_for_ai(output_path, 'json')
        
        print("\n" + "="*50)
        print("ETL Analysis Complete!")
        print(f"Filtered events exported to: {output_path}.json")
        print("This file is now ready for AI-assisted analysis.")
        
    except Exception as e:
        print(f"Error during analysis: {e}")
        return

if __name__ == "__main__":
    main()