AI-Assisted Bug Triaging: Intelligent Defect Prioritization at Scale

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from (as discussed in [AI Test Metrics Analytics: Intelligent Analysis of QA Metrics](/blog/ai-test-metrics)) sklearn.metrics import classification_report
import numpy as np

class BugSeverityPredictor:
    def __init__(self):
        self.vectorizer = TfidfVectorizer(
            max_features=5000,
            ngram_range=(1, 3),
            stop_words='english'
        )
        self.classifier = RandomForestClassifier(
            n_estimators=200,
            max_depth=20,
            min_samples_split=5,
            class_weight='balanced',
            random_state=42
        )

    def prepare_features(self, df):
        """Combine text fields and extract features"""
        # Combine title and description
        df['combined_text'] = df['title'] + ' ' + df['description']

        # Extract additional features
        df['title_length'] = df['title'].str.len()
        df['desc_length'] = df['description'].str.len()
        df['has_stacktrace'] = df['description'].str.contains(
            'at |Traceback|Exception', regex=True
        ).astype(int)
        df['error_keyword_count'] = df['description'].str.lower().str.count(
            'crash|error|fail|exception|critical'
        )

        return df

    def train(self, bugs_df):
        """Train the severity prediction model"""
        # Prepare features
        bugs_df = self.prepare_features(bugs_df)

        # Text vectorization
        text_features = self.vectorizer.fit_transform(
            bugs_df['combined_text']
        )

        # Numerical features
        numerical_features = bugs_df[[
            'title_length', 'desc_length',
            'has_stacktrace', 'error_keyword_count'
        ]].values

        # Combine features
        X = np.hstack([text_features.toarray(), numerical_features])
        y = bugs_df['severity']

        # Train model
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, stratify=y
        )

        self.classifier.fit(X_train, y_train)

        # Evaluate
        y_pred = self.classifier.predict(X_test)
        print(classification_report(y_test, y_pred))

        return self

    def predict(self, title, description):
        """Predict severity for a new bug"""
        df = pd.DataFrame({
            'title': [title],
            'description': [description]
        })
        df = self.prepare_features(df)

        text_features = self.vectorizer.transform(df['combined_text'])
        numerical_features = df[[
            'title_length', 'desc_length',
            'has_stacktrace', 'error_keyword_count'
        ]].values

        X = np.hstack([text_features.toarray(), numerical_features])

        severity = self.classifier.predict(X)[0]
        confidence = self.classifier.predict_proba(X).max()

        return {
            'severity': severity,
            'confidence': confidence
        }

# Usage example
predictor = BugSeverityPredictor()

# Load historical bug data
bugs = pd.read_csv('bug_reports.csv')
predictor.train(bugs)

# Predict severity for new bug
result = predictor.predict(
    title="Application crashes on login",
    description="Users cannot log in. System throws NullPointerException at AuthService.java:245"
)
print(f"Predicted severity: {result['severity']} (confidence: {result['confidence']:.2%})")

from transformers import BertTokenizer, BertForSequenceClassification
import torch
from torch.utils.data import Dataset, DataLoader

class BugDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_length=512):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_length = max_length

    def __len__(self):
        return len(self.texts)

    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]

        encoding = self.tokenizer(
            text,
            max_length=self.max_length,
            padding='max_length',
            truncation=True,
            return_tensors='pt'
        )

        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }

