AI for DevOps

#!/usr/bin/env python3
"""AI Code Review Automation using OpenAI API"""
import os
import openai
from git import Repo

def review_pr_with_ai(diff_text):
    """Analyze PR diff with AI for code quality"""
    openai.api_key = os.getenv('OPENAI_API_KEY')
    
    prompt = f"""
    Analyze this code diff and provide feedback:
    
    {diff_text}
    
    Focus on:
    1. Security vulnerabilities
    2. Code quality issues
    3. Best practice violations
    4. Potential bugs
    
    Format: JSON with "issues" and "suggestions"
    """
    
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are an expert code reviewer"},
            {"role": "user", "content": prompt}
        ]
    )
    
    return response.choices[0].message.content

def get_last_commit_diff(repo_path):
    """Get diff of the last commit"""
    repo = Repo(repo_path)
    commit = repo.head.commit
    diff = commit.parents[0].diff(commit)
    
    return "\n".join([
        f"File: {item.a_path}\n{item.diff.decode('utf-8')}"
        for item in diff
    ])

if __name__ == "__main__":
    diff = get_last_commit_diff(".")
    review = review_pr_with_ai(diff)
    print(f"AI Code Review Results:\n{review}")

#!/usr/bin/env python3
"""Auto-generate test cases from code"""
import openai
import ast

def extract_functions_from_code(code):
    """Extract function definitions from Python code"""
    tree = ast.parse(code)
    functions = []
    
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef):
            functions.append({
                'name': node.name,
                'args': [arg.arg for arg in node.args.args],
                'docstring': ast.get_docstring(node)
            })
    
    return functions

def generate_tests_for_function(function_info):
    """Generate unit tests using AI"""
    openai.api_key = os.getenv('OPENAI_API_KEY')
    
    prompt = f"""
    Generate pytest test cases for this function:
    
    Function: {function_info['name']}
    Parameters: {function_info['args']}
    Description: {function_info['docstring']}
    
    Create tests for:
    1. Normal cases
    2. Edge cases
    3. Error handling
    """
    
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a Python testing expert"},
            {"role": "user", "content": prompt}
        ]
    )
    
    return response.choices[0].message.content

def auto_test_generation(file_path):
    """Full pipeline for auto test generation"""
    with open(file_path, 'r') as f:
        code = f.read()
    
    functions = extract_functions_from_code(code)
    
    for func in functions:
        test_code = generate_tests_for_function(func)
        test_file = f"test_{func['name']}.py"
        
        with open(test_file, 'w') as f:
            f.write(test_code)
        
        print(f"Generated {test_file}")

if __name__ == "__main__":
    auto_test_generation("my_module.py")

#!/usr/bin/env python3
"""Predictive Pipeline Failure Detection"""
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

class PipelineFailurePredictor:
    """ML model to predict CI/CD pipeline failures"""
    
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100)
        self.scaler = StandardScaler()
        self.feature_names = [
            'code_changes', 'test_coverage', 'complexity',
            'previous_failures', 'build_time', 'team_size'
        ]
    
    def extract_features(self, pipeline_data):
        """Extract features from pipeline metadata"""
        features = {
            'code_changes': pipeline_data.get('files_changed', 0),
            'test_coverage': pipeline_data.get('coverage', 0),
            'complexity': pipeline_data.get('cyclomatic_complexity', 0),
            'previous_failures': pipeline_data.get('failure_history', 0),
            'build_time': pipeline_data.get('duration', 0),
            'team_size': pipeline_data.get('contributors', 0)
        }
        
        return np.array([list(features.values())])
    
    def train(self, historical_data):
        """Train model on historical pipeline data"""
        df = pd.DataFrame(historical_data)
        
        X = df[self.feature_names].values
        y = df['success'].values
        
        X_scaled = self.scaler.fit_transform(X)
        X_train, X_test, y_train, y_test = train_test_split(
            X_scaled, y, test_size=0.2, random_state=42
        )
        
        self.model.fit(X_train, y_train)
        
        accuracy = self.model.score(X_test, y_test)
        print(f"Model accuracy: {accuracy:.2%}")
    
    def predict_failure(self, pipeline_data):
        """Predict probability of pipeline failure"""
        features = self.extract_features(pipeline_data)
        features_scaled = self.scaler.transform(features)
        
        probability = self.model.predict_proba(features_scaled)[0][1]
        return probability

