#!/usr/bin/env python3 """ AI Analysis Service HTTP Server Provides REST API endpoints for repository analysis. """ import os import asyncio import json import tempfile import shutil from pathlib import Path from typing import Dict, Any from datetime import datetime from fastapi import FastAPI, HTTPException, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from pydantic import BaseModel import uvicorn # Import the AI analysis components # Note: ai-analyze.py has a hyphen, so we need to handle the import specially import sys import importlib.util # Load the ai-analyze.py module spec = importlib.util.spec_from_file_location("ai_analyze", "/app/ai-analyze.py") ai_analyze_module = importlib.util.module_from_spec(spec) sys.modules["ai_analyze"] = ai_analyze_module spec.loader.exec_module(ai_analyze_module) # Now import the classes from ai_analyze import EnhancedGitHubAnalyzer, get_memory_config app = FastAPI( title="AI Analysis Service", description="AI-powered repository analysis with memory system", version="1.0.0" ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Global analyzer instance analyzer = None class AnalysisRequest(BaseModel): repo_path: str output_format: str = "pdf" # pdf, json max_files: int = 50 class AnalysisResponse(BaseModel): success: bool message: str analysis_id: str = None report_path: str = None stats: Dict[str, Any] = None @app.on_event("startup") async def startup_event(): """Initialize the analyzer on startup.""" global analyzer try: # Load environment variables from dotenv import load_dotenv load_dotenv() # Get API key api_key = os.getenv('ANTHROPIC_API_KEY') if not api_key: raise Exception("ANTHROPIC_API_KEY not found in environment") # Initialize analyzer config = get_memory_config() analyzer = EnhancedGitHubAnalyzer(api_key, config) print("✅ AI Analysis Service initialized successfully") except Exception as e: print(f"❌ Failed to initialize AI Analysis Service: {e}") raise @app.get("/health") async def health_check(): """Health check endpoint.""" return { "status": "healthy", "service": "ai-analysis-service", "timestamp": datetime.now().isoformat(), "version": "1.0.0" } @app.post("/analyze", response_model=AnalysisResponse) async def analyze_repository(request: AnalysisRequest, background_tasks: BackgroundTasks): """Analyze a repository.""" try: if not analyzer: raise HTTPException(status_code=500, detail="Analyzer not initialized") # Generate unique analysis ID analysis_id = f"analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}" # Create temporary directory for this analysis temp_dir = tempfile.mkdtemp(prefix=f"ai_analysis_{analysis_id}_") try: # Run analysis analysis = await analyzer.analyze_repository_with_memory( request.repo_path, max_files=request.max_files ) # Generate report if request.output_format == "pdf": report_path = f"/app/reports/{analysis_id}_analysis.pdf" analyzer.create_pdf_report(analysis, report_path) else: report_path = f"/app/reports/{analysis_id}_analysis.json" with open(report_path, 'w') as f: json.dump({ "repo_path": analysis.repo_path, "total_files": analysis.total_files, "total_lines": analysis.total_lines, "languages": analysis.languages, "code_quality_score": analysis.code_quality_score, "architecture_assessment": analysis.architecture_assessment, "security_assessment": analysis.security_assessment, "executive_summary": analysis.executive_summary, "file_analyses": [ { "path": fa.path, "language": fa.language, "lines_of_code": fa.lines_of_code, "severity_score": fa.severity_score, "issues_found": fa.issues_found, "recommendations": fa.recommendations } for fa in analysis.file_analyses ] }, f, indent=2) # Calculate stats stats = { "total_files": analysis.total_files, "total_lines": analysis.total_lines, "languages": analysis.languages, "code_quality_score": analysis.code_quality_score, "high_quality_files": len([fa for fa in analysis.file_analyses if fa.severity_score >= 8]), "medium_quality_files": len([fa for fa in analysis.file_analyses if 5 <= fa.severity_score < 8]), "low_quality_files": len([fa for fa in analysis.file_analyses if fa.severity_score < 5]), "total_issues": sum(len(fa.issues_found) for fa in analysis.file_analyses) } return AnalysisResponse( success=True, message="Analysis completed successfully", analysis_id=analysis_id, report_path=report_path, stats=stats ) finally: # Cleanup temporary directory if os.path.exists(temp_dir): shutil.rmtree(temp_dir) except Exception as e: return AnalysisResponse( success=False, message=f"Analysis failed: {str(e)}", analysis_id=None, report_path=None, stats=None ) @app.get("/reports/{filename}") async def download_report(filename: str): """Download analysis report.""" report_path = f"/app/reports/{filename}" if not os.path.exists(report_path): raise HTTPException(status_code=404, detail="Report not found") return FileResponse( report_path, media_type='application/octet-stream', filename=filename ) @app.get("/memory/stats") async def get_memory_stats(): """Get memory system statistics.""" try: if not analyzer: raise HTTPException(status_code=500, detail="Analyzer not initialized") stats = await analyzer.memory_manager.get_memory_stats() return { "success": True, "memory_stats": stats } except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to get memory stats: {str(e)}") @app.post("/memory/query") async def query_memory(query: str, repo_context: str = ""): """Query the memory system.""" try: if not analyzer: raise HTTPException(status_code=500, detail="Analyzer not initialized") result = await analyzer.query_memory(query, repo_context) return { "success": True, "query": query, "result": result } except Exception as e: raise HTTPException(status_code=500, detail=f"Memory query failed: {str(e)}") if __name__ == "__main__": port = int(os.getenv('PORT', 8022)) host = os.getenv('HOST', '0.0.0.0') print(f"🚀 Starting AI Analysis Service on {host}:{port}") uvicorn.run(app, host=host, port=port)