class BERTBugClassifier:
    def __init__(self, num_labels=4):
        self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
        self.model = BertForSequenceClassification.from_pretrained(
            'bert-base-uncased',
            num_labels=num_labels
        )
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.model.to(self.device)

    def train(self, train_texts, train_labels, epochs=3, batch_size=16):
        dataset = BugDataset(train_texts, train_labels, self.tokenizer)
        dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

        optimizer = torch.optim.AdamW(self.model.parameters(), lr=2e-5)

        self.model.train()
        for epoch in range(epochs):
            total_loss = 0
            for batch in dataloader:
                optimizer.zero_grad()

                input_ids = batch['input_ids'].to(self.device)
                attention_mask = batch['attention_mask'].to(self.device)
                labels = batch['labels'].to(self.device)

                outputs = self.model(
                    input_ids=input_ids,
                    attention_mask=attention_mask,
                    labels=labels
                )

                loss = outputs.loss
                total_loss += loss.item()

                loss.backward()
                optimizer.step()

            print(f"Epoch {epoch+1}, Loss: {total_loss/len(dataloader):.4f}")

    def predict(self, text):
        self.model.eval()
        encoding = self.tokenizer(
            text,
            max_length=512,
            padding='max_length',
            truncation=True,
            return_tensors='pt'
        )

        with torch.no_grad():
            input_ids = encoding['input_ids'].to(self.device)
            attention_mask = encoding['attention_mask'].to(self.device)

            outputs = self.model(input_ids=input_ids, attention_mask=attention_mask)
            probabilities = torch.nn.functional.softmax(outputs.logits, dim=1)

        severity_map = {0: 'low', 1: 'medium', 2: 'high', 3: 'critical'}
        predicted_class = torch.argmax(probabilities).item()

        return {
            'severity': severity_map[predicted_class],
            'confidence': probabilities[0][predicted_class].item()
        }

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import faiss

class DuplicateBugDetector:
    def __init__(self):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.bug_embeddings = None
        self.bug_ids = None
        self.index = None

    def build_index(self, bugs_df):
        """Build FAISS index for fast similarity search"""
        # Create bug texts
        texts = (bugs_df['title'] + ' ' + bugs_df['description']).tolist()
        self.bug_ids = bugs_df['bug_id'].tolist()

        # Generate embeddings
        self.bug_embeddings = self.model.encode(texts, show_progress_bar=True)

        # Build FAISS index
        dimension = self.bug_embeddings.shape[1]
        self.index = faiss.IndexFlatIP(dimension)  # Inner Product for cosine similarity

        # Normalize embeddings for cosine similarity
        faiss.normalize_L2(self.bug_embeddings)
        self.index.add(self.bug_embeddings)

        return self

    def find_duplicates(self, new_bug_title, new_bug_description, threshold=0.85, top_k=5):
        """Find potential duplicate bugs"""
        # Create embedding for new bug
        new_text = f"{new_bug_title} {new_bug_description}"
        new_embedding = self.model.encode([new_text])
        faiss.normalize_L2(new_embedding)

        # Search for similar bugs
        similarities, indices = self.index.search(new_embedding, top_k)

        # Filter by threshold
        duplicates = []
        for similarity, idx in zip(similarities[0], indices[0]):
            if similarity >= threshold:
                duplicates.append({
                    'bug_id': self.bug_ids[idx],
                    'similarity_score': float(similarity)
                })

        return duplicates

    def get_similarity_matrix(self, bug_ids_list):
        """Calculate pairwise similarity for a set of bugs"""
        indices = [self.bug_ids.index(bid) for bid in bug_ids_list]
        embeddings_subset = self.bug_embeddings[indices]

        similarity_matrix = cosine_similarity(embeddings_subset)
        return similarity_matrix

# Usage example
detector = DuplicateBugDetector()

# Build index from existing bugs
bugs_df = pd.read_csv('bugs.csv')
detector.build_index(bugs_df)

# Check for duplicates
duplicates = detector.find_duplicates(
    new_bug_title="Login button not working",
    new_bug_description="When clicking the login button, nothing happens. Console shows no errors.",
    threshold=0.85
)

print("Potential duplicates found:")
for dup in duplicates:
    print(f"Bug ID: {dup['bug_id']}, Similarity: {dup['similarity_score']:.2%}")

