智能调试与错误预测的未来_03

import ast
import re
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, precision_score, recall_score
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler
import joblib
import os
from datetime import datetime
import matplotlib.pyplot as plt
import seaborn as sns
from collections import defaultdict
import networkx as nx
from pylint import epylint as lint

# 设置中文显示
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
plt.rcParams['axes.unicode_minus'] = False

class CodeErrorPredictor:
    def __init__(self, model_path=None):
        """初始化代码错误预测器"""
        self.feature_extractor = self._create_feature_extractor()
        if model_path and os.path.exists(model_path):
            self.model = joblib.load(model_path)
        else:
            self.model = self._create_model()
        self.error_database = defaultdict(list)
        self.complexity_metrics = {}
        self.ast_graph = None

    def _create_feature_extractor(self):
        """创建特征提取器"""
        # 定义特征提取管道
        text_features = Pipeline([
            ('tfidf', TfidfVectorizer(ngram_range=(1, 3), max_features=1000))
        ])

        numeric_features = Pipeline([
            ('scaler', StandardScaler())
        ])

        # 列转换器，用于处理不同类型的特征
        preprocessor = ColumnTransformer(
            transformers=[
                ('text', text_features, 'code_text'),
                ('numeric', numeric_features, ['complexity', 'loc', 'cyclomatic_complexity'])
            ])

        return preprocessor

    def _create_model(self):
        """创建错误预测模型"""
        # 创建包含特征提取和分类器的完整管道
        model = Pipeline([
            ('preprocessor', self.feature_extractor),
            ('classifier', RandomForestClassifier(n_estimators=100, max_depth=10, random_state=42))
        ])
        return model

    def _calculate_complexity_metrics(self, code):
        """计算代码复杂度指标"""
        # 解析AST
        try:
            tree = ast.parse(code)
        except SyntaxError:
            return {
                'loc': 0,
                'cyclomatic_complexity': 0,
                'function_count': 0,
                'branch_count': 0,
                'nested_depth': 0
            }

        # 计算代码行数
        loc = len(code.split('\n'))

        # 计算圈复杂度
        cyclomatic_complexity = self._calculate_cyclomatic_complexity(tree)

        # 计算函数数量
        function_count = len([node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))])

        # 计算分支数量
        branch_count = self._count_branches(tree)

        # 计算嵌套深度
        nested_depth = self._calculate_nested_depth(tree)

        metrics = {
            'loc': loc,
            'cyclomatic_complexity': cyclomatic_complexity,
            'function_count': function_count,
            'branch_count': branch_count,
            'nested_depth': nested_depth
        }

        self.complexity_metrics = metrics
        return metrics

    def _calculate_cyclomatic_complexity(self, tree):
        """计算圈复杂度"""
        complexity = 1
        for node in ast.walk(tree):
            if isinstance(node, (ast.If, ast.For, ast.While, ast.And, ast.Or, ast.Assert, ast.ExceptHandler)):
                complexity += 1
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                complexity += 1  # 函数本身算一个分支点
        return complexity

    def _count_branches(self, tree):
        """计算分支数量"""
        branches = 0
        for node in ast.walk(tree):
            if isinstance(node, (ast.If, ast.For, ast.While)):
                branches += 1
            elif isinstance(node, ast.Try):
                branches += len(node.handlers) + (1 if node.finalbody else 0)
        return branches

    def _calculate_nested_depth(self, tree):
        """计算最大嵌套深度"""
        max_depth = 0
        def traverse(node, current_depth):
            nonlocal max_depth
            if isinstance(node, (ast.If, ast.For, ast.While, ast.With, ast.Try)):
                current_depth += 1
                if current_depth > max_depth:
                    max_depth = current_depth
            for child in ast.iter_child_nodes(node):
                traverse(child, current_depth)

        traverse(tree, 0)
        return max_depth

    def _build_ast_graph(self, code):
        """构建AST图用于可视化和分析"""
        try:
            tree = ast.parse(code)
        except SyntaxError:
            return None

        graph = nx.DiGraph()
        node_id = 0
        node_map = {}

        def add_node(node, parent_id=None):
            nonlocal node_id
            node_type = type(node).__name__
            node_label = f"{node_type}"
            if isinstance(node, ast.FunctionDef):
                node_label = f"Function: {node.name}"
            elif isinstance(node, (ast.If, ast.For, ast.While)):
                node_label = node_type

            node_map[id(node)] = node_id
            graph.add_node(node_id, label=node_label, type=node_type)
            current_id = node_id
            node_id += 1

            if parent_id is not None:
                graph.add_edge(parent_id, current_id)

            for child in ast.iter_child_nodes(node):
                add_node(child, current_id)

        add_node(tree)
        self.ast_graph = graph
        return graph

    def analyze_code_quality(self, code):
        """使用pylint分析代码质量"""
        # 将代码写入临时文件
        with open('temp_code.py', 'w', encoding='utf-8') as f:
            f.write(code)

        # 运行pylint
        (pylint_stdout, _) = lint.py_run('temp_code.py --disable=C0114,C0115,C0116', return_std=True)
        result = pylint_stdout.getvalue()

        # 解析结果
        errors = []
        for line in result.split('\n'):
            if 'error' in line.lower() or 'warning' in line.lower():
                match = re.search(r'^(.*?):(\d+):(\d+): (\w\d+): (.*)$', line)
                if match:
                    errors.append({
                        'file': match.group(1),
                        'line': int(match.group(2)),
                        'column': int(match.group(3)),
                        'code': match.group(4),
                        'message': match.group(5)
                    })

        # 删除临时文件
        os.remove('temp_code.py')

        return errors

    def extract_features(self, code):
        """从代码中提取特征"""
        # 计算复杂度指标
        metrics = self._calculate_complexity_metrics(code)

        # 构建特征字典
        features = {
            'code_text': code,
            'complexity': metrics['cyclomatic_complexity'],
            'loc': metrics['loc'],
            'cyclomatic_complexity': metrics['cyclomatic_complexity']
        }

        return pd.DataFrame([features])

    def predict_errors(self, code):
        """预测代码中的潜在错误"""
        # 提取特征
        features = self.extract_features(code)

        # 预测错误概率
        try:
            # 预测错误概率
            error_probability = self.model.predict_proba(features)[0, 1]

            # 预测错误类型
            error_type = self._predict_error_type(code)

            # 识别高风险行
            high_risk_lines = self._identify_high_risk_lines(code)

            return {
                'error_probability': error_probability,
                'risk_level': 'high' if error_probability > 0.7 else 'medium' if error_probability > 0.3 else 'low',
                'predicted_error_type': error_type,
                'high_risk_lines': high_risk_lines,
                'complexity_metrics': self.complexity_metrics
            }
        except Exception as e:
            print(f"Error predicting errors: {e}")
            return {
                'error_probability': 0.0,
                'risk_level': 'unknown',
                'predicted_error_type': 'unknown',
                'high_risk_lines': [],
                'complexity_metrics': self.complexity_metrics
            }

    def _predict_error_type(self, code):
        """预测错误类型"""
        # 简单规则匹配识别常见错误类型
        error_patterns = {
            'NameError': r'\b(print|return|if|for|while)\b.*?\b(\w+)\b.*?=',
            'TypeError': r'\b(int|str|list|dict|float)\b.*?\+(\w+)',
            'IndexError': r'\b(\w+)\[(\d+)\]',
            'KeyError': r'\b(\w+)\[("|\').+?("|\')\]',
            'ValueError': r'\b(int|float|bool)\((\w+)\)',
            'SyntaxError': r':\s*$|\bif\b.*?\bthen\b|\bfor\b.*?\bto\b'
        }

        for error_type, pattern in error_patterns.items():
            if re.search(pattern, code):
                return error_type
        return 'GeneralError'

    def _identify_high_risk_lines(self, code):
        """识别高风险代码行"""
        lines = code.split('\n')
        high_risk_lines = []

        # 构建AST
        try:
            tree = ast.parse(code)
        except SyntaxError as e:
            # 返回语法错误行
            return [e.lineno]

        # 识别包含复杂逻辑的行
        for node in ast.walk(tree):
            if isinstance(node, (ast.If, ast.For, ast.While, ast.Try, ast.With)) and hasattr(node, 'lineno'):
                line_number = node.lineno
                if 1 <= line_number <= len(lines):
                    line_content = lines[line_number - 1].strip()
                    if line_content:
                        high_risk_lines.append({
                            'line_number': line_number,
                            'content': line_content,
                            'risk_reason': f'Complex control structure: {type(node).__name__}'
                        })

        # 识别可能的空指针/未定义变量
        for node in ast.walk(tree):
            if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load):
                # 简单检查是否有赋值
                assigned = any(isinstance(parent, ast.Assign) and any(
                    isinstance(target, ast.Name) and target.id == node.id
                    for target in parent.targets
                ) for parent in ast.walk(tree) if isinstance(parent, ast.Assign))
                if not assigned and hasattr(node, 'lineno'):
                    line_number = node.lineno
                    if 1 <= line_number <= len(lines):
                        line_content = lines[line_number - 1].strip()
                        high_risk_lines.append({
                            'line_number': line_number,
                            'content': line_content,
                            'risk_reason': f'Potential undefined variable: {node.id}'
                        })

        # 去重并按行号排序
        seen_lines = set()
        unique_high_risk = []
        for line in high_risk_lines:
            if line['line_number'] not in seen_lines:
                seen_lines.add(line['line_number'])
                unique_high_risk.append(line)

        return sorted(unique_high_risk, key=lambda x: x['line_number'])

    def generate_fix_suggestions(self, code, error_prediction):
        """生成错误修复建议"""
        if error_prediction['risk_level'] == 'low' and error_prediction['error_probability'] < 0.3:
            return ["代码风险较低，无需特殊修复建议。建议进行常规测试。"]

        # 分析代码质量问题
        quality_errors = self.analyze_code_quality(code)
        if quality_errors:
            return [f"行 {err['line']}: {err['message']} (错误代码: {err['code']})" for err in quality_errors[:5]]

        # 根据预测的错误类型生成修复建议
        error_type = error_prediction['predicted_error_type']
        fix_suggestions = []

        if error_type == 'NameError':
            fix_suggestions.append("可能存在未定义变量。建议检查变量拼写和作用域。")
            fix_suggestions.append("考虑在使用前初始化所有变量，或添加 None 检查。")
        elif error_type == 'TypeError':
            fix_suggestions.append("可能存在类型不匹配问题。建议添加类型检查或转换。")
            fix_suggestions.append("考虑使用类型注解提高代码清晰度和IDE支持。")
        elif error_type == 'IndexError':
            fix_suggestions.append("可能存在索引越界风险。建议添加数组长度检查。")
            fix_suggestions.append("考虑使用 try-except 块捕获索引错误。")
        else:
            fix_suggestions.append("发现潜在错误风险。建议添加详细的单元测试覆盖。")
            fix_suggestions.append("考虑重构复杂代码块，提高可读性和可维护性。")

        # 添加复杂度相关建议
        if error_prediction['complexity_metrics']['cyclomatic_complexity'] > 5:
            fix_suggestions.append(f"代码圈复杂度较高 ({error_prediction['complexity_metrics']['cyclomatic_complexity']})。建议拆分为更小的函数。")

        # 添加高风险行修复建议
        for line in error_prediction['high_risk_lines'][:3]:
            fix_suggestions.append(f"行 {line['line_number']}: {line['risk_reason']}。建议简化或添加测试。")

        return fix_suggestions[:5]

    def train_model(self, dataset_path, save_path=None):
        """使用数据集训练模型"""
        """
        数据集格式应为CSV文件，包含以下列：
        - code: 代码片段
        - has_error: 布尔值，表示代码是否包含错误
        - error_type: 错误类型（可选）
        """
        # 加载数据集
        df = pd.read_csv(dataset_path)
        if 'has_error' not in df.columns:
            raise ValueError("数据集必须包含'has_error'列")

        # 提取特征
        X = []
        y = []
        complexities = []

        for _, row in df.iterrows():
            code = row['code']
            has_error = row['has_error']
            metrics = self._calculate_complexity_metrics(code)
            complexities.append({
                'complexity': metrics['cyclomatic_complexity'],
                'loc': metrics['loc'],
                'cyclomatic_complexity': metrics['cyclomatic_complexity']
            })
            X.append({'code_text': code, **complexities[-1]})
            y.append(1 if has_error else 0)

        # 转换为DataFrame
        X_df = pd.DataFrame(X)
        y = np.array(y)

        # 分割训练集和测试集
        X_train, X_test, y_train, y_test = train_test_split(X_df, y, test_size=0.2, random_state=42)

        # 训练模型
        self.model.fit(X_train, y_train)

        # 评估模型
        y_pred = self.model.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)
        precision = precision_score(y_test, y_pred)
        recall = recall_score(y_test, y_pred)

        print(f"模型评估结果: Accuracy={accuracy:.4f}, Precision={precision:.4f}, Recall={recall:.4f}")

        # 保存模型
        if save_path:
            joblib.dump(self.model, save_path)
            print(f"模型已保存至: {save_path}")

        return {
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall
        }

    def visualize_code_analysis(self, code, output_path='code_analysis.png'):
        """可视化代码分析结果"""
        # 预测错误
        error_prediction = self.predict_errors(code)

        # 创建可视化
        fig, axes = plt.subplots(1, 2, figsize=(15, 7))

        # 第一个子图：风险评估
        risk_levels = ['Low', 'Medium', 'High']
        risk_values = [0, 0, 0]
        if error_prediction['risk_level'] == 'low':
            risk_values[0] = error_prediction['error_probability']
            risk_values[1] = 1 - error_prediction['error_probability']
        elif error_prediction['risk_level'] == 'medium':
            risk_values[0] = 0.3
            risk_values[1] = error_prediction['error_probability'] - 0.3
            risk_values[2] = 1 - error_prediction['error_probability']
        else:
            risk_values[0] = 0.3
            risk_values[1] = 0.4
            risk_values[2] = error_prediction['error_probability'] - 0.7

        axes[0].bar(risk_levels, risk_values, color=['#4CAF50', '#FFC107', '#F44336'])
        axes[0].set_title('代码错误风险评估')
        axes[0].set_ylabel('风险概率')
        axes[0].set_ylim(0, 1)
        axes[0].text(0.5, -0.15, f'预测错误类型: {error_prediction["predicted_error_type"]}', 
                     ha='center', va='center', transform=axes[0].transAxes)

        # 第二个子图：复杂度指标
        metrics = error_prediction['complexity_metrics']
        metric_names = ['代码行数', '圈复杂度', '函数数量', '分支数量', '嵌套深度']
        metric_values = [
            metrics.get('loc', 0),
            metrics.get('cyclomatic_complexity', 0),
            metrics.get('function_count', 0),
            metrics.get('branch_count', 0),
            metrics.get('nested_depth', 0)
        ]

        axes[1].barh(metric_names, metric_values, color='#2196F3')
        axes[1].set_title('代码复杂度指标')
        axes[1].set_xlabel('值')

        plt.tight_layout()
        plt.savefig(output_path, dpi=300, bbox_inches='tight')
        plt.close()

        return output_path

