code-review基础方法

ai_review/analyzers/ast_parser.py

@@ -21,4 +21,7 @@ def visit_Assign(self, node):
         # 检测未使用变量
         if isinstance(node.targets[0].ctx, ast.Store):
             if not any(ref.id == node.targets[0].id for ref in self.references):
-                self.findings.append(f"未使用变量: {node.targets[0].id}")
+                self.findings.append(f"未使用变量: {node.targets[0].id}")
+
+    def visit_Assert(self, node: ast.Assert) -> ast.Any:
+        return super().visit_Assert(node)

ai_review/cli/main.py

@@ -1,27 +1,54 @@
-from analyzers.ast_parser import ASTAnalyzer
-from rules import style, security
+# from analyzers.ast_parser import ASTAnalyzer
+# from rules import style, security
 
-def code_review(file_path):
-    # AST解析
-    with open(file_path) as f:
-        tree = ast.parse(f.read())
+# def code_review(file_path):
+#     # AST解析
+#     with open(file_path) as f:
+#         tree = ast.parse(f.read())
+
+#     # 执行静态分析
+#     analyzer = ASTAnalyzer()
+#     analyzer.visit(tree)
+
+#     # 应用规则检查
+#     style_violations = StyleRules.check_naming_convention(tree)
+#     security_issues = SecurityRules.check_injection(tree)
 
-    # 执行静态分析
-    analyzer = ASTAnalyzer()
-    analyzer.visit(tree)
+#     # AI增强审查
+#     if config['ai']['enable']:
+#         ai_reviewer = AICodeReviewer(config['ai']['api_key'])
+#         ai_suggestions = ai_reviewer.get_optimization_suggestion(code_snippet)
+
+#     # 生成报告
+#     generate_report({
+#         'static_analysis': analyzer.findings,
+#         'style_violations': style_violations,
+#         'ai_suggestions': ai_suggestions
+#     })
+import click
+from ai_review.core.analyzer import CodeAnalyzer
+from ai_review.core.model_adapter import AIModelHandler
+
+@click.command()
+@click.argument('file_path')
+@click.option('--ai', is_flag=True, help='启用AI增强审查')
+def review(file_path: str, ai: bool):
+    """执行代码审查流水线"""
+    with open(file_path) as f:
+        code = f.read()
 
-    # 应用规则检查
-    style_violations = StyleRules.check_naming_convention(tree)
-    security_issues = SecurityRules.check_injection(tree)
+    # 静态规则审查
+    analyzer = CodeAnalyzer(['ai_review.rules.security_rules'])
+    findings = analyzer.analyze(code)
 
-    # AI增强审查
-    if config['ai']['enable']:
-        ai_reviewer = AICodeReviewer(config['ai']['api_key'])
-        ai_suggestions = ai_reviewer.get_optimization_suggestion(code_snippet)
+    # AI增强分析
+    if ai:
+        ai_handler = AIModelHandler()
+        ai_comment = ai_handler.get_code_review(code, {'findings': findings})
+        click.echo(f"\nAI审查建议:\n{ai_comment}")
 
-    # 生成报告
-    generate_report({
-        'static_analysis': analyzer.findings,
-        'style_violations': style_violations,
-        'ai_suggestions': ai_suggestions
-    })
+    # 输出格式化结果
+    click.echo("\n静态审查结果:")
+    for issue in findings:
+        click.secho(f"[{issue['severity'].upper()}] Line {issue['line']}: {issue['message']}", 
+                   fg='red' if issue['severity'] == 'critical' else 'yellow')

ai_review/config/model_config.yaml

@@ -0,0 +1,10 @@
+openai:
+  api_key: ${OPENAI_API_KEY}
+  model_name: gpt-4-turbo
+  temperature: 0.3
+  max_tokens: 1024
+anthropic:
+  api_key: ${ANTHROPIC_API_KEY}
+  model_name: claude-3-opus
+local_models:
+  codegen2_path: /models/codegen2-3.7B

ai_review/core/analyzer.py

@@ -0,0 +1,64 @@
+from typing import Dict, List
+from .parser import ASTParser
+
+import importlib
+from typing import Dict, List, Callable
+
+class CodeAnalyzer:
+    def __init__(self, rule_modules: List[str]):
+        """Initialize the analyzer with rule modules."""
+        if not rule_modules:
+            raise ValueError("No rule modules specified")
+        self.rules = self._load_rules(rule_modules)
+
+    def _load_rules(self, modules: List[str]) -> Dict[str, Callable]:
+        """动态加载规则检测器"""
+        rules = {}
+        for module_name in modules:
+            try:
+                module = importlib.import_module(module_name)
+                for rule_name in dir(module):
+                    if rule_name.startswith('_'):
+                        continue
+                    rule = getattr(module, rule_name)
+                    if callable(rule):
+                        rules[rule_name] = rule
+            except ImportError as e:
+                raise ImportError(f"Failed to import rule module {module_name}: {e}")
+            except Exception as e:
+                raise RuntimeError(f"Error loading rules from {module_name}: {e}")
+        return rules
+    def analyze(self, code: str) -> List[dict]:
+        """执行多维度代码审查"""
+        try:
+            ast_parser = ASTParser(code)
+        except ValueError as e:
+            return [{
+                'rule': 'syntax_check',
+                'message': str(e),
+                'severity': 'critical',
+                'line': 0
+            }]
+
+        findings = []
+
+        # 执行静态规则检查
+        for rule_name, check_func in self.rules.items():
+            try:
+                if issues := check_func(ast_parser.tree):
+                    findings.extend({
+                        'rule': rule_name,
+                        'message': issue['msg'],
+                        'severity': issue['level'],
+                        'line': issue['lineno'],
+                        'code_snippet': ast_parser.raw_code.splitlines()[issue['lineno']-1]
+                    } for issue in issues)
+            except Exception as e:
+                findings.append({
+                    'rule': rule_name,
+                    'message': f"Rule execution failed: {e}",
+                    'severity': 'error',
+                    'line': 0
+                })
+
+        return findings