class HybridDuplicateDetector:
    def __init__(self, text_weight=0.7, metadata_weight=0.3):
        self.text_detector = DuplicateBugDetector()
        self.text_weight = text_weight
        self.metadata_weight = metadata_weight

    def calculate_metadata_similarity(self, bug1, bug2):
        """Calculate similarity based on metadata"""
        score = 0.0

        # Component match
        if bug1['component'] == bug2['component']:
            score += 0.3

        # Same reporter
        if bug1['reporter'] == bug2['reporter']:
            score += 0.2

        # Similar creation time (within 7 days)
        time_diff = abs((bug1['created_at'] - bug2['created_at']).days)
        if time_diff <= 7:
            score += 0.2 * (1 - time_diff / 7)

        # Same OS/browser
        if bug1.get('os') == bug2.get('os'):
            score += 0.15
        if bug1.get('browser') == bug2.get('browser'):
            score += 0.15

        return score

    def find_duplicates_hybrid(self, new_bug, bugs_df, threshold=0.80):
        """Find duplicates using hybrid approach"""
        # Get text-based duplicates
        text_duplicates = self.text_detector.find_duplicates(
            new_bug['title'],
            new_bug['description'],
            threshold=0.70,  # Lower threshold for initial filtering
            top_k=20
        )

        # Calculate hybrid scores
        results = []
        for dup in text_duplicates:
            bug_data = bugs_df[bugs_df['bug_id'] == dup['bug_id']].iloc[0]

            text_score = dup['similarity_score']
            metadata_score = self.calculate_metadata_similarity(new_bug, bug_data)

            hybrid_score = (
                self.text_weight * text_score +
                self.metadata_weight * metadata_score
            )

            if hybrid_score >= threshold:
                results.append({
                    'bug_id': dup['bug_id'],
                    'hybrid_score': hybrid_score,
                    'text_score': text_score,
                    'metadata_score': metadata_score
                })

        # Sort by hybrid score
        results.sort(key=lambda x: x['hybrid_score'], reverse=True)
        return results

from sklearn.feature_extraction.text import TfidfVectorizer
from collections import defaultdict
import numpy as np

class BugAssignmentRecommender:
    def __init__(self):
        self.developer_profiles = {}
        self.vectorizer = TfidfVectorizer(max_features=1000)
        self.component_experts = defaultdict(list)

    def build_developer_profiles(self, historical_bugs):
        """Build expertise profiles for developers"""
        developer_bugs = defaultdict(list)

        # Group bugs by assignee
        for _, bug in historical_bugs.iterrows():
            if bug['assignee'] and bug['status'] == 'resolved':
                developer_bugs[bug['assignee']].append(
                    f"{bug['title']} {bug['description']}"
                )

                # Track component expertise
                if bug['component']:
                    self.component_experts[bug['component']].append({
                        'developer': bug['assignee'],
                        'resolution_time': bug['resolution_time_hours']
                    })

        # Create TF-IDF profiles
        all_developers = list(developer_bugs.keys())
        all_texts = [' '.join(developer_bugs[dev]) for dev in all_developers]

        if all_texts:
            tfidf_matrix = self.vectorizer.fit_transform(all_texts)

            for idx, developer in enumerate(all_developers):
                self.developer_profiles[developer] = {
                    'expertise_vector': tfidf_matrix[idx],
                    'bugs_resolved': len(developer_bugs[developer]),
                    'avg_resolution_time': self._calculate_avg_time(
                        historical_bugs, developer
                    )
                }

        # Calculate component expertise scores
        for component, assignments in self.component_experts.items():
            dev_stats = defaultdict(lambda: {'count': 0, 'total_time': 0})

            for assignment in assignments:
                dev = assignment['developer']
                dev_stats[dev]['count'] += 1
                dev_stats[dev]['total_time'] += assignment['resolution_time']

            # Calculate average and store
            self.component_experts[component] = [
                {
                    'developer': dev,
                    'bug_count': stats['count'],
                    'avg_time': stats['total_time'] / stats['count']
                }
                for dev, stats in dev_stats.items()
            ]

    def _calculate_avg_time(self, df, developer):
        """Calculate average resolution time for developer"""
        dev_bugs = df[df['assignee'] == developer]
        return dev_bugs['resolution_time_hours'].mean()

    def recommend_assignee(self, bug_title, bug_description, component=None, top_k=3):
        """Recommend best assignees for a new bug"""
        # Create bug vector
        bug_text = f"{bug_title} {bug_description}"
        bug_vector = self.vectorizer.transform([bug_text])

        scores = []

        for developer, profile in self.developer_profiles.items():
            # Text similarity score
            similarity = cosine_similarity(
                bug_vector,
                profile['expertise_vector']
            )[0][0]

            # Component expertise bonus
            component_bonus = 0
            if component and component in self.component_experts:
                experts = self.component_experts[component]
                for expert in experts:
                    if expert['developer'] == developer:
                        # Bonus based on experience and speed
                        component_bonus = 0.2 * min(expert['bug_count'] / 10, 1.0)
                        if expert['avg_time'] < 24:  # Fast resolver
                            component_bonus += 0.1

            # Workload penalty (simplified - would integrate with real-time data)
            workload_penalty = 0  # Would query current open bugs

            # Combined score
            final_score = similarity + component_bonus - workload_penalty

            scores.append({
                'developer': developer,
                'score': final_score,
                'similarity': similarity,
                'component_bonus': component_bonus,
                'avg_resolution_time': profile['avg_resolution_time']
            })

        # Sort and return top K
        scores.sort(key=lambda x: x['score'], reverse=True)
        return scores[:top_k]

