app.py

"""
🐞 Enhanced Bug Report Analysis Agent
=====================================
A comprehensive RAG-based system for analyzing bug reports, finding similar issues,
and suggesting fixes with evaluation metrics for retrieval relevance and usefulness.
"""

import os
import pandas as pd
import numpy as np
import gradio as gr
import sqlite3
import json
import ast
import re
from datetime import datetime, timedelta
from typing import List, Dict, Tuple, Optional
import logging

# Core RAG and ML imports
from sentence_transformers import SentenceTransformer
import faiss
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.feature_extraction.text import TfidfVectorizer
import nltk
from fuzzywuzzy import fuzz, process

# LangChain imports
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document

# Evaluation metrics
from rouge_score import rouge_scorer
import difflib

# Download required NLTK data
try:
    nltk.download('punkt', quiet=True)
    nltk.download('stopwords', quiet=True)
except:
    pass

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class BugReportRAG:
    """Enhanced RAG system for bug report analysis"""
    
    def __init__(self):
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.tfidf_vectorizer = TfidfVectorizer(stop_words='english', max_features=1000)
        self.bug_index = None
        self.code_index = None
        self.bug_data = None
        self.code_data = None
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
        )
        
    def load_and_index_data(self):
        """Load and index bug reports and code files"""
        logger.info("Loading and indexing data...")
        
        # Load bug reports
        self._load_bug_reports()
        
        # Load and process code files
        self._load_code_files()
        
        # Create FAISS indices
        self._create_faiss_indices()
        
        logger.info("Data loading and indexing completed")
    
    def _load_bug_reports(self):
        """Load and process bug reports from CSV"""
        try:
            df = pd.read_csv("bug_reports.csv")
            
            # Create comprehensive text representation for each bug
            bug_texts = []
            bug_metadata = []
            
            for _, row in df.iterrows():
                # Combine relevant fields for better semantic search
                text_parts = [
                    f"Title: {row.get('title', '')}",
                    f"Description: {row.get('description', '')}",
                    f"Component: {row.get('component', '')}",
                    f"Severity: {row.get('severity', '')}",
                    f"Status: {row.get('status', '')}",
                ]
                
                if pd.notna(row.get('fix_description')):
                    text_parts.append(f"Fix: {row['fix_description']}")
                
                bug_text = " | ".join(text_parts)
                bug_texts.append(bug_text)
                
                # Store metadata
                metadata = {
                    'id': row.get('id', ''),
                    'title': row.get('title', ''),
                    'description': row.get('description', ''),
                    'severity': row.get('severity', ''),
                    'status': row.get('status', ''),
                    'component': row.get('component', ''),
                    'fix_description': row.get('fix_description', ''),
                    'related_files': row.get('related_files', ''),
                    'created_date': row.get('created_date', ''),
                    'resolved_date': row.get('resolved_date', ''),
                }
                bug_metadata.append(metadata)
            
            self.bug_data = {
                'texts': bug_texts,
                'metadata': bug_metadata
            }
            
        except Exception as e:
            logger.error(f"Error loading bug reports: {e}")
            self.bug_data = {'texts': [], 'metadata': []}
    
    def _load_code_files(self):
        """Load and process code files"""
        code_texts = []
        code_metadata = []
        
        for root, dirs, files in os.walk("codebase"):
            for file in files:
                if file.endswith(('.py', '.js', '.html', '.css')):
                    file_path = os.path.join(root, file)
                    try:
                        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                            content = f.read()
                            
                        # Split large files into chunks
                        if len(content) > 1000:
                            chunks = self.text_splitter.split_text(content)
                            for i, chunk in enumerate(chunks):
                                code_texts.append(f"File: {file} | {chunk}")
                                code_metadata.append({
                                    'file_path': file_path,
                                    'file_name': file,
                                    'chunk_index': i,
                                    'total_chunks': len(chunks)
                                })
                        else:
                            code_texts.append(f"File: {file} | {content}")
                            code_metadata.append({
                                'file_path': file_path,
                                'file_name': file,
                                'chunk_index': 0,
                                'total_chunks': 1
                            })
                            
                    except Exception as e:
                        logger.warning(f"Error reading {file_path}: {e}")
        
        self.code_data = {
            'texts': code_texts,
            'metadata': code_metadata
        }
    
    def _create_faiss_indices(self):
        """Create FAISS indices for efficient similarity search"""
        # Create bug report index
        if self.bug_data['texts']:
            bug_embeddings = self.embedding_model.encode(self.bug_data['texts'])
            self.bug_index = faiss.IndexFlatIP(bug_embeddings.shape[1])
            # Normalize embeddings for cosine similarity
            faiss.normalize_L2(bug_embeddings)
            self.bug_index.add(bug_embeddings.astype('float32'))
        
        # Create code index
        if self.code_data['texts']:
            code_embeddings = self.embedding_model.encode(self.code_data['texts'])
            self.code_index = faiss.IndexFlatIP(code_embeddings.shape[1])
            faiss.normalize_L2(code_embeddings)
            self.code_index.add(code_embeddings.astype('float32'))
    
    def search_similar_bugs(self, query: str, k: int = 5) -> List[Dict]:
        """Search for similar bug reports"""
        if not self.bug_index or not self.bug_data['texts']:
            return []
        
        # Encode query
        query_embedding = self.embedding_model.encode([query])
        faiss.normalize_L2(query_embedding)
        
        # Search
        scores, indices = self.bug_index.search(query_embedding.astype('float32'), k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < len(self.bug_data['metadata']):
                result = self.bug_data['metadata'][idx].copy()
                result['similarity_score'] = float(score)
                results.append(result)
        
        return results
    
    def search_relevant_code(self, query: str, k: int = 5) -> List[Dict]:
        """Search for relevant code sections"""
        if not self.code_index or not self.code_data['texts']:
            return []
        
        # Encode query
        query_embedding = self.embedding_model.encode([query])
        faiss.normalize_L2(query_embedding)
        
        # Search
        scores, indices = self.code_index.search(query_embedding.astype('float32'), k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < len(self.code_data['metadata']):
                result = self.code_data['metadata'][idx].copy()
                result['similarity_score'] = float(score)
                result['code_text'] = self.code_data['texts'][idx]
                results.append(result)
        
        return results

class BugAnalysisEvaluator:
    """Evaluate the quality and relevance of bug analysis results"""
    
    def __init__(self):
        self.rouge_scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
    
    def evaluate_retrieval_relevance(self, query: str, results: List[Dict]) -> Dict:
        """Evaluate how relevant retrieved results are to the query"""
        if not results:
            return {
                'average_similarity': 0.0,
                'relevance_score': 0.0,
                'result_count': 0
            }
        
        # Calculate average similarity score
        similarity_scores = [r.get('similarity_score', 0.0) for r in results]
        average_similarity = np.mean(similarity_scores) if similarity_scores else 0.0
        
        # Calculate semantic relevance using text similarity
        query_lower = query.lower()
        relevance_scores = []
        
        for result in results:
            # Combine title and description for relevance calculation
            result_text = f"{result.get('title', '')} {result.get('description', '')}"
            relevance_score = fuzz.partial_ratio(query_lower, result_text.lower()) / 100.0
            relevance_scores.append(relevance_score)
        
        relevance_score = np.mean(relevance_scores) if relevance_scores else 0.0
        
        return {
            'average_similarity': float(average_similarity),
            'relevance_score': float(relevance_score),
            'result_count': len(results),
            'individual_scores': similarity_scores
        }
    
    def evaluate_suggestion_usefulness(self, query: str, suggestions: str) -> Dict:
        """Evaluate the usefulness of generated suggestions"""
        if not suggestions or not query:
            return {
                'completeness_score': 0.0,
                'specificity_score': 0.0,
                'actionability_score': 0.0,
                'overall_usefulness': 0.0
            }
        
        # Completeness: How well suggestions address the query
        rouge_scores = self.rouge_scorer.score(query.lower(), suggestions.lower())
        completeness_score = rouge_scores['rougeL'].fmeasure
        
        # Specificity: Presence of specific technical terms, file names, functions
        specificity_indicators = [
            r'\b\w+\.py\b',  # Python files
            r'\bdef \w+\b',   # Function definitions
            r'\bclass \w+\b', # Class definitions
            r'\b\w+\(\)',     # Function calls
            r'\bfix\b|\bupdate\b|\bchange\b|\bmodify\b',  # Action words
        ]
        
        specificity_count = sum(len(re.findall(pattern, suggestions.lower())) 
                               for pattern in specificity_indicators)
        specificity_score = min(specificity_count / 5.0, 1.0)  # Normalize to 0-1
        
        # Actionability: Presence of actionable steps
        actionable_phrases = [
            'check', 'verify', 'update', 'modify', 'fix', 'add', 'remove',
            'ensure', 'validate', 'test', 'debug', 'implement', 'configure'
        ]
        
        actionability_count = sum(1 for phrase in actionable_phrases 
                                 if phrase in suggestions.lower())
        actionability_score = min(actionability_count / 5.0, 1.0)
        
        # Overall usefulness (weighted average)
        overall_usefulness = (
            0.3 * completeness_score +
            0.4 * specificity_score +
            0.3 * actionability_score
        )
        
        return {
            'completeness_score': float(completeness_score),
            'specificity_score': float(specificity_score),
            'actionability_score': float(actionability_score),
            'overall_usefulness': float(overall_usefulness)
        }

class FixSuggestionEngine:
    """Generate intelligent fix suggestions based on analysis"""
    
    def __init__(self):
        self.common_fixes = {
            'authentication': [
                "Check password validation regex patterns",
                "Verify session management configuration",
                "Ensure proper error handling in login flow",
                "Review authentication middleware setup"
            ],
            'database': [
                "Check database connection pooling settings",
                "Review query optimization and indexing",
                "Verify transaction handling and rollbacks",
                "Check for connection timeout configurations"
            ],
            'email': [
                "Verify SMTP server configuration",
                "Check email template rendering",
                "Ensure email credentials are properly set",
                "Review email queue processing"
            ],
            'ui': [
                "Check JavaScript event listeners",
                "Verify CSS styling and responsive design",
                "Review form validation logic",
                "Ensure proper DOM element targeting"
            ]
        }
    
    def generate_suggestions(self, query: str, similar_bugs: List[Dict], 
                           relevant_code: List[Dict]) -> str:
        """Generate fix suggestions based on analysis"""
        suggestions = []
        
        # Add context-based suggestions
        suggestions.append("## 🔍 Analysis Summary")
        suggestions.append(f"Based on the query: '{query}'")
        suggestions.append("")
        
        # Add similar bug insights
        if similar_bugs:
            suggestions.append("## 🪲 Similar Issues Found")
            for i, bug in enumerate(similar_bugs[:3], 1):
                status = bug.get('status', 'Unknown')
                severity = bug.get('severity', 'Unknown')
                suggestions.append(f"{i}. **{bug.get('title', 'Untitled')}** (Status: {status}, Severity: {severity})")
                
                if bug.get('fix_description'):
                    suggestions.append(f"   - Previous fix: {bug['fix_description']}")
                suggestions.append("")
        
        # Add code analysis
        if relevant_code:
            suggestions.append("## 💻 Relevant Code Sections")
            for i, code in enumerate(relevant_code[:3], 1):
                file_name = code.get('file_name', 'Unknown file')
                suggestions.append(f"{i}. **{file_name}** (Similarity: {code.get('similarity_score', 0):.2f})")
                suggestions.append("")
        
        # Add specific fix suggestions based on component analysis
        component_suggestions = self._get_component_suggestions(query, similar_bugs)
        if component_suggestions:
            suggestions.append("## 🛠️ Suggested Actions")
            for suggestion in component_suggestions:
                suggestions.append(f"- {suggestion}")
            suggestions.append("")
        
        # Add general debugging steps
        suggestions.append("## 🔧 General Debugging Steps")
        suggestions.extend([
            "- Review error logs and stack traces",
            "- Test in different environments (dev/staging/prod)",
            "- Check recent code changes in related files",
            "- Verify configuration settings",
            "- Run relevant test suites",
            "- Consider rollback if issue is critical"
        ])
        
        return "\n".join(suggestions)
    
    def _get_component_suggestions(self, query: str, similar_bugs: List[Dict]) -> List[str]:
        """Get component-specific suggestions"""
        suggestions = []
        query_lower = query.lower()
        
        # Identify likely component based on keywords and similar bugs
        components = [bug.get('component', '').lower() for bug in similar_bugs]
        
        # Keyword-based component detection
        if any(keyword in query_lower for keyword in ['login', 'auth', 'password', 'session']):
            suggestions.extend(self.common_fixes.get('authentication', []))
        
        if any(keyword in query_lower for keyword in ['database', 'db', 'query', 'connection']):
            suggestions.extend(self.common_fixes.get('database', []))
        
        if any(keyword in query_lower for keyword in ['email', 'smtp', 'mail', 'notification']):
            suggestions.extend(self.common_fixes.get('email', []))
        
        if any(keyword in query_lower for keyword in ['button', 'form', 'ui', 'interface', 'display']):
            suggestions.extend(self.common_fixes.get('ui', []))
        
        # Component-based suggestions from similar bugs
        for component in components:
            if component and component in self.common_fixes:
                suggestions.extend(self.common_fixes[component])
        
        return list(set(suggestions))  # Remove duplicates

# Initialize the RAG system and other components
rag_system = BugReportRAG()
evaluator = BugAnalysisEvaluator()
suggestion_engine = FixSuggestionEngine()

# Load and index data on startup
rag_system.load_and_index_data()

def analyze_bug_report(query: str) -> Tuple[str, str, str, str]:
    """Main function to analyze bug reports"""
    try:
        if not query.strip():
            return "Please enter a bug description", "", "", ""
        
        logger.info(f"Analyzing query: {query}")
        
        # Search for similar bugs and relevant code
        similar_bugs = rag_system.search_similar_bugs(query, k=5)
        relevant_code = rag_system.search_relevant_code(query, k=5)
        
        # Generate suggestions
        suggestions = suggestion_engine.generate_suggestions(query, similar_bugs, relevant_code)
        
        # Evaluate results
        bug_evaluation = evaluator.evaluate_retrieval_relevance(query, similar_bugs)
        suggestion_evaluation = evaluator.evaluate_suggestion_usefulness(query, suggestions)
        
        # Format similar bugs output
        similar_bugs_output = format_similar_bugs(similar_bugs, bug_evaluation)
        
        # Format relevant code output
        relevant_code_output = format_relevant_code(relevant_code)
        
        # Format evaluation metrics
        evaluation_output = format_evaluation_metrics(bug_evaluation, suggestion_evaluation)
        
        return similar_bugs_output, relevant_code_output, suggestions, evaluation_output
        
    except Exception as e:
        logger.error(f"Error analyzing bug report: {e}")
        return f"Error: {str(e)}", "", "", ""

def format_similar_bugs(bugs: List[Dict], evaluation: Dict) -> str:
    """Format similar bugs for display"""
    if not bugs:
        return "No similar bugs found in the database."
    
    output = [f"## 🔍 Found {len(bugs)} Similar Bug Reports"]
    output.append(f"**Relevance Score: {evaluation['relevance_score']:.2f}/1.0**")
    output.append(f"**Average Similarity: {evaluation['average_similarity']:.2f}/1.0**")
    output.append("")
    
    for i, bug in enumerate(bugs, 1):
        output.append(f"### {i}. {bug.get('title', 'Untitled Bug')}")
        output.append(f"**ID:** {bug.get('id', 'N/A')} | **Severity:** {bug.get('severity', 'N/A')} | **Status:** {bug.get('status', 'N/A')}")
        output.append(f"**Similarity:** {bug.get('similarity_score', 0):.3f}")
        output.append(f"**Component:** {bug.get('component', 'N/A')}")
        output.append("")
        output.append(f"**Description:** {bug.get('description', 'No description available')}")
        
        if bug.get('fix_description'):
            output.append(f"**Previous Fix:** {bug['fix_description']}")
        
        if bug.get('related_files'):
            output.append(f"**Related Files:** {bug['related_files']}")
        
        output.append("---")
    
    return "\n".join(output)

def format_relevant_code(code_results: List[Dict]) -> str:
    """Format relevant code sections for display"""
    if not code_results:
        return "No relevant code sections found."
    
    output = [f"## 💻 Found {len(code_results)} Relevant Code Sections"]
    output.append("")
    
    for i, code in enumerate(code_results, 1):
        file_name = code.get('file_name', 'Unknown file')
        similarity = code.get('similarity_score', 0)
        
        output.append(f"### {i}. {file_name}")
        output.append(f"**Similarity:** {similarity:.3f} | **Path:** {code.get('file_path', 'N/A')}")
        
        if code.get('chunk_index', 0) > 0:
            total_chunks = code.get('total_chunks', 1)
            output.append(f"**Chunk:** {code['chunk_index'] + 1}/{total_chunks}")
        
        output.append("")
        
        # Extract and display code snippet
        code_text = code.get('code_text', '')
        if 'File:' in code_text:
            _, code_content = code_text.split('|', 1)
            code_content = code_content.strip()
        else:
            code_content = code_text
        
        # Limit code display length
        if len(code_content) > 500:
            code_content = code_content[:500] + "\n... (truncated)"
        
        output.append("```python")
        output.append(code_content)
        output.append("```")
        output.append("---")
    
    return "\n".join(output)

def format_evaluation_metrics(bug_eval: Dict, suggestion_eval: Dict) -> str:
    """Format evaluation metrics for display"""
    output = ["## 📊 Analysis Quality Metrics"]
    output.append("")
    
    # Bug retrieval metrics
    output.append("### 🔍 Retrieval Relevance")
    output.append(f"- **Average Similarity Score:** {bug_eval['average_similarity']:.3f}/1.0")
    output.append(f"- **Semantic Relevance:** {bug_eval['relevance_score']:.3f}/1.0")
    output.append(f"- **Results Retrieved:** {bug_eval['result_count']}")
    
    # Suggestion quality metrics
    output.append("")
    output.append("### 🛠️ Suggestion Quality")
    output.append(f"- **Completeness:** {suggestion_eval['completeness_score']:.3f}/1.0")
    output.append(f"- **Specificity:** {suggestion_eval['specificity_score']:.3f}/1.0")
    output.append(f"- **Actionability:** {suggestion_eval['actionability_score']:.3f}/1.0")
    output.append(f"- **Overall Usefulness:** {suggestion_eval['overall_usefulness']:.3f}/1.0")
    
    # Quality assessment
    overall_quality = (bug_eval['relevance_score'] + suggestion_eval['overall_usefulness']) / 2
    output.append("")
    output.append("### ⭐ Overall Analysis Quality")
    
    if overall_quality >= 0.8:
        quality_label = "🟢 Excellent"
    elif overall_quality >= 0.6:
        quality_label = "🟡 Good"
    elif overall_quality >= 0.4:
        quality_label = "🟠 Fair"
    else:
        quality_label = "🔴 Poor"
    
    output.append(f"**Quality Rating:** {quality_label} ({overall_quality:.3f}/1.0)")
    
    return "\n".join(output)

# Create Gradio interface
def create_interface():
    """Create the Gradio interface for the Bug Report Analysis Agent"""
    
    with gr.Blocks(
        title="🐞 Bug Report Analysis Agent",
        theme=gr.themes.Soft(),
        css="""
        .gradio-container {
            max-width: 1200px !important;
        }
        .tab-nav {
            font-weight: bold;
        }
        """
    ) as demo:
        
        gr.Markdown("""
        # 🐞 Bug Report Analysis Agent
        
        **Advanced RAG-powered system for intelligent bug analysis**
        
        This system analyzes bug reports using Retrieval-Augmented Generation (RAG) to:
        - 🔍 Find similar past issues in the bug database
        - 💻 Identify relevant code sections that might be related
        - 🛠️ Suggest potential causes and fixes
        - 📊 Evaluate retrieval relevance and suggestion usefulness
        
        ---
        """)
        
        with gr.Row():
            with gr.Column(scale=1):
                input_box = gr.Textbox(
                    lines=6,
                    label="🔍 Bug Description",
                    placeholder="Describe the bug you're experiencing...\n\nExample: 'Login form redirects back to login page after entering correct credentials'",
                    info="Provide as much detail as possible for better analysis"
                )
                
                with gr.Row():
                    analyze_btn = gr.Button("🔍 Analyze Bug", variant="primary", size="lg")
                    clear_btn = gr.Button("🗑️ Clear", variant="secondary")
        
        with gr.Row():
            with gr.Column(scale=1):
                similar_bugs_output = gr.Markdown(
                    label="🪲 Similar Bug Reports",
                    value="Enter a bug description and click 'Analyze Bug' to see similar issues..."
                )
            
            with gr.Column(scale=1):
                relevant_code_output = gr.Markdown(
                    label="💻 Relevant Code Sections", 
                    value="Code analysis will appear here..."
                )
        
        with gr.Row():
            with gr.Column(scale=1):
                suggestions_output = gr.Markdown(
                    label="🛠️ Fix Suggestions",
                    value="Intelligent fix suggestions will be generated here..."
                )
            
            with gr.Column(scale=1):
                evaluation_output = gr.Markdown(
                    label="📊 Quality Metrics",
                    value="Analysis quality metrics will be shown here..."
                )
        
        # Event handlers
        analyze_btn.click(
            fn=analyze_bug_report,
            inputs=[input_box],
            outputs=[similar_bugs_output, relevant_code_output, suggestions_output, evaluation_output],
            api_name="analyze_bug"
        )
        
        clear_btn.click(
            fn=lambda: ("", "Enter a bug description and click 'Analyze Bug' to see similar issues...", 
                       "Code analysis will appear here...", 
                       "Intelligent fix suggestions will be generated here...",
                       "Analysis quality metrics will be shown here..."),
            inputs=[],
            outputs=[input_box, similar_bugs_output, relevant_code_output, suggestions_output, evaluation_output]
        )
        
        # Footer
        gr.Markdown("""
        ---
        **🚀 Built with:** LangChain • Sentence Transformers • FAISS • Gradio
        
        **📈 Features:** Semantic Search • Similarity Scoring • Code Analysis • Fix Suggestions • Quality Evaluation
        """)
    
    return demo

if __name__ == "__main__":
    # Create and launch the interface
    demo = create_interface()
    demo.launch(
        share=True,
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True
    )