# Example Usage
if __name__ == "__main__":
    # Train with historical data
    historical = [
        {'files_changed': 5, 'coverage': 85, 'cyclomatic_complexity': 10,
         'failure_history': 0, 'duration': 120, 'contributors': 2, 'success': 1},
        {'files_changed': 15, 'coverage': 60, 'cyclomatic_complexity': 25,
         'failure_history': 3, 'duration': 300, 'contributors': 1, 'success': 0},
    ]
    
    predictor = PipelineFailurePredictor()
    predictor.train(historical)
    
    # Predict new pipeline
    new_pipeline = {
        'files_changed': 12,
        'coverage': 70,
        'cyclomatic_complexity': 20,
        'failure_history': 1,
        'duration': 250,
        'contributors': 1
    }
    
    failure_prob = predictor.predict_failure(new_pipeline)
    
    if failure_prob > 0.7:
        print(f"⚠️  High risk: {failure_prob:.0%} chance of failure")
        print("Recommendations:")
        print("- Increase test coverage")
        print("- Review code complexity")
        print("- Add more reviewers")
    else:
        print(f"✅ Low risk: {failure_prob:.0%} chance of failure")

#!/usr/bin/env python3
"""Intelligent Error Diagnosis with NLP"""
import re
import openai
from collections import Counter

class ErrorDiagnosisAgent:
    """AI-powered error diagnosis and solution recommendation"""
    
    def __init__(self):
        openai.api_key = os.getenv('OPENAI_API_KEY')
        self.error_patterns = {
            'NullPointerException': ['Check null checks', 'Initialize variables'],
            'ConnectionTimeout': ['Check network', 'Increase timeout'],
            'OutOfMemory': ['Check memory leaks', 'Increase heap size'],
            'ImportError': ['Install missing package', 'Check PYTHONPATH'],
        }
    
    def parse_error_logs(self, log_text):
        """Extract error messages and stack traces"""
        errors = []
        
        # Common error patterns
        patterns = [
            r'Error:\s*(.+)',
            r'Exception:\s*(.+)',
            r'(\w+Error):',
            r'Failed\s+(.+)',
        ]
        
        for pattern in patterns:
            matches = re.findall(pattern, log_text)
            errors.extend(matches)
        
        return list(set(errors))
    
    def diagnose_error_with_ai(self, error_message, context=""):
        """Use AI to diagnose error and suggest solutions"""
        prompt = f"""
        Analyze this error and provide diagnosis:
        
        Error: {error_message}
        Context: {context}
        
        Provide:
        1. Root cause analysis
        2. Recommended fixes (with code examples if applicable)
        3. Prevention strategies
        
        Format: Clear sections with bullet points
        """
        
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert developer"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )
        
        return response.choices[0].message.content
    
    def analyze_trends(self, historical_errors):
        """Identify recurring error patterns"""
        error_counts = Counter(historical_errors)
        
        return [
            {'error': error, 'count': count}
            for error, count in error_counts.most_common(5)
        ]
    
    def generate_report(self, logs):
        """Generate comprehensive error report"""
        errors = self.parse_error_logs(logs)
        
        report = "# Error Diagnosis Report\n\n"
        report += f"## Errors Found: {len(errors)}\n\n"
        
        for error in errors[:3]:  # Top 3 errors
            diagnosis = self.diagnose_error_with_ai(error)
            report += f"### {error}\n\n{diagnosis}\n\n---\n\n"
        
        return report

# Example Usage
if __name__ == "__main__":
    agent = ErrorDiagnosisAgent()
    
    # Sample error logs
    logs = """
    2024-01-15 10:30:45 ERROR: ConnectionTimeout: Failed to connect to database
    2024-01-15 10:31:00 ERROR: ImportError: No module named 'pandas'
    2024-01-15 10:32:15 ERROR: NullPointerException at line 45
    """
    
    # Parse errors
    errors = agent.parse_error_logs(logs)
    print(f"Detected errors: {errors}")
    
    # Diagnose first error
    if errors:
        diagnosis = agent.diagnose_error_with_ai(
            errors[0],
            context="Python application running on Ubuntu 20.04"
        )
        print("\n=== AI Diagnosis ===\n")
        print(diagnosis)
    
    # Generate full report
    report = agent.generate_report(logs)
    print("\n=== Full Report ===\n")
    print(report)