validators.py

import subprocess
import os
import tempfile
from radon.complexity import cc_visit
from radon.metrics import mi_visit
from typing import Dict, List


class CodeValidator:
    """Validates code using various tools: pylint, black, radon."""
    
    def __init__(self):
        self.results = {}
    
    def validate_file(self, file_path: str, content: str) -> Dict:
        """Run all validators on a file."""
        results = {
            'file': file_path,
            'lint_score': 0,
            'complexity_score': 0,
            'maintainability_index': 0,
            'issues': [],
            'passed': False
        }
        
        # Only validate Python files
        if not file_path.endswith('.py'):
            results['passed'] = True
            results['issues'].append("Non-Python file - skipping validation")
            return results
        
        # Write content to temp file for validation
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            temp_file = f.name
            f.write(content)
        
        try:
            # Run pylint
            lint_result = self._run_pylint(temp_file)
            results['lint_score'] = lint_result['score']
            results['issues'].extend(lint_result['issues'])
            
            # Run black formatter check
            black_result = self._check_black_format(temp_file)
            if not black_result['formatted']:
                results['issues'].append("Code is not formatted with black")
            
            # Run radon complexity analysis
            complexity_result = self._analyze_complexity(content)
            results['complexity_score'] = complexity_result['score']
            results['maintainability_index'] = complexity_result['mi']
            results['issues'].extend(complexity_result['issues'])
            
            # Determine if validation passed
            results['passed'] = (
                results['lint_score'] >= 7.0 and
                results['complexity_score'] >= 6.0 and
                results['maintainability_index'] >= 60
            )
            
        finally:
            os.unlink(temp_file)
        
        return results
    
    def _run_pylint(self, file_path: str) -> Dict:
        """Run pylint on a file."""
        try:
            result = subprocess.run(
                ['pylint', file_path, '--output-format=json'],
                capture_output=True,
                text=True,
                timeout=30
            )
            
            # Parse pylint output
            import json
            try:
                messages = json.loads(result.stdout) if result.stdout else []
            except:
                messages = []
            
            # Extract score from stderr (pylint outputs score there)
            score = 10.0
            if result.stderr:
                for line in result.stderr.split('\n'):
                    if 'rated at' in line.lower():
                        try:
                            score = float(line.split('rated at')[1].split('/')[0].strip())
                        except:
                            pass
            
            issues = []
            for msg in messages[:10]:  # Limit to top 10 issues
                if isinstance(msg, dict):
                    issues.append(f"Line {msg.get('line', '?')}: {msg.get('message', 'Unknown issue')}")
            
            return {'score': score, 'issues': issues}
        except subprocess.TimeoutExpired:
            return {'score': 5.0, 'issues': ['Pylint timeout']}
        except Exception as e:
            return {'score': 5.0, 'issues': [f'Pylint error: {str(e)}']}
    
    def _check_black_format(self, file_path: str) -> Dict:
        """Check if file is formatted with black."""
        try:
            result = subprocess.run(
                ['black', '--check', file_path],
                capture_output=True,
                text=True,
                timeout=10
            )
            return {'formatted': result.returncode == 0}
        except:
            return {'formatted': True}  # If black fails, assume it's okay
    
    def _analyze_complexity(self, content: str) -> Dict:
        """Analyze code complexity using radon."""
        try:
            # Cyclomatic complexity
            complexity_results = cc_visit(content)
            avg_complexity = 0
            high_complexity_count = 0
            issues = []
            
            if complexity_results:
                complexities = [item.complexity for item in complexity_results]
                avg_complexity = sum(complexities) / len(complexities) if complexities else 0
                
                for item in complexity_results:
                    if item.complexity > 10:
                        high_complexity_count += 1
                        issues.append(f"High complexity in {item.name}: {item.complexity}")
            
            # Maintainability Index
            mi_results = mi_visit(content, multi=True)
            avg_mi = 0
            if mi_results:
                avg_mi = sum(mi_results) / len(mi_results) if mi_results else 100
            
            # Score complexity (inverse of average complexity, scaled to 0-10)
            complexity_score = max(0, min(10, 10 - avg_complexity))
            
            return {
                'score': complexity_score,
                'mi': avg_mi,
                'issues': issues[:5]  # Limit to top 5
            }
        except Exception as e:
            return {
                'score': 7.0,
                'mi': 70.0,
                'issues': [f'Complexity analysis error: {str(e)}']
            }
    
    def validate_patch(self, file_path: str, original_content: str, modified_content: str) -> Dict:
        """Validate a code patch."""
        original_results = self.validate_file(file_path, original_content)
        modified_results = self.validate_file(file_path, modified_content)
        
        return {
            'original': original_results,
            'modified': modified_results,
            'improved': (
                modified_results['lint_score'] >= original_results['lint_score'] and
                modified_results['complexity_score'] >= original_results['complexity_score']
            ),
            'passed': modified_results['passed']
        }


def check_safety_limits(files_changed: int, lines_changed: int, max_files: int = 10, max_lines: int = 500) -> Dict:
    """Check if changes are within safety limits."""
    return {
        'within_limits': files_changed <= max_files and lines_changed <= max_lines,
        'files_changed': files_changed,
        'max_files': max_files,
        'lines_changed': lines_changed,
        'max_lines': max_lines,
        'issues': [
            f"Too many files changed: {files_changed} > {max_files}" if files_changed > max_files else None,
            f"Too many lines changed: {lines_changed} > {max_lines}" if lines_changed > max_lines else None
        ]
    }