# Usage example
recommender = BugAssignmentRecommender()
recommender.build_developer_profiles(historical_bugs_df)

recommendations = recommender.recommend_assignee(
    bug_title="Memory leak in data processing module",
    bug_description="Application consumes increasing memory when processing large datasets",
    component="data-processing"
)

print("Recommended assignees:")
for idx, rec in enumerate(recommendations, 1):
    print(f"{idx}. {rec['developer']}")
    print(f"   Score: {rec['score']:.3f}")
    print(f"   Avg resolution time: {rec['avg_resolution_time']:.1f} hours")

from sklearn.ensemble import GradientBoostingRegressor
import pandas as pd

class SLAOptimizer:
    def __init__(self):
        self.time_predictor = GradientBoostingRegressor(
            n_estimators=100,
            learning_rate=0.1,
            max_depth=5
        )
        self.severity_sla = {
            'critical': 4,   # 4 hours
            'high': 24,      # 24 hours
            'medium': 72,    # 3 days
            'low': 168       # 1 week
        }

    def prepare_features(self, bugs_df):
        """Extract features for time prediction"""
        features = pd.DataFrame()

        # Severity encoding
        severity_map = {'low': 1, 'medium': 2, 'high': 3, 'critical': 4}
        features['severity_code'] = bugs_df['severity'].map(severity_map)

        # Text complexity
        features['title_length'] = bugs_df['title'].str.len()
        features['desc_length'] = bugs_df['description'].str.len()
        features['has_stacktrace'] = bugs_df['description'].str.contains(
            'at |Traceback', regex=True
        ).astype(int)

        # Component complexity (based on historical data)
        component_avg_time = bugs_df.groupby('component')['resolution_time_hours'].mean()
        features['component_complexity'] = bugs_df['component'].map(component_avg_time)

        # Reporter history
        reporter_bug_count = bugs_df.groupby('reporter').size()
        features['reporter_experience'] = bugs_df['reporter'].map(reporter_bug_count)

        # Time features
        bugs_df['created_at'] = pd.to_datetime(bugs_df['created_at'])
        features['hour_of_day'] = bugs_df['created_at'].dt.hour
        features['day_of_week'] = bugs_df['created_at'].dt.dayofweek
        features['is_weekend'] = (features['day_of_week'] >= 5).astype(int)

        return features

    def train(self, bugs_df):
        """Train resolution time predictor"""
        X = self.prepare_features(bugs_df)
        y = bugs_df['resolution_time_hours']

        self.time_predictor.fit(X, y)
        return self

    def predict_resolution_time(self, bug_data):
        """Predict resolution time for a bug"""
        features = self.prepare_features(pd.DataFrame([bug_data]))
        predicted_hours = self.time_predictor.predict(features)[0]
        return predicted_hours

    def calculate_sla_risk(self, bug_data):
        """Calculate SLA breach risk"""
        predicted_time = self.predict_resolution_time(bug_data)
        sla_limit = self.severity_sla.get(bug_data['severity'], 168)

        risk_score = predicted_time / sla_limit

        if risk_score >= 1.0:
            risk_level = 'HIGH'
        elif risk_score >= 0.7:
            risk_level = 'MEDIUM'
        else:
            risk_level = 'LOW'

        return {
            'predicted_hours': predicted_time,
            'sla_hours': sla_limit,
            'risk_score': risk_score,
            'risk_level': risk_level,
            'recommended_action': self._get_recommendation(risk_level)
        }

    def _get_recommendation(self, risk_level):
        """Get recommended actions based on risk"""
        if risk_level == 'HIGH':
            return "URGENT: Assign to senior developer immediately"
        elif risk_level == 'MEDIUM':
            return "Monitor closely, consider escalation"
        else:
            return "Standard workflow"

    def optimize_queue(self, open_bugs_df):
        """Prioritize bug queue to optimize SLA compliance"""
        priorities = []

        for _, bug in open_bugs_df.iterrows():
            sla_analysis = self.calculate_sla_risk(bug.to_dict())

            # Calculate urgency score
            time_remaining = sla_analysis['sla_hours'] - bug['hours_open']
            urgency = sla_analysis['risk_score'] * (1 / max(time_remaining, 1))

            priorities.append({
                'bug_id': bug['bug_id'],
                'urgency_score': urgency,
                'sla_risk': sla_analysis['risk_level'],
                'time_remaining': time_remaining,
                'predicted_resolution': sla_analysis['predicted_hours']
            })

        # Sort by urgency
        priorities.sort(key=lambda x: x['urgency_score'], reverse=True)
        return priorities