ai_review/core/model_adapter.py

@@ -0,0 +1,28 @@
+import openai
+from configparser import ConfigParser
+
+class AIModelHandler:
+    def __init__(self, config_path='model_config.yaml'):
+        self.config = self._load_config(config_path)
+
+    def _load_config(self, path):
+        with open(path) as f:
+            return yaml.safe_load(f)
+
+    def get_code_review(self, code_snippet: str, context: dict) -> str:
+        """调用AI模型进行语义级审查"""
+        prompt = f"""作为资深Python架构师，请审查以下代码：
+{code_snippet}
+审查重点：
+1. 架构设计合理性
+2. 异常处理完整性
+3. 性能优化空间
+4. 安全合规性
+请用中文按严重性分级输出建议"""
+
+        response = openai.ChatCompletion.create(
+            model=self.config['openai']['model_name'],
+            messages=[{"role": "user", "content": prompt}],
+            temperature=self.config['openai']['temperature']
+        )
+        return response.choices[0].message.content

ai_review/core/parser.py

@@ -0,0 +1,30 @@
+import ast
+import astunparse
+
+class ASTParser:
+    def __init__(self, source_code: str):
+        self.raw_code = source_code
+        try:
+            self.tree = ast.parse(source_code)
+        except SyntaxError as e:
+            raise ValueError(f"Invalid Python syntax at line {e.lineno}: {e.msg}")
+    def get_function_defs(self) -> list:
+        """提取所有函数定义元数据"""
+        return [{
+            'name': node.name,
+            'args': [arg.arg for arg in node.args.args],
+            'lineno': node.lineno,
+            'docstring': ast.get_docstring(node)
+        } for node in ast.walk(self.tree) if isinstance(node, ast.FunctionDef)]
+
+    def get_class_hierarchy(self) -> list:
+        """解析类继承结构"""
+        return [{
+            'name': node.name,
+            'bases': [base.id for base in node.bases if isinstance(base, ast.Name)],
+            'methods': [n.name for n in node.body if isinstance(n, ast.FunctionDef)]
+        } for node in ast.walk(self.tree) if isinstance(node, ast.ClassDef)]
+
+    def code_visualization(self) -> str:
+        """生成AST可视化结构"""
+        return astunparse.dump(self.tree)

ai_review/rules/security_rules.py

@@ -0,0 +1,28 @@
+import ast
+
+def detect_unsafe_functions(tree) -> list:
+    """识别高危函数调用"""
+    unsafe_calls = {
+        'eval', 'exec', 'pickle.loads', 'pickle.load',
+        'subprocess.call', 'subprocess.Popen', 'os.system',
+        'tempfile.mktemp'  # Use mkstemp instead
+    }
+    return [{
+        'msg': f"检测到不安全函数调用: {node.func.id}",
+        'level': 'critical',
+        'lineno': node.lineno
+    } for node in ast.walk(tree) 
+     if isinstance(node, ast.Call) 
+     and hasattr(node.func, 'id') 
+     and node.func.id in unsafe_calls]
+
+def check_hardcoded_secrets(tree) -> list:
+    """检测硬编码密钥"""
+    secret_patterns = {'password', 'secret_key', 'api_key'}
+    return [{
+        'msg': f"疑似硬编码凭证: {node.targets[0].id}",
+        'level': 'high',
+        'lineno': node.lineno
+    } for node in ast.walk(tree) 
+     if isinstance(node, ast.Assign)
+     and any(pattern in node.targets[0].id.lower() for pattern in secret_patterns)]

requirements.txt

@@ -14,12 +14,18 @@ tree-sitter==0.20.2
 # AI模型集成
 transformers==4.37.2
 torch==2.2.0
-openai==1.12.0
-
 # 代码格式化
 black==23.12.0
 isort==5.13.2
 
 # 项目构建
 setuptools==69.0.3
-wheel==0.42.0
+wheel==0.42.0
+
+astunparse==1.6.3
+PyYAML==6.0.1
+openai>=1.12.0
+click==8.1.7
+tqdm==4.66.2
+requests==2.31.0
+typing_extensions==4.9.0