353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""Radon integration for professional complexity analysis."""
|
|
|
|
import ast
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
from radon.complexity import cc_rank, cc_visit
|
|
from radon.metrics import h_visit, mi_visit
|
|
from radon.raw import analyze
|
|
|
|
RADON_AVAILABLE = True
|
|
except ImportError:
|
|
RADON_AVAILABLE = False
|
|
|
|
from .calculator import ComplexityCalculator
|
|
from .metrics import ComplexityMetrics
|
|
|
|
|
|
class RadonComplexityAnalyzer:
|
|
"""Professional complexity analyzer using Radon library."""
|
|
|
|
def __init__(self, fallback_to_manual: bool = True):
|
|
self.fallback_to_manual = fallback_to_manual
|
|
self.manual_calculator = ComplexityCalculator()
|
|
|
|
def analyze_code(self, code: str, filename: str = "<string>") -> ComplexityMetrics:
|
|
"""Analyze code complexity using Radon or fallback to manual calculation."""
|
|
if RADON_AVAILABLE:
|
|
return self._analyze_with_radon(code, filename)
|
|
if self.fallback_to_manual:
|
|
return self.manual_calculator.calculate_complexity(code)
|
|
msg = "Radon is not available and fallback is disabled"
|
|
raise ImportError(msg)
|
|
|
|
def analyze_file(self, file_path: Path) -> ComplexityMetrics:
|
|
"""Analyze complexity of a file."""
|
|
try:
|
|
with open(file_path, encoding="utf-8") as f:
|
|
code = f.read()
|
|
return self.analyze_code(code, str(file_path))
|
|
except (OSError, PermissionError, UnicodeDecodeError):
|
|
# Return empty metrics for unreadable files
|
|
return ComplexityMetrics()
|
|
|
|
def _analyze_with_radon(self, code: str, filename: str) -> ComplexityMetrics: # noqa: ARG002
|
|
"""Analyze code using Radon library."""
|
|
metrics = ComplexityMetrics()
|
|
|
|
try:
|
|
# Raw metrics (lines of code, etc.)
|
|
raw_metrics = analyze(code)
|
|
if raw_metrics:
|
|
metrics.lines_of_code = raw_metrics.loc
|
|
metrics.logical_lines_of_code = raw_metrics.lloc
|
|
metrics.source_lines_of_code = raw_metrics.sloc
|
|
metrics.comment_lines = raw_metrics.comments
|
|
metrics.blank_lines = raw_metrics.blank
|
|
|
|
# Cyclomatic complexity
|
|
cc_results = cc_visit(code)
|
|
if cc_results:
|
|
# Sum up complexity from all functions/methods
|
|
total_complexity = sum(block.complexity for block in cc_results)
|
|
metrics.cyclomatic_complexity = total_complexity
|
|
|
|
# Count functions and classes
|
|
metrics.function_count = len(
|
|
[b for b in cc_results if b.is_method or b.type == "function"],
|
|
)
|
|
metrics.class_count = len([b for b in cc_results if b.type == "class"])
|
|
metrics.method_count = len([b for b in cc_results if b.is_method])
|
|
|
|
# Halstead metrics
|
|
try:
|
|
halstead_data = h_visit(code)
|
|
if halstead_data:
|
|
metrics.halstead_difficulty = halstead_data.difficulty
|
|
metrics.halstead_effort = halstead_data.effort
|
|
metrics.halstead_volume = halstead_data.volume
|
|
metrics.halstead_time = halstead_data.time
|
|
metrics.halstead_bugs = halstead_data.bugs
|
|
except (ValueError, TypeError, AttributeError):
|
|
# Halstead calculation can fail for some code patterns
|
|
pass
|
|
|
|
# Maintainability Index
|
|
try:
|
|
mi_data = mi_visit(code, multi=True)
|
|
if mi_data and hasattr(mi_data, "mi"):
|
|
metrics.maintainability_index = mi_data.mi
|
|
except (ValueError, TypeError, AttributeError):
|
|
# MI calculation can fail, calculate manually
|
|
metrics.maintainability_index = self._calculate_mi_fallback(metrics)
|
|
|
|
# Calculate additional metrics manually
|
|
metrics = self._enhance_with_manual_metrics(code, metrics)
|
|
|
|
except (ValueError, TypeError, SyntaxError, AttributeError):
|
|
# If Radon fails completely, fallback to manual calculation
|
|
if self.fallback_to_manual:
|
|
return self.manual_calculator.calculate_complexity(code)
|
|
raise
|
|
|
|
return metrics
|
|
|
|
def _enhance_with_manual_metrics(
|
|
self,
|
|
code: str,
|
|
metrics: ComplexityMetrics,
|
|
) -> ComplexityMetrics:
|
|
"""Add metrics not provided by Radon using manual calculation."""
|
|
import ast
|
|
|
|
try:
|
|
tree = ast.parse(code)
|
|
|
|
# Calculate cognitive complexity manually
|
|
metrics.cognitive_complexity = self._calculate_cognitive_complexity(tree)
|
|
|
|
# Calculate nesting metrics
|
|
max_depth, avg_depth = self._calculate_nesting_metrics(tree)
|
|
metrics.max_nesting_depth = max_depth
|
|
metrics.average_nesting_depth = avg_depth
|
|
|
|
# Count variables, parameters, returns
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
|
|
metrics.parameters_count += len(node.args.args)
|
|
metrics.returns_count += len(
|
|
[n for n in ast.walk(node) if isinstance(n, ast.Return)],
|
|
)
|
|
|
|
# Count variables
|
|
variables = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Name) and isinstance(
|
|
node.ctx,
|
|
ast.Store | ast.Del,
|
|
):
|
|
variables.add(node.id)
|
|
metrics.variables_count = len(variables)
|
|
|
|
except SyntaxError:
|
|
# If AST parsing fails, keep existing metrics
|
|
pass
|
|
|
|
return metrics
|
|
|
|
def _calculate_cognitive_complexity(self, tree: ast.AST) -> int:
|
|
"""Calculate cognitive complexity manually."""
|
|
complexity = 0
|
|
|
|
def visit_node(node: ast.AST, depth: int = 0) -> None:
|
|
nonlocal complexity
|
|
local_complexity = 0
|
|
|
|
if isinstance(
|
|
node,
|
|
ast.If
|
|
| ast.While
|
|
| ast.For
|
|
| ast.AsyncFor
|
|
| ast.ExceptHandler
|
|
| ast.With,
|
|
):
|
|
local_complexity += 1 + depth
|
|
elif isinstance(node, ast.BoolOp):
|
|
local_complexity += len(node.values) - 1
|
|
elif isinstance(node, ast.Lambda) or (
|
|
isinstance(node, ast.Expr) and isinstance(node.value, ast.IfExp)
|
|
):
|
|
local_complexity += 1
|
|
|
|
complexity += local_complexity
|
|
|
|
# Increase nesting for control structures
|
|
new_depth = (
|
|
depth + 1
|
|
if isinstance(
|
|
node,
|
|
ast.If
|
|
| ast.While
|
|
| ast.For
|
|
| ast.AsyncFor
|
|
| ast.ExceptHandler
|
|
| ast.With,
|
|
)
|
|
else depth
|
|
)
|
|
|
|
for child in ast.iter_child_nodes(node):
|
|
visit_node(child, new_depth)
|
|
|
|
visit_node(tree)
|
|
return complexity
|
|
|
|
def _calculate_nesting_metrics(self, tree: ast.AST) -> tuple[int, float]:
|
|
"""Calculate nesting depth metrics."""
|
|
depths = []
|
|
|
|
def visit_node(node: ast.AST, depth: int = 0) -> None:
|
|
current_depth = depth
|
|
|
|
if isinstance(
|
|
node,
|
|
ast.If | ast.While | ast.For | ast.AsyncFor | ast.With | ast.Try,
|
|
):
|
|
current_depth += 1
|
|
depths.append(current_depth)
|
|
|
|
for child in ast.iter_child_nodes(node):
|
|
visit_node(child, current_depth)
|
|
|
|
visit_node(tree)
|
|
|
|
max_depth = max(depths) if depths else 0
|
|
avg_depth = sum(depths) / len(depths) if depths else 0.0
|
|
|
|
return max_depth, round(avg_depth, 2)
|
|
|
|
def _calculate_mi_fallback(self, metrics: ComplexityMetrics) -> float:
|
|
"""Calculate maintainability index when Radon fails."""
|
|
import math
|
|
|
|
if metrics.halstead_volume <= 0 or metrics.source_lines_of_code <= 0:
|
|
return 100.0
|
|
|
|
try:
|
|
mi = (
|
|
171
|
|
- 5.2 * math.log(metrics.halstead_volume)
|
|
- 0.23 * metrics.cyclomatic_complexity
|
|
- 16.2 * math.log(metrics.source_lines_of_code)
|
|
)
|
|
|
|
return max(0, min(100, round(mi, 2)))
|
|
except (ValueError, ZeroDivisionError):
|
|
return 50.0
|
|
|
|
def get_complexity_rank(self, complexity_score: int) -> str:
|
|
"""Get complexity rank using Radon's ranking system."""
|
|
if not RADON_AVAILABLE:
|
|
# Manual ranking
|
|
if complexity_score <= 5:
|
|
return "A" # Low
|
|
if complexity_score <= 10:
|
|
return "B" # Moderate
|
|
if complexity_score <= 20:
|
|
return "C" # High
|
|
if complexity_score <= 30:
|
|
return "D" # Very High
|
|
return "F" # Extreme
|
|
|
|
return str(cc_rank(complexity_score))
|
|
|
|
def batch_analyze_files(
|
|
self,
|
|
file_paths: list[Path],
|
|
max_workers: int | None = None,
|
|
) -> dict[Path, ComplexityMetrics]:
|
|
"""Analyze multiple files in parallel."""
|
|
import concurrent.futures
|
|
import os
|
|
|
|
if max_workers is None:
|
|
max_workers = os.cpu_count() or 4
|
|
|
|
results = {}
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
|
|
# Submit all tasks
|
|
future_to_path = {
|
|
executor.submit(self.analyze_file, path): path for path in file_paths
|
|
}
|
|
|
|
# Collect results
|
|
for future in concurrent.futures.as_completed(future_to_path):
|
|
path = future_to_path[future]
|
|
try:
|
|
results[path] = future.result()
|
|
except (OSError, PermissionError, UnicodeDecodeError):
|
|
# Create empty metrics for failed files
|
|
results[path] = ComplexityMetrics()
|
|
|
|
return results
|
|
|
|
def get_detailed_complexity_report(
|
|
self,
|
|
code: str,
|
|
filename: str = "<string>",
|
|
) -> dict[str, Any]:
|
|
"""Get detailed complexity report including function-level analysis."""
|
|
if not RADON_AVAILABLE:
|
|
metrics = self.manual_calculator.calculate_complexity(code)
|
|
return {
|
|
"file_metrics": metrics.to_dict(),
|
|
"functions": [],
|
|
"classes": [],
|
|
"radon_available": False,
|
|
}
|
|
|
|
metrics = self._analyze_with_radon(code, filename)
|
|
|
|
# Get function-level details from Radon
|
|
functions = []
|
|
classes = []
|
|
|
|
try:
|
|
cc_results = cc_visit(code)
|
|
for block in cc_results:
|
|
item = {
|
|
"name": block.name,
|
|
"complexity": block.complexity,
|
|
"rank": self.get_complexity_rank(block.complexity),
|
|
"line_number": block.lineno,
|
|
"end_line": getattr(block, "endline", None),
|
|
"type": block.type,
|
|
"is_method": getattr(block, "is_method", False),
|
|
}
|
|
|
|
if block.type == "function" or getattr(block, "is_method", False):
|
|
functions.append(item)
|
|
elif block.type == "class":
|
|
classes.append(item)
|
|
except (ValueError, TypeError, AttributeError):
|
|
pass
|
|
|
|
return {
|
|
"file_metrics": metrics.to_dict(),
|
|
"functions": functions,
|
|
"classes": classes,
|
|
"radon_available": True,
|
|
}
|
|
|
|
@staticmethod
|
|
def is_available() -> bool:
|
|
"""Check if Radon is available."""
|
|
return RADON_AVAILABLE
|
|
|
|
@staticmethod
|
|
def get_radon_version() -> str | None:
|
|
"""Get Radon version if available."""
|
|
if not RADON_AVAILABLE:
|
|
return None
|
|
|
|
try:
|
|
import radon
|
|
|
|
return getattr(radon, "__version__", "unknown")
|
|
except AttributeError:
|
|
return "unknown"
|