# Usage example
optimizer = SLAOptimizer()
optimizer.train(historical_bugs_df)

# Analyze a new bug
bug = {
    'severity': 'high',
    'title': 'Payment processing fails',
    'description': 'Critical bug affecting checkout. Stack trace included...',
    'component': 'payment-gateway',
    'reporter': 'user@example.com',
    'created_at': '2025-10-04 14:30:00'
}

risk_analysis = optimizer.calculate_sla_risk(bug)
print(f"SLA Risk: {risk_analysis['risk_level']}")
print(f"Predicted resolution: {risk_analysis['predicted_hours']:.1f} hours")
print(f"Recommendation: {risk_analysis['recommended_action']}")

from jira import JIRA
import requests

class JIRATriagingIntegration:
    def __init__(self, server, email, api_token):
        self.jira = JIRA(server=server, basic_auth=(email, api_token))
        self.predictor = BugSeverityPredictor()
        self.duplicate_detector = DuplicateBugDetector()
        self.recommender = BugAssignmentRecommender()
        self.sla_optimizer = SLAOptimizer()

    def process_new_issue(self, issue_key):
        """Automatically triage a new JIRA issue"""
        # Fetch issue
        issue = self.jira.issue(issue_key)

        title = issue.fields.summary
        description = issue.fields.description or ""

        # 1. Predict severity
        severity_result = self.predictor.predict(title, description)

        # 2. Check for duplicates
        duplicates = self.duplicate_detector.find_duplicates(
            title, description, threshold=0.85
        )

        # 3. Recommend assignee
        component = issue.fields.components[0].name if issue.fields.components else None
        assignee_recommendations = self.recommender.recommend_assignee(
            title, description, component
        )

        # 4. Calculate SLA risk
        bug_data = {
            'severity': severity_result['severity'],
            'title': title,
            'description': description,
            'component': component,
            'reporter': issue.fields.reporter.emailAddress,
            'created_at': issue.fields.created
        }
        sla_risk = self.sla_optimizer.calculate_sla_risk(bug_data)

        # 5. Update JIRA issue
        updates = {}

        # Set priority based on predicted severity
        priority_map = {
            'critical': 'Highest',
            'high': 'High',
            'medium': 'Medium',
            'low': 'Low'
        }
        updates['priority'] = {'name': priority_map[severity_result['severity']]}

        # Add AI analysis comment
        comment = f"""AI Triaging Analysis:

**Predicted Severity:** {severity_result['severity']} (confidence: {severity_result['confidence']:.1%})

**Duplicate Detection:**
{self._format_duplicates(duplicates)}

**Recommended Assignees:**
{self._format_recommendations(assignee_recommendations)}

**SLA Analysis:**
- Risk Level: {sla_risk['risk_level']}
- Predicted Resolution: {sla_risk['predicted_hours']:.1f} hours
- SLA Limit: {sla_risk['sla_hours']} hours
- Recommendation: {sla_risk['recommended_action']}
"""

        # Update issue
        issue.update(fields=updates)
        self.jira.add_comment(issue, comment)

        # Auto-assign if high confidence
        if assignee_recommendations and assignee_recommendations[0]['score'] > 0.8:
            best_assignee = assignee_recommendations[0]['developer']
            issue.update(assignee={'name': best_assignee})

        # Add labels
        labels = issue.fields.labels or []
        labels.append(f"ai-severity-{severity_result['severity']}")
        if duplicates:
            labels.append('possible-duplicate')
        if sla_risk['risk_level'] == 'HIGH':
            labels.append('sla-at-risk')
        issue.update(fields={'labels': labels})

        return {
            'severity': severity_result,
            'duplicates': duplicates,
            'assignee_recommendations': assignee_recommendations,
            'sla_risk': sla_risk
        }

    def _format_duplicates(self, duplicates):
        if not duplicates:
            return "No duplicates found"

        text = ""
        for dup in duplicates[:3]:
            text += f"- {dup['bug_id']} (similarity: {dup['similarity_score']:.1%})\n"
        return text

    def _format_recommendations(self, recommendations):
        text = ""
        for idx, rec in enumerate(recommendations, 1):
            text += f"{idx}. {rec['developer']} (score: {rec['score']:.2f}, "
            text += f"avg time: {rec['avg_resolution_time']:.1f}h)\n"
        return text

    def batch_process_untriaged(self, jql_query="status = Open AND priority is EMPTY"):
        """Process all untriaged issues"""
        issues = self.jira.search_issues(jql_query, maxResults=100)

        results = []
        for issue in issues:
            try:
                result = self.process_new_issue(issue.key)
                results.append({'issue': issue.key, 'status': 'success', 'result': result})
            except Exception as e:
                results.append({'issue': issue.key, 'status': 'error', 'error': str(e)})

        return results