# 使用示例
if __name__ == '__main__':
    # 创建错误预测器实例
    error_predictor = CodeErrorPredictor()

    # 示例代码（包含潜在错误）
    sample_code = """
def calculate_average(numbers):
    total = 0
    for i in range(len(numbers)):
        total += numbers[i+1]
    return total / len(numbers)

# 使用示例
data = [1, 2, 3, 4]
result = calculate_average(data)
print("平均值:", result)"""

    # 分析代码
    print("=== 代码分析结果 ===")
    error_prediction = error_predictor.predict_errors(sample_code)
    print(f"错误概率: {error_prediction['error_probability']:.2f}")
    print(f"风险等级: {error_prediction['risk_level']}")
    print(f"预测错误类型: {error_prediction['predicted_error_type']}")

    # 显示高风险行
    if error_prediction['high_risk_lines']:
        print("\n高风险代码行:")
        for line in error_prediction['high_risk_lines']:
            print(f"行 {line['line_number']}: {line['content']} ({line['risk_reason']})")

    # 生成修复建议
    print("\n=== 修复建议 ===")
    fix_suggestions = error_predictor.generate_fix_suggestions(sample_code, error_prediction)
    for i, suggestion in enumerate(fix_suggestions, 1):
        print(f"{i}. {suggestion}")

    # 可视化分析结果
    visualization_path = error_predictor.visualize_code_analysis(sample_code)
    print(f"\n分析可视化已保存至: {visualization_path}")

    # 注意：在实际应用中，应使用大量带标签的代码数据训练模型
    # error_predictor.train_model('code_error_dataset.csv', 'error_prediction_model.pkl')