Triaje de Bugs con IA: Priorización Inteligente de Defectos a Escala

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 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):
        """Combinar campos de texto y extraer características"""
        # Combinar título y descripción
        df['combined_text'] = df['title'] + ' ' + df['description']

        # Extraer características adicionales
        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):
        """Entrenar el modelo de predicción de severidad"""
        # Preparar características
        bugs_df = self.prepare_features(bugs_df)

        # Vectorización de texto
        text_features = self.vectorizer.fit_transform(
            bugs_df['combined_text']
        )

        # Características numéricas
        numerical_features = bugs_df[[
            'title_length', 'desc_length',
            'has_stacktrace', 'error_keyword_count'
        ]].values

        # Combinar características
        X = np.hstack([text_features.toarray(), numerical_features])
        y = bugs_df['severity']

        # Entrenar modelo
        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)

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

        return self

    def predict(self, title, description):
        """Predecir severidad para un nuevo 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
        }

# Ejemplo de uso
predictor = BugSeverityPredictor()

# Cargar datos históricos de bugs
bugs = pd.read_csv('bug_reports.csv')
predictor.train(bugs)

# Predecir severidad para nuevo bug
result = predictor.predict(
    title="La aplicación se bloquea al iniciar sesión",
    description="Los usuarios no pueden iniciar sesión. El sistema lanza NullPointerException en AuthService.java:245"
)
print(f"Severidad predicha: {result['severity']} (confianza: {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"Época {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: 'bajo', 1: 'medio', 2: 'alto', 3: 'crítico'}
        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):
        """Construir índice FAISS para búsqueda rápida de similitud"""
        # Crear textos de bugs
        texts = (bugs_df['title'] + ' ' + bugs_df['description']).tolist()
        self.bug_ids = bugs_df['bug_id'].tolist()

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

        # Construir índice FAISS
        dimension = self.bug_embeddings.shape[1]
        self.index = faiss.IndexFlatIP(dimension)

        # Normalizar embeddings para similitud coseno
        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):
        """Encontrar bugs duplicados potenciales"""
        new_text = f"{new_bug_title} {new_bug_description}"
        new_embedding = self.model.encode([new_text])
        faiss.normalize_L2(new_embedding)

        similarities, indices = self.index.search(new_embedding, top_k)

        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

# Ejemplo de uso
detector = DuplicateBugDetector()
bugs_df = pd.read_csv('bugs.csv')
detector.build_index(bugs_df)

duplicates = detector.find_duplicates(
    new_bug_title="El botón de login no funciona",
    new_bug_description="Al hacer clic en el botón de login, no pasa nada.",
    threshold=0.85
)

print("Duplicados potenciales encontrados:")
for dup in duplicates:
    print(f"Bug ID: {dup['bug_id']}, Similitud: {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):
        """Calcular similitud basada en metadatos"""
        score = 0.0

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

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

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

        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):
        """Encontrar duplicados usando enfoque híbrido"""
        text_duplicates = self.text_detector.find_duplicates(
            new_bug['title'],
            new_bug['description'],
            threshold=0.70,
            top_k=20
        )

        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
                })

        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):
        """Construir perfiles de experiencia para desarrolladores"""
        developer_bugs = defaultdict(list)

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

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

        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
                    )
                }

    def recommend_assignee(self, bug_title, bug_description, component=None, top_k=3):
        """Recomendar mejores asignados para un nuevo bug"""
        bug_text = f"{bug_title} {bug_description}"
        bug_vector = self.vectorizer.transform([bug_text])

        scores = []

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

            component_bonus = 0
            if component and component in self.component_experts:
                experts = self.component_experts[component]
                for expert in experts:
                    if expert['developer'] == developer:
                        component_bonus = 0.2 * min(expert['bug_count'] / 10, 1.0)
                        if expert['avg_time'] < 24:
                            component_bonus += 0.1

            final_score = similarity + component_bonus

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

        scores.sort(key=lambda x: x['score'], reverse=True)
        return scores[:top_k]

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 horas
            'high': 24,      # 24 horas
            'medium': 72,    # 3 días
            'low': 168       # 1 semana
        }

    def calculate_sla_risk(self, bug_data):
        """Calcular riesgo de violación de SLA"""
        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 = 'ALTO'
        elif risk_score >= 0.7:
            risk_level = 'MEDIO'
        else:
            risk_level = 'BAJO'

        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):
        """Obtener acciones recomendadas según el riesgo"""
        if risk_level == 'ALTO':
            return "URGENTE: Asignar a desarrollador senior inmediatamente"
        elif risk_level == 'MEDIO':
            return "Monitorear de cerca, considerar escalación"
        else:
            return "Flujo de trabajo estándar"

    def optimize_queue(self, open_bugs_df):
        """Priorizar cola de bugs para optimizar cumplimiento de SLA"""
        priorities = []

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

            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']
            })

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

from jira import JIRA

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):
        """Triaje automático de un nuevo issue de JIRA"""
        issue = self.jira.issue(issue_key)

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

        # 1. Predecir severidad
        severity_result = self.predictor.predict(title, description)

        # 2. Verificar duplicados
        duplicates = self.duplicate_detector.find_duplicates(
            title, description, threshold=0.85
        )

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

        # 4. Calcular riesgo de SLA
        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. Actualizar issue de JIRA
        comment = f"""Análisis de Triaje con IA:

**Severidad Predicha:** {severity_result['severity']} (confianza: {severity_result['confidence']:.1%})

**Detección de Duplicados:**
{self._format_duplicates(duplicates)}

**Asignados Recomendados:**
{self._format_recommendations(assignee_recommendations)}

**Análisis de SLA:**

- Nivel de Riesgo: {sla_risk['risk_level']}
- Resolución Predicha: {sla_risk['predicted_hours']:.1f} horas
- Límite SLA: {sla_risk['sla_hours']} horas
- Recomendación: {sla_risk['recommended_action']}
"""
        self.jira.add_comment(issue, comment)

        # Auto-asignar si hay alta confianza
        if assignee_recommendations and assignee_recommendations[0]['score'] > 0.8:
            best_assignee = assignee_recommendations[0]['developer']
            issue.update(assignee={'name': best_assignee})

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

    def batch_process_untriaged(self, jql_query="status = Open AND priority is EMPTY"):
        """Procesar todos los issues sin triaje"""
        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

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):
        """Triaje de un issue de GitHub"""
        issue = self.repo.get_issue(issue_number)

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

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

        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)

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

            comment = f"""## Análisis del Bot de Triaje con IA

**Severidad Predicha:** {severity_result['severity']} (confianza: {severity_result['confidence']:.1%})

**Posibles Duplicados:**
{dup_text}

Por favor revisa estos duplicados potenciales antes de continuar.
"""
            issue.create_comment(comment)

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