insightflow/auto_fix_code.py

#!/usr/bin/env python3
"""
InsightFlow 代码自动修复脚本 - 增强版
自动修复代码中的常见问题
"""

import json
import os
import re
import subprocess
from pathlib import Path


def run_ruff_check(directory: str) -> list[dict]:
    """运行 ruff 检查并返回问题列表"""
    try:
        result = subprocess.run(
            ["ruff", "check", "--select=E,W,F,I", "--output-format=json", directory],
            capture_output=True,
            text=True,
            check=False,
        )
        if result.stdout:
            return json.loads(result.stdout)
        return []
    except Exception as e:
        print(f"Ruff check failed: {e}")
        return []


def fix_bare_except(content: str) -> str:
    """修复裸异常捕获 - 将 bare except: 改为 except Exception:"""
    pattern = r'except\s*:\s*\n'
    replacement = 'except Exception:\n'
    return re.sub(pattern, replacement, content)


def fix_undefined_names(content: str, filepath: str) -> str:
    """修复未定义的名称"""
    lines = content.split('\n')
    modified = False
    
    import_map = {
        'ExportEntity': 'from export_manager import ExportEntity',
        'ExportRelation': 'from export_manager import ExportRelation',
        'ExportTranscript': 'from export_manager import ExportTranscript',
        'WorkflowManager': 'from workflow_manager import WorkflowManager',
        'PluginManager': 'from plugin_manager import PluginManager',
        'OpsManager': 'from ops_manager import OpsManager',
        'urllib': 'import urllib.parse',
    }
    
    undefined_names = set()
    for name, import_stmt in import_map.items():
        if name in content and import_stmt not in content:
            undefined_names.add((name, import_stmt))
    
    if undefined_names:
        import_idx = 0
        for i, line in enumerate(lines):
            if line.startswith('import ') or line.startswith('from '):
                import_idx = i + 1
        
        for name, import_stmt in sorted(undefined_names):
            lines.insert(import_idx, import_stmt)
            import_idx += 1
            modified = True
    
    if modified:
        return '\n'.join(lines)
    return content


def fix_file(filepath: str, issues: list[dict]) -> tuple[bool, list[str], list[str]]:
    """修复单个文件的问题"""
    with open(filepath, 'r', encoding='utf-8') as f:
        original_content = f.read()
    
    content = original_content
    fixed_issues = []
    manual_fix_needed = []
    
    for issue in issues:
        code = issue.get('code', '')
        message = issue.get('message', '')
        line_num = issue['location']['row']
        
        if code == 'F821':
            content = fix_undefined_names(content, filepath)
            if content != original_content:
                fixed_issues.append(f"F821 - {message} (line {line_num})")
            else:
                manual_fix_needed.append(f"F821 - {message} (line {line_num})")
        elif code == 'E501':
            manual_fix_needed.append(f"E501 (line {line_num})")
    
    content = fix_bare_except(content)
    
    if content != original_content:
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content)
        return True, fixed_issues, manual_fix_needed
    
    return False, fixed_issues, manual_fix_needed


def main():
    base_dir = Path("/root/.openclaw/workspace/projects/insightflow")
    backend_dir = base_dir / "backend"
    
    print("=" * 60)
    print("InsightFlow 代码自动修复")
    print("=" * 60)
    
    print("\n1. 扫描代码问题...")
    issues = run_ruff_check(str(backend_dir))
    
    issues_by_file = {}
    for issue in issues:
        filepath = issue.get('filename', '')
        if filepath not in issues_by_file:
            issues_by_file[filepath] = []
        issues_by_file[filepath].append(issue)
    
    print(f"   发现 {len(issues)} 个问题，分布在 {len(issues_by_file)} 个文件中")
    
    issue_types = {}
    for issue in issues:
        code = issue.get('code', 'UNKNOWN')
        issue_types[code] = issue_types.get(code, 0) + 1
    
    print("\n2. 问题类型统计:")
    for code, count in sorted(issue_types.items(), key=lambda x: -x[1]):
        print(f"   - {code}: {count} 个")
    
    print("\n3. 尝试自动修复...")
    fixed_files = []
    all_fixed_issues = []
    all_manual_fixes = []
    
    for filepath, file_issues in issues_by_file.items():
        if not os.path.exists(filepath):
            continue
        
        modified, fixed, manual = fix_file(filepath, file_issues)
        if modified:
            fixed_files.append(filepath)
        all_fixed_issues.extend(fixed)
        all_manual_fixes.extend([(filepath, m) for m in manual])
    
    print(f"   直接修改了 {len(fixed_files)} 个文件")
    print(f"   自动修复了 {len(all_fixed_issues)} 个问题")
    
    print("\n4. 运行 ruff 自动格式化...")
    try:
        subprocess.run(
            ["ruff", "format", str(backend_dir)],
            capture_output=True,
            check=False,
        )
        print("   格式化完成")
    except Exception as e:
        print(f"   格式化失败: {e}")
    
    print("\n5. 再次检查...")
    remaining_issues = run_ruff_check(str(backend_dir))
    print(f"   剩余 {len(remaining_issues)} 个问题需要手动处理")
    
    report = {
        'total_issues': len(issues),
        'fixed_files': len(fixed_files),
        'fixed_issues': len(all_fixed_issues),
        'remaining_issues': len(remaining_issues),
        'issue_types': issue_types,
        'manual_fix_needed': all_manual_fixes[:30],
    }
    
    return report


if __name__ == "__main__":
    report = main()
    print("\n" + "=" * 60)
    print("修复报告")
    print("=" * 60)
    print(f"总问题数: {report['total_issues']}")
    print(f"修复文件数: {report['fixed_files']}")
    print(f"自动修复问题数: {report['fixed_issues']}")
    print(f"剩余问题数: {report['remaining_issues']}")
    print(f"\n需要手动处理的问题 (前30个):")
    for filepath, issue in report['manual_fix_needed']:
        print(f"  - {filepath}: {issue}")