legacy_code_modernizer / src /agents /pattern_matcher.py
naazimsnh02's picture
Initial deployment: Autonomous AI agent for code modernization
ec4aa90
"""
Production-grade pattern matching system with AI-powered file type detection.
Replaces the simple primary/secondary classification with intelligent pattern detection.
"""
import os
import logging
from typing import Dict, List, Optional, Tuple
from pathlib import Path
import json
from dataclasses import dataclass
from enum import Enum
from src.config import AIManager, GeminiSchemas
logger = logging.getLogger(__name__)
class PatternSeverity(Enum):
"""Severity levels for detected patterns."""
CRITICAL = "critical" # Security issues, breaking changes
HIGH = "high" # Deprecated APIs, performance issues
MEDIUM = "medium" # Code quality, maintainability
LOW = "low" # Style, minor improvements
INFO = "info" # Informational only
@dataclass
class DetectedPattern:
"""Represents a detected legacy pattern."""
pattern_type: str
severity: PatternSeverity
file_path: str
language: str
description: str
line_numbers: List[int]
confidence: float # 0.0 to 1.0
recommendation: str
estimated_effort_hours: float
@dataclass
class FileAnalysis:
"""Complete analysis of a single file."""
file_path: str
language: str
framework: Optional[str]
patterns: List[DetectedPattern]
overall_priority: PatternSeverity
modernization_score: float # 0-100, higher = more modern
requires_modernization: bool
class IntelligentPatternMatcher:
"""
Production-grade pattern matcher using AI for intelligent detection.
Features:
- Language-agnostic pattern detection
- Context-aware analysis
- Confidence scoring
- Batch processing optimization
- Caching for performance
"""
# Language detection patterns
LANGUAGE_PATTERNS = {
# Python
'.py': 'Python',
'.pyw': 'Python',
'.pyx': 'Python (Cython)',
# Java
'.java': 'Java',
# JavaScript/TypeScript
'.js': 'JavaScript',
'.jsx': 'JavaScript (React)',
'.mjs': 'JavaScript (ES Module)',
'.cjs': 'JavaScript (CommonJS)',
'.ts': 'TypeScript',
'.tsx': 'TypeScript (React)',
# PHP
'.php': 'PHP',
'.php3': 'PHP',
'.php4': 'PHP',
'.php5': 'PHP',
'.phtml': 'PHP',
# Ruby
'.rb': 'Ruby',
'.rbw': 'Ruby',
# Go
'.go': 'Go',
# C/C++
'.c': 'C',
'.h': 'C/C++ Header',
'.cpp': 'C++',
'.cc': 'C++',
'.cxx': 'C++',
'.c++': 'C++',
'.hpp': 'C++ Header',
'.hh': 'C++ Header',
'.hxx': 'C++ Header',
'.h++': 'C++ Header',
# C#
'.cs': 'C#',
# Rust
'.rs': 'Rust',
# Kotlin
'.kt': 'Kotlin',
'.kts': 'Kotlin Script',
# Swift
'.swift': 'Swift',
# Scala
'.scala': 'Scala',
'.sc': 'Scala Script',
# R
'.r': 'R',
'.R': 'R',
# Perl
'.pl': 'Perl',
'.pm': 'Perl Module',
'.t': 'Perl Test',
'.pod': 'Perl Documentation',
# Shell
'.sh': 'Shell',
'.bash': 'Bash',
'.zsh': 'Zsh',
'.fish': 'Fish Shell'
}
# Common legacy patterns by language
LEGACY_PATTERNS = {
'Python': [
'Python 2 syntax (print statements, old-style classes)',
'Deprecated libraries (MySQLdb, urllib2, optparse)',
'Missing type hints',
'Hardcoded credentials',
'SQL injection vulnerabilities',
'Insecure cryptography (MD5, SHA1 for passwords)',
'Global variables and mutable defaults',
'Missing error handling',
'Synchronous I/O in async contexts'
],
'Java': [
'Pre-Java 8 code (no lambdas, streams)',
'Deprecated APIs (Vector, Hashtable, Date)',
'Missing generics',
'Raw JDBC without ORM',
'Synchronization issues',
'Resource leaks (missing try-with-resources)',
'Hardcoded configuration',
'Missing null checks'
],
'JavaScript': [
'var instead of let/const',
'Callback hell (no Promises/async-await)',
'jQuery for DOM manipulation',
'eval() usage',
'Missing strict mode',
'Prototype-based inheritance',
'Global namespace pollution',
'XSS vulnerabilities'
],
'TypeScript': [
'any type overuse',
'Missing strict mode',
'Old module syntax',
'Missing null checks',
'Implicit any',
'Type assertions instead of guards'
],
'PHP': [
'mysql_* functions (deprecated)',
'No prepared statements',
'register_globals usage',
'eval() and create_function()',
'Missing input validation',
'Outdated PHP version syntax',
'No namespace usage',
'Missing error handling'
],
'Ruby': [
'Ruby 1.8/1.9 syntax',
'Missing bundler',
'Deprecated gem versions',
'Missing RSpec/Minitest',
'Global variables',
'Missing error handling',
'Synchronous I/O'
],
'Go': [
'Missing error handling',
'Deprecated packages',
'No context usage',
'Missing defer for cleanup',
'Goroutine leaks',
'Race conditions'
],
'C++': [
'Raw pointers instead of smart pointers',
'Manual memory management',
'Missing RAII',
'C-style casts',
'Missing const correctness',
'No move semantics',
'Deprecated C++98/03 features'
],
'C#': [
'Missing async/await patterns',
'Old collection types',
'Missing LINQ usage',
'Deprecated .NET Framework APIs',
'Missing nullable reference types',
'Old string concatenation',
'Missing using statements'
],
'Rust': [
'Deprecated Rust 2015/2018 syntax',
'Missing error handling with Result',
'Unsafe code blocks',
'Missing lifetime annotations',
'Deprecated crate versions',
'Missing async/await'
],
'Kotlin': [
'Java-style code in Kotlin',
'Missing null safety',
'Not using coroutines',
'Missing data classes',
'Old collection APIs',
'Missing extension functions'
],
'Swift': [
'Objective-C style code',
'Missing optionals',
'Old closure syntax',
'Missing guard statements',
'Deprecated Swift 4 features',
'Missing Codable protocol'
],
'Scala': [
'Scala 2.x syntax',
'Missing for-comprehensions',
'Old collection APIs',
'Missing implicit conversions',
'Deprecated Future usage',
'Missing case classes'
],
'R': [
'Old R syntax',
'Missing tidyverse usage',
'Deprecated package versions',
'Missing pipe operators',
'Old data.frame usage',
'Missing ggplot2'
],
'Perl': [
'Perl 4 syntax',
'Missing strict and warnings',
'Old module system',
'Deprecated CPAN modules',
'Missing Moose/Moo',
'Old regex syntax'
],
'Shell': [
'Missing error handling (set -e)',
'Unquoted variables',
'Missing shellcheck compliance',
'Deprecated commands',
'Missing function usage',
'Security vulnerabilities'
]
}
def __init__(self, cache_dir: Optional[str] = None):
"""
Initialize pattern matcher.
Args:
cache_dir: Optional directory for caching analysis results
"""
# Use centralized AI manager
self.ai_manager = AIManager()
self.cache_dir = Path(cache_dir) if cache_dir else None
if self.cache_dir:
self.cache_dir.mkdir(exist_ok=True, parents=True)
logger.info(
f"IntelligentPatternMatcher initialized with provider: {self.ai_manager.provider_name}, "
f"model: {self.ai_manager.model_name}"
)
def detect_language(self, file_path: str, code_sample: str) -> Tuple[str, Optional[str]]:
"""
Detect programming language and framework using AI.
Args:
file_path: Path to the file
code_sample: Sample of code (first 500 chars)
Returns:
Tuple of (language, framework)
"""
# First try extension-based detection
ext = Path(file_path).suffix.lower()
base_language = self.LANGUAGE_PATTERNS.get(ext, 'Unknown')
# Use AI for framework detection
prompt = f"""Analyze this code and identify:
1. Programming language (confirm or correct: {base_language})
2. Framework/library being used (if any)
FILE: {file_path}
CODE SAMPLE:
```
{code_sample[:500]}
```
Respond in JSON format:
{{
"language": "detected language",
"framework": "framework name or null",
"confidence": 0.0-1.0
}}
"""
try:
# Use JSON schema for guaranteed structure
schema = GeminiSchemas.language_detection()
response_text = self.ai_manager.generate_content(
prompt=prompt,
temperature=AIManager.TEMPERATURE_PRECISE,
max_tokens=AIManager.MAX_OUTPUT_TOKENS_SMALL,
response_format="json",
response_schema=schema if self.ai_manager.provider_type == "gemini" else None
)
result = json.loads(response_text)
language = result.get('language', base_language)
framework = result.get('framework')
logger.info(f"Language detection: {language}, Framework: {framework}, Confidence: {result.get('confidence', 0)}")
return language, framework
except Exception as e:
logger.warning(f"AI language detection failed: {e}, using extension-based")
return base_language, None
def analyze_file(self, file_path: str, code: str) -> FileAnalysis:
"""
Perform comprehensive pattern analysis on a single file.
Args:
file_path: Path to the file
code: File contents
Returns:
FileAnalysis object with detected patterns
"""
logger.info(f"Analyzing patterns in {file_path}")
# Check cache
if self.cache_dir:
cache_file = self.cache_dir / f"{hash(file_path + code)}.json"
if cache_file.exists():
try:
cached = json.loads(cache_file.read_text())
return self._deserialize_analysis(cached)
except Exception as e:
logger.warning(f"Cache read failed: {e}")
# Detect language and framework
language, framework = self.detect_language(file_path, code[:500])
# Get relevant patterns for this language
relevant_patterns = self.LEGACY_PATTERNS.get(language, [])
# Build analysis prompt - limit code size to prevent output token overflow
# For large files, we need to be more conservative to leave room for detailed analysis
code_limit = 4000 if len(code) > 6000 else 6000
prompt = f"""You are a senior code auditor. Analyze this code for legacy patterns and modernization opportunities.
FILE: {file_path}
LANGUAGE: {language}
FRAMEWORK: {framework or 'None detected'}
PATTERNS TO CHECK:
{json.dumps(relevant_patterns, indent=2)}
CODE:
```{language.lower()}
{code[:code_limit]}
```
IMPORTANT: Focus on the MOST CRITICAL patterns. Limit your response to the top 10 most important issues.
For each detected pattern, provide:
1. Pattern type (from the list above or new if discovered)
2. Severity (critical/high/medium/low/info)
3. Line numbers where pattern appears (first occurrence only)
4. Confidence score (0.0-1.0)
5. Brief description (max 100 chars)
6. Concise recommendation (max 100 chars)
7. Estimated effort in hours
Also provide:
- Overall modernization score (0-100, where 100 is fully modern)
- Whether modernization is required (true/false)
- Overall priority (critical/high/medium/low/info)
Respond in JSON format:
{{
"patterns": [
{{
"pattern_type": "string",
"severity": "critical|high|medium|low|info",
"line_numbers": [1],
"confidence": 0.95,
"description": "brief description",
"recommendation": "concise fix",
"estimated_effort_hours": 2.5
}}
],
"modernization_score": 65,
"requires_modernization": true,
"overall_priority": "high"
}}
"""
try:
# Use JSON schema for guaranteed structure - no more parsing failures!
# Use LARGE token limit for detailed pattern analysis
schema = GeminiSchemas.pattern_analysis()
response_text = self.ai_manager.generate_content(
prompt=prompt,
temperature=AIManager.TEMPERATURE_PRECISE,
max_tokens=AIManager.MAX_OUTPUT_TOKENS_LARGE,
response_format="json",
response_schema=schema if self.ai_manager.provider_type == "gemini" else None
)
if not response_text:
logger.error(f"Empty response from AI for {file_path}")
raise ValueError(f"Empty response from AI API for {file_path}")
# With JSON schema, response is guaranteed to be valid JSON
result = json.loads(response_text)
logger.info(f"Pattern analysis successful for {file_path}: {len(result.get('patterns', []))} patterns found")
# Convert to DetectedPattern objects
patterns = []
for p in result.get('patterns', []):
patterns.append(DetectedPattern(
pattern_type=p['pattern_type'],
severity=PatternSeverity(p['severity']),
file_path=file_path,
language=language,
description=p['description'],
line_numbers=p.get('line_numbers', []),
confidence=p.get('confidence', 0.8),
recommendation=p['recommendation'],
estimated_effort_hours=p.get('estimated_effort_hours', 1.0)
))
analysis = FileAnalysis(
file_path=file_path,
language=language,
framework=framework,
patterns=patterns,
overall_priority=PatternSeverity(result.get('overall_priority', 'medium')),
modernization_score=result.get('modernization_score', 50),
requires_modernization=result.get('requires_modernization', True)
)
# Cache the result
if self.cache_dir:
try:
cache_file = self.cache_dir / f"{hash(file_path + code)}.json"
cache_file.write_text(json.dumps(self._serialize_analysis(analysis), indent=2))
except Exception as e:
logger.warning(f"Cache write failed: {e}")
logger.info(f"Found {len(patterns)} patterns in {file_path}")
return analysis
except Exception as e:
logger.error(f"Pattern analysis failed for {file_path}: {e}")
# Return minimal analysis on error
return FileAnalysis(
file_path=file_path,
language=language,
framework=framework,
patterns=[],
overall_priority=PatternSeverity.INFO,
modernization_score=100,
requires_modernization=False
)
def analyze_batch(self, files: Dict[str, str], batch_size: int = 3) -> Dict[str, FileAnalysis]:
"""
Analyze multiple files efficiently by batching API calls.
Args:
files: Dictionary mapping file paths to contents
batch_size: Number of files to analyze per API call (default: 3)
Returns:
Dictionary mapping file paths to FileAnalysis objects
"""
logger.info(f"Batch analyzing {len(files)} files with batch_size={batch_size}")
results = {}
file_items = list(files.items())
# Process in batches to reduce API calls
for i in range(0, len(file_items), batch_size):
batch = file_items[i:i + batch_size]
if len(batch) == 1:
# Single file - use individual analysis
file_path, code = batch[0]
try:
analysis = self.analyze_file(file_path, code)
results[file_path] = analysis
except Exception as e:
logger.error(f"Failed to analyze {file_path}: {e}")
else:
# Multiple files - use batch analysis
try:
batch_results = self._analyze_batch_api(batch)
results.update(batch_results)
except Exception as e:
logger.error(f"Batch analysis failed: {e}, falling back to individual")
# Fallback to individual analysis
for file_path, code in batch:
try:
analysis = self.analyze_file(file_path, code)
results[file_path] = analysis
except Exception as e2:
logger.error(f"Failed to analyze {file_path}: {e2}")
logger.info(f"Batch analysis complete: {len(results)} files analyzed")
return results
def _analyze_batch_api(self, batch: List[Tuple[str, str]]) -> Dict[str, FileAnalysis]:
"""
Analyze multiple files in a single API call.
Args:
batch: List of (file_path, code) tuples
Returns:
Dictionary mapping file paths to FileAnalysis objects
"""
logger.info(f"Analyzing {len(batch)} files in single API call")
# Build combined prompt for all files
# Reduce code sample size for batch processing to prevent token overflow
files_info = []
for file_path, code in batch:
ext = Path(file_path).suffix.lower()
language = self.LANGUAGE_PATTERNS.get(ext, 'Unknown')
# Use smaller samples for batch to leave room for multiple file analyses
code_sample_size = 2000 if len(batch) > 2 else 3000
files_info.append({
'file_path': file_path,
'language': language,
'code_sample': code[:code_sample_size]
})
prompt = f"""Analyze these {len(batch)} code files for legacy patterns and modernization opportunities.
For EACH file, provide a complete analysis with patterns, scores, and priorities.
IMPORTANT: Limit to top 5 most critical patterns per file to keep response concise.
FILES TO ANALYZE:
{json.dumps(files_info, indent=2)}
For each file, detect:
- Deprecated libraries and APIs
- Security vulnerabilities (SQL injection, XSS, hardcoded credentials)
- Code quality issues (missing type hints, error handling)
- Performance problems
Keep descriptions and recommendations brief (max 80 chars each).
Respond in JSON format with this structure:
{{
"files": [
{{
"file_path": "file1.py",
"language": "Python",
"framework": "Flask or null",
"patterns": [
{{
"pattern_type": "SQL injection vulnerability",
"severity": "critical",
"line_numbers": [10, 11],
"confidence": 0.95,
"description": "Direct string concatenation in SQL query",
"recommendation": "Use parameterized queries",
"estimated_effort_hours": 2.0
}}
],
"modernization_score": 35,
"requires_modernization": true,
"overall_priority": "critical"
}}
]
}}
"""
try:
# Use JSON schema for guaranteed structure
schema = GeminiSchemas.batch_pattern_analysis()
response_text = self.ai_manager.generate_content(
prompt=prompt,
temperature=AIManager.TEMPERATURE_PRECISE,
max_tokens=AIManager.MAX_OUTPUT_TOKENS_LARGE,
response_format="json",
response_schema=schema if self.ai_manager.provider_type == "gemini" else None
)
# With JSON schema, response is guaranteed to be valid JSON
result = json.loads(response_text)
logger.info(f"Batch analysis successful: received data for {len(result.get('files', []))} files")
# Schema guarantees 'files' key exists
files_data = result.get('files', [])
# Convert to FileAnalysis objects
analyses = {}
for file_data in files_data:
file_path = file_data['file_path']
language = file_data.get('language', 'Unknown')
framework = file_data.get('framework')
patterns = []
for p in file_data.get('patterns', []):
patterns.append(DetectedPattern(
pattern_type=p['pattern_type'],
severity=PatternSeverity(p['severity']),
file_path=file_path,
language=language,
description=p['description'],
line_numbers=p.get('line_numbers', []),
confidence=p.get('confidence', 0.8),
recommendation=p['recommendation'],
estimated_effort_hours=p.get('estimated_effort_hours', 1.0)
))
analysis = FileAnalysis(
file_path=file_path,
language=language,
framework=framework,
patterns=patterns,
overall_priority=PatternSeverity(file_data.get('overall_priority', 'medium')),
modernization_score=file_data.get('modernization_score', 50),
requires_modernization=file_data.get('requires_modernization', True)
)
analyses[file_path] = analysis
logger.info(f"Batch API call successful: analyzed {len(analyses)} files")
return analyses
except Exception as e:
logger.error(f"Batch API call failed: {e}")
raise
def prioritize_files(self, analyses: Dict[str, FileAnalysis]) -> List[Tuple[str, FileAnalysis]]:
"""
Prioritize files for modernization based on analysis.
Args:
analyses: Dictionary of file analyses
Returns:
Sorted list of (file_path, analysis) tuples, highest priority first
"""
# Define priority weights
severity_weights = {
PatternSeverity.CRITICAL: 100,
PatternSeverity.HIGH: 75,
PatternSeverity.MEDIUM: 50,
PatternSeverity.LOW: 25,
PatternSeverity.INFO: 10
}
def calculate_priority_score(analysis: FileAnalysis) -> float:
"""Calculate priority score for an analysis."""
# Base score from overall priority
base_score = severity_weights.get(analysis.overall_priority, 50)
# Add points for each pattern weighted by severity and confidence
pattern_score = sum(
severity_weights.get(p.severity, 25) * p.confidence
for p in analysis.patterns
)
# Factor in modernization score (lower = higher priority)
modernization_penalty = (100 - analysis.modernization_score) / 10
return base_score + pattern_score + modernization_penalty
# Sort by priority score
prioritized = sorted(
analyses.items(),
key=lambda x: calculate_priority_score(x[1]),
reverse=True
)
return prioritized
def generate_report(self, analyses: Dict[str, FileAnalysis]) -> str:
"""
Generate human-readable report from analyses.
Args:
analyses: Dictionary of file analyses
Returns:
Formatted report string
"""
report = []
report.append("=" * 80)
report.append("INTELLIGENT PATTERN MATCHING REPORT")
report.append("=" * 80)
report.append("")
# Summary statistics
total_files = len(analyses)
files_needing_modernization = sum(1 for a in analyses.values() if a.requires_modernization)
total_patterns = sum(len(a.patterns) for a in analyses.values())
avg_modernization_score = sum(a.modernization_score for a in analyses.values()) / max(total_files, 1)
report.append("SUMMARY:")
report.append(f" Total Files Analyzed: {total_files}")
report.append(f" Files Requiring Modernization: {files_needing_modernization}")
report.append(f" Total Patterns Detected: {total_patterns}")
report.append(f" Average Modernization Score: {avg_modernization_score:.1f}/100")
report.append("")
# Language breakdown
language_counts = {}
for analysis in analyses.values():
language_counts[analysis.language] = language_counts.get(analysis.language, 0) + 1
report.append("LANGUAGES:")
for lang, count in sorted(language_counts.items(), key=lambda x: x[1], reverse=True):
report.append(f" {lang}: {count} files")
report.append("")
# Severity breakdown
severity_counts = {s: 0 for s in PatternSeverity}
for analysis in analyses.values():
for pattern in analysis.patterns:
severity_counts[pattern.severity] += 1
report.append("PATTERNS BY SEVERITY:")
for severity in [PatternSeverity.CRITICAL, PatternSeverity.HIGH,
PatternSeverity.MEDIUM, PatternSeverity.LOW, PatternSeverity.INFO]:
count = severity_counts[severity]
if count > 0:
report.append(f" {severity.value.upper()}: {count}")
report.append("")
# Top priority files
prioritized = self.prioritize_files(analyses)[:10]
report.append("TOP 10 PRIORITY FILES:")
for i, (file_path, analysis) in enumerate(prioritized, 1):
report.append(f" {i}. {file_path}")
report.append(f" Priority: {analysis.overall_priority.value}")
report.append(f" Modernization Score: {analysis.modernization_score}/100")
report.append(f" Patterns: {len(analysis.patterns)}")
report.append("")
report.append("=" * 80)
return "\n".join(report)
def _serialize_analysis(self, analysis: FileAnalysis) -> dict:
"""Serialize FileAnalysis to dict for caching."""
return {
'file_path': analysis.file_path,
'language': analysis.language,
'framework': analysis.framework,
'patterns': [
{
'pattern_type': p.pattern_type,
'severity': p.severity.value,
'file_path': p.file_path,
'language': p.language,
'description': p.description,
'line_numbers': p.line_numbers,
'confidence': p.confidence,
'recommendation': p.recommendation,
'estimated_effort_hours': p.estimated_effort_hours
}
for p in analysis.patterns
],
'overall_priority': analysis.overall_priority.value,
'modernization_score': analysis.modernization_score,
'requires_modernization': analysis.requires_modernization
}
def _deserialize_analysis(self, data: dict) -> FileAnalysis:
"""Deserialize dict to FileAnalysis."""
patterns = [
DetectedPattern(
pattern_type=p['pattern_type'],
severity=PatternSeverity(p['severity']),
file_path=p['file_path'],
language=p['language'],
description=p['description'],
line_numbers=p['line_numbers'],
confidence=p['confidence'],
recommendation=p['recommendation'],
estimated_effort_hours=p['estimated_effort_hours']
)
for p in data['patterns']
]
return FileAnalysis(
file_path=data['file_path'],
language=data['language'],
framework=data['framework'],
patterns=patterns,
overall_priority=PatternSeverity(data['overall_priority']),
modernization_score=data['modernization_score'],
requires_modernization=data['requires_modernization']
)