pr_scorer.py

from typing import Dict, List
from validators import CodeValidator
from gemini_ai import analyze_pr_quality


class PRScorer:
    """Score and rank pull requests based on quality metrics."""
    
    def __init__(self):
        self.validator = CodeValidator()
    
    def score_pr(self, pr_data: Dict, diff_content: str) -> Dict:
        """Score a single pull request."""
        score_breakdown = {
            'pr_number': pr_data['number'],
            'title': pr_data['title'],
            'author': pr_data['author'],
            'url': pr_data['url'],
            'metrics': {},
            'total_score': 0,
            'recommendation': 'needs_work'
        }
        
        # Metric 1: Size appropriateness (0-10)
        size_score = self._score_size(
            pr_data['files_changed'],
            pr_data['additions'],
            pr_data['deletions']
        )
        score_breakdown['metrics']['size'] = size_score
        
        # Metric 2: Recency (0-10)
        recency_score = self._score_recency(pr_data['created_at'], pr_data['updated_at'])
        score_breakdown['metrics']['recency'] = recency_score
        
        # Metric 3: Description quality (0-10)
        description_score = self._score_description(pr_data['description'])
        score_breakdown['metrics']['description'] = description_score
        
        # Metric 4: AI quality analysis (0-10)
        ai_analysis = analyze_pr_quality(
            pr_data['title'],
            pr_data['description'],
            diff_content,
            pr_data['files_changed']
        )
        ai_score = ai_analysis.get('overall_score', 5)
        score_breakdown['metrics']['ai_quality'] = ai_score
        score_breakdown['ai_analysis'] = ai_analysis
        
        # Metric 5: Mergeable status (bonus points)
        mergeable_bonus = 2 if pr_data.get('mergeable') else 0
        score_breakdown['metrics']['mergeable_bonus'] = mergeable_bonus
        
        # Calculate total score (weighted average)
        total_score = (
            size_score * 0.2 +
            recency_score * 0.15 +
            description_score * 0.15 +
            ai_score * 0.5 +
            mergeable_bonus * 0.1
        )
        
        score_breakdown['total_score'] = round(total_score, 2)
        
        # Determine recommendation
        if total_score >= 8.0:
            score_breakdown['recommendation'] = 'merge'
        elif total_score >= 6.0:
            score_breakdown['recommendation'] = 'review'
        else:
            score_breakdown['recommendation'] = 'needs_work'
        
        return score_breakdown
    
    def _score_size(self, files_changed: int, additions: int, deletions: int) -> float:
        """Score based on PR size - prefer focused PRs."""
        total_changes = additions + deletions
        
        # Ideal PR: 1-3 files, 10-200 lines
        if 1 <= files_changed <= 3 and 10 <= total_changes <= 200:
            return 10.0
        elif files_changed <= 5 and total_changes <= 400:
            return 8.0
        elif files_changed <= 10 and total_changes <= 800:
            return 6.0
        elif files_changed <= 15 and total_changes <= 1500:
            return 4.0
        else:
            return 2.0
    
    def _score_recency(self, created_at: str, updated_at: str) -> float:
        """Score based on how recent the PR is."""
        from datetime import datetime, timezone
        
        try:
            updated = datetime.fromisoformat(updated_at.replace('Z', '+00:00'))
            now = datetime.now(timezone.utc)
            days_old = (now - updated).days
            
            if days_old <= 1:
                return 10.0
            elif days_old <= 7:
                return 8.0
            elif days_old <= 30:
                return 6.0
            elif days_old <= 90:
                return 4.0
            else:
                return 2.0
        except:
            return 5.0
    
    def _score_description(self, description: str) -> float:
        """Score based on description quality."""
        if not description or len(description.strip()) < 20:
            return 2.0
        
        desc_lower = description.lower()
        
        # Check for key elements
        has_problem = any(word in desc_lower for word in ['fix', 'bug', 'issue', 'problem', 'error'])
        has_solution = any(word in desc_lower for word in ['solution', 'change', 'add', 'update', 'implement'])
        has_testing = any(word in desc_lower for word in ['test', 'tested', 'testing', 'verify'])
        
        score = 5.0
        if has_problem:
            score += 2.0
        if has_solution:
            score += 2.0
        if has_testing:
            score += 1.0
        
        # Bonus for detailed description
        if len(description) > 200:
            score += 1.0
        
        return min(10.0, score)
    
    def rank_prs(self, pr_list: List[Dict], diff_contents: Dict[int, str]) -> List[Dict]:
        """Rank a list of pull requests."""
        scored_prs = []
        
        for pr in pr_list:
            diff = diff_contents.get(pr['number'], '')
            score = self.score_pr(pr, diff)
            scored_prs.append(score)
        
        # Sort by total score (descending)
        scored_prs.sort(key=lambda x: x['total_score'], reverse=True)
        
        return scored_prs
    
    def get_top_recommendations(self, ranked_prs: List[Dict], top_n: int = 5) -> List[Dict]:
        """Get top N PR recommendations."""
        merge_candidates = [pr for pr in ranked_prs if pr['recommendation'] == 'merge']
        review_candidates = [pr for pr in ranked_prs if pr['recommendation'] == 'review']
        
        recommendations = merge_candidates[:top_n]
        if len(recommendations) < top_n:
            recommendations.extend(review_candidates[:top_n - len(recommendations)])
        
        return recommendations