# Usage example
integration = JIRATriagingIntegration(
    server='https://your-domain.atlassian.net',
    email='your-email@example.com',
    api_token='your-api-token'
)

# Process a new issue
result = integration.process_new_issue('PROJ-1234')
print(f"Triaging complete: {result}")

# Batch process untriaged issues
batch_results = integration.batch_process_untriaged()
print(f"Processed {len(batch_results)} issues")

from github import Github

class GitHubTriagingBot:
    def __init__(self, access_token, repo_name):
        self.gh = Github(access_token)
        self.repo = self.gh.get_repo(repo_name)
        self.predictor = BugSeverityPredictor()
        self.duplicate_detector = DuplicateBugDetector()

    def process_issue(self, issue_number):
        """Triage a GitHub issue"""
        issue = self.repo.get_issue(issue_number)

        # Predict severity
        severity_result = self.predictor.predict(
            issue.title,
            issue.body or ""
        )

        # Find duplicates
        duplicates = self.duplicate_detector.find_duplicates(
            issue.title,
            issue.body or "",
            threshold=0.85
        )

        # Apply labels
        labels = []
        labels.append(f"severity:{severity_result['severity']}")

        if severity_result['severity'] in ['critical', 'high']:
            labels.append('priority:high')

        if duplicates:
            labels.append('duplicate?')

        issue.add_to_labels(*labels)

        # Add comment with analysis
        if duplicates:
            dup_text = "\n".join([
                f"- #{dup['bug_id']} (similarity: {dup['similarity_score']:.1%})"
                for dup in duplicates[:3]
            ])

            comment = f"""## AI Triaging Bot Analysis

**Predicted Severity:** {severity_result['severity']} (confidence: {severity_result['confidence']:.1%})

**Possible Duplicates:**
{dup_text}

Please review these potential duplicates before proceeding.
"""
            issue.create_comment(comment)

        return {
            'severity': severity_result,
            'duplicates': duplicates,
            'labels_applied': labels
        }

    def webhook_handler(self, payload):
        """Handle GitHub webhook for new issues"""
        if payload['action'] == 'opened':
            issue_number = payload['issue']['number']
            return self.process_issue(issue_number)

# Usage
bot = GitHubTriagingBot(
    access_token='ghp_xxx',
    repo_name='username/repository'
)

result = bot.process_issue(42)
print(f"Triaging result: {result}")