Spaces:
Sleeping
Sleeping
File size: 12,999 Bytes
aa300a4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 |
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Agent Manager
This module provides the main orchestrator for the Code Review Agent.
It coordinates the review process and manages the state of the application.
"""
import os
import time
import logging
import tempfile
import json
from datetime import datetime
import gradio as gr
from src.core.language_detector import LanguageDetector
from src.services.code_analyzer import CodeAnalyzer
from src.services.report_generator import ReportGenerator
from src.services.repository_service import RepositoryService
from src.services.security_scanner import SecurityScanner
from src.services.performance_analyzer import PerformanceAnalyzer
logger = logging.getLogger(__name__)
class AgentManager:
"""
Main orchestrator for the Code Review Agent.
This class coordinates the review process, manages the application state,
and provides the interface between the UI and the business logic.
"""
def __init__(self):
"""
Initialize the AgentManager.
"""
# Initialize state management
self.state = {
'repo_url': None,
'progress': {},
'results': {},
'current_step': None
}
# Initialize services
self.language_detector = LanguageDetector()
self.code_analyzer = CodeAnalyzer()
self.report_generator = ReportGenerator()
self.repository_service = RepositoryService()
self.security_scanner = SecurityScanner()
self.performance_analyzer = PerformanceAnalyzer()
self.temp_dir = tempfile.mkdtemp(prefix="code_review_agent_")
logger.info(f"Initialized AgentManager with temp directory: {self.temp_dir}")
def start_review(self, repo_url, github_token=None, selected_languages=None):
"""
Start the code review process for a GitHub repository.
Args:
repo_url (str): The URL of the GitHub repository to review.
github_token (str, optional): GitHub authentication token for private repositories.
selected_languages (list, optional): List of languages to analyze. If None,
languages will be auto-detected.
Returns:
tuple: (progress_group, overall_progress, status_message, results_dashboard) - Updated UI components.
"""
# Initialize progress components outside the try block
progress_group = gr.Group(visible=True)
overall_progress = gr.Slider(value=0)
status_message = gr.Markdown("*Starting review...*")
try:
# Initialize state for new review
self.state = {
'repo_url': repo_url,
'progress': {},
'results': {},
'current_step': None
}
# Clone repository
self._update_progress("Repository Cloning", 0, overall_progress, status_message)
repo_path = self._clone_repository(repo_url, github_token)
self._update_progress("Repository Cloning", 100, overall_progress, status_message)
# Detect languages
self._update_progress("Language Detection", 0, overall_progress, status_message)
if selected_languages and len(selected_languages) > 0:
languages = selected_languages
logger.info(f"Using selected languages: {languages}")
else:
languages = self.language_detector.detect_languages(repo_path)
logger.info(f"Auto-detected languages: {languages}")
self.state['languages'] = languages
self._update_progress("Language Detection", 100, overall_progress, status_message)
# Perform code analysis
self._update_progress("Code Analysis", 0, overall_progress, status_message)
code_analysis_results = self.code_analyzer.analyze_repository(repo_path, languages)
self.state['results']['code_analysis'] = code_analysis_results
self._update_progress("Code Analysis", 100, overall_progress, status_message)
# Perform security scanning
self._update_progress("Security Scanning", 0, overall_progress, status_message)
security_results = self.security_scanner.scan_repository(repo_path, languages)
self.state['results']['security'] = security_results
self._update_progress("Security Scanning", 100, overall_progress, status_message)
# Perform performance analysis
self._update_progress("Performance Analysis", 0, overall_progress, status_message)
performance_results = self.performance_analyzer.analyze_repository(repo_path, languages)
self.state['results']['performance'] = performance_results
self._update_progress("Performance Analysis", 100, overall_progress, status_message)
# Perform AI review
self._update_progress("AI Review", 0, overall_progress, status_message)
ai_review_results = self._perform_ai_review(repo_path, languages)
self.state['results']['ai_review'] = ai_review_results
self._update_progress("AI Review", 100, overall_progress, status_message)
# Generate report
self._update_progress("Report Generation", 0, overall_progress, status_message)
repo_name = repo_url.split('/')[-1].replace('.git', '')
report_paths = self.report_generator.generate_report(
repo_name, self.state['results']
)
self.state['report_paths'] = report_paths
self._update_progress("Report Generation", 100, overall_progress, status_message)
# Update results dashboard
results_dashboard = self._create_results_dashboard(self.state['results'])
results_dashboard.update(visible=True)
return progress_group, overall_progress, status_message, results_dashboard
except Exception as e:
logger.exception(f"Error during code review: {e}")
# Update progress components with error
status_message.update(value=f"*Error: {str(e)}*")
return progress_group, overall_progress, status_message, None
def export_report(self, results_dashboard, export_format):
"""
Export the code review report in the specified format.
Args:
results_dashboard: The results dashboard component.
export_format (str): The format to export the report in ('pdf', 'json', 'html', 'csv').
Returns:
str: The path to the exported file.
"""
try:
if not self.state.get('results'):
logger.warning("No results available to export")
return None
# Get the actual format value from the textbox component
format_value = export_format.value if hasattr(export_format, 'value') else export_format
# Create exports directory if it doesn't exist
exports_dir = os.path.join(os.path.dirname(__file__), '..', '..', 'exports')
os.makedirs(exports_dir, exist_ok=True)
# Generate filename with timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
repo_name = self.state['repo_url'].split('/')[-1].replace('.git', '')
filename = f"{repo_name}_review_{timestamp}.{format_value}"
filepath = os.path.join(exports_dir, filename)
# Export report in the specified format using report_generator
report_paths = self.report_generator.generate_report(
repo_name, self.state['results'], format_value
)
if format_value in report_paths:
return report_paths[format_value]
else:
logger.warning(f"Unsupported export format: {format_value}")
return None
logger.info(f"Exported report to {filepath}")
return filepath
except Exception as e:
logger.exception(f"Error exporting report: {e}")
return None
def _clone_repository(self, repo_url, github_token=None):
"""
Clone the GitHub repository to a temporary directory.
Args:
repo_url (str): The URL of the GitHub repository to clone.
github_token (str, optional): GitHub authentication token for private repositories.
Returns:
str: The path to the cloned repository.
"""
# Import the repository service here to avoid circular imports
from src.services.repository_service import RepositoryService
# Create a repository service instance
repo_service = RepositoryService(base_temp_dir=self.temp_dir)
# Clone the repository using the service
try:
# If a GitHub token is provided, use it for authentication
if github_token and github_token.strip():
# Modify the URL to include the token for authentication
auth_url = repo_url.replace('https://', f'https://{github_token}@')
repo_path = repo_service.clone_repository(auth_url)
logger.info(f"Cloned repository using GitHub token authentication")
else:
# Clone without authentication (for public repositories)
repo_path = repo_service.clone_repository(repo_url)
logger.info(f"Cloned repository without authentication")
return repo_path
except Exception as e:
logger.error(f"Error cloning repository: {e}")
raise
def _perform_ai_review(self, repo_path, languages):
"""
Perform AI-powered code review.
Args:
repo_path (str): The path to the repository.
languages (list): List of programming languages to analyze.
Returns:
dict: AI review results.
"""
try:
# This is a placeholder for AI review functionality
# In a real implementation, this would use the MCP AI review service
from src.mcp.ai_review import AIReviewMCP
ai_reviewer = AIReviewMCP()
results = ai_reviewer.review_repository(repo_path, languages)
logger.info(f"AI review completed for {len(languages)} languages")
return results
except Exception as e:
logger.error(f"Error during AI review: {e}")
return {
'error': str(e),
'suggestions': [],
'issues': []
}
def _update_progress(self, step, value, overall_progress, status_message):
"""
Update the progress components for a specific step.
Args:
step (str): The step to update.
value (int): The progress value (0-100).
overall_progress: The overall progress slider component.
status_message: The status message markdown component.
"""
# Update state
self.state['progress'][step] = value
self.state['current_step'] = step
# Calculate overall progress
total_steps = 7 # Total number of steps in the review process
completed_steps = sum(1 for v in self.state['progress'].values() if v == 100)
current_step_progress = value if step in self.state['progress'] else 0
overall_value = (completed_steps * 100 + current_step_progress) / total_steps
# Update UI components
overall_progress.update(value=overall_value)
status_message.update(value=f"*{step}: {value}%*")
logger.info(f"Progress update: {step} - {value}% (Overall: {overall_value:.1f}%)")
time.sleep(0.5) # Simulate progress update time
def _create_results_dashboard(self, report):
"""
Create a results dashboard component for the UI.
Args:
report (dict): The code review report.
Returns:
object: A results dashboard component.
"""
# This is a placeholder. In a real implementation, this would create a
# results dashboard component for the UI.
class ResultsDashboard:
def __init__(self):
self.visible = False
def update(self, visible=None):
if visible is not None:
self.visible = visible
return self
return ResultsDashboard() |