Bosch_Demo/code_scan
4
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

代码测试

Browse Source
This commit is contained in:
Dang Zerong
2026-03-13 11:26:01 +08:00
parent 8f9e5bf4f5
commit cb90b66f09
6 changed files with 1291 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

438
scanner/ai_reviewer.py
View File

@@ -5,6 +5,7 @@ AI 代码审查器
使用大模型进行智能代码审查
"""
import os
import re
import json
import logging
from typing import Dict, Any, List, Optional
@@ -73,15 +74,26 @@ class AIReviewer(BaseScanner):
changed_files: 可选的变更文件列表（来自 PR）
Returns:
审查结果
审查结果（与 python_scanner.py 兼容的格式）
"""
result = {
'tool': 'AI Code Reviewer',
'language': language,
'status': 'success',
'issues': [],
'summary': {
'total': 0,
'error': 0,
'warning': 0,
'info': 0
},
'files_scanned': 0
}
if not self.enabled:
return {
'enabled': False,
'tool': 'AI Code Reviewer',
'reviews': [],
'summary': 'AI 审查已禁用'
}
result['status'] = 'disabled'
result['summary'] = 'AI 审查已禁用'
return result
try:
# 如果没有传入 clone_dir，需要克隆
@@ -89,52 +101,141 @@ class AIReviewer(BaseScanner):
clone_dir = self.clone_repo(repo_url, commit_id, branch)
if not clone_dir or not os.path.exists(clone_dir):
return {
'enabled': True,
'tool': 'AI Code Reviewer',
'reviews': [],
'summary': '无法获取代码目录'
}
result['status'] = 'error'
result['error'] = '无法获取代码目录'
return result
# 获取要审查的代码文件
files = self._get_code_files(clone_dir, language, changed_files)
if not files:
return {
'enabled': True,
'tool': 'AI Code Reviewer',
'reviews': [],
'summary': '未找到可审查的代码文件'
}
result['summary'] = '未找到可审查的代码文件'
return result
# 对每个文件进行 AI 审查
all_reviews = []
all_issues = []
for file_path in files[:5]: # 限制最多审查 5 个文件
review = self._review_file(file_path, language, clone_dir)
if review:
all_reviews.append(review)
if review and review.get('issues'):
all_issues.extend(review['issues'])
# 生成总结
summary = self._generate_summary(all_reviews)
result['issues'] = all_issues[:self.max_issues] if self.detailed else all_issues
result['summary'] = self._calculate_summary(all_issues)
result['files_scanned'] = len(files[:5])
result['clone_dir'] = clone_dir
return {
'enabled': True,
'tool': 'AI Code Reviewer',
'reviews': all_reviews,
'summary': summary,
'files_reviewed': len(all_reviews),
'clone_dir': clone_dir # 返回 clone_dir 用于后续清理
}
# 生成质量评分
result['quality_score'] = self._calculate_quality_score(all_issues, files[:5])
return result
except Exception as e:
logger.error(f'AI 审查失败: {str(e)}')
return {
'enabled': True,
'tool': 'AI Code Reviewer',
'error': str(e),
'reviews': [],
'summary': f'AI 审查出错: {str(e)}'
result['status'] = 'error'
result['error'] = str(e)
return result
def _calculate_summary(self, issues: List[Dict]) -> Dict[str, int]:
"""计算问题摘要"""
summary = {
'total': len(issues),
'error': 0,
'warning': 0,
'info': 0
}
for issue in issues:
severity = issue.get('severity', '').lower()
if severity in ['error', 'critical', 'fatal']:
summary['error'] += 1
elif severity in ['warning', 'moderate']:
summary['warning'] += 1
else:
summary['info'] += 1
return summary
def _calculate_quality_score(self, issues: List[Dict], files: List[str]) -> Dict[str, Any]:
"""
计算代码质量评分
返回：总分(0-100)及各维度评分
"""
if not files:
return {'total': 100, 'maintainability': 100, 'security': 100, 'readability': 100, 'best_practices': 100}
# 统计问题
error_count = sum(1 for i in issues if i.get('severity', '').lower() in ['error', 'critical'])
warning_count = sum(1 for i in issues if i.get('severity', '').lower() == 'warning')
info_count = sum(1 for i in issues if i.get('severity', '').lower() == 'info')
# 分类统计
security_keywords = ['sql injection', 'xss', 'csrf', 'password', 'secret', 'token', '权限', '注入', '认证']
security_issues = sum(1 for i in issues if any(k in (i.get('message', '') + i.get('symbol', '')).lower() for k in security_keywords))
# 计算各维度分数
# 可维护性：基于错误和警告数量
issue_weight = error_count * 5 + warning_count * 2 + info_count * 0.5
maintainability = max(0, 100 - issue_weight)
# 安全性：基于安全问题
security_score = max(0, 100 - security_issues * 15)
# 可读性：基于 info 级别问题（风格类）
readability = max(0, 100 - info_count * 3)
# 最佳实践：基于 warning 级别
best_practices = max(0, 100 - warning_count * 5)
# 总分：加权平均
total = int((maintainability * 0.3 + security_score * 0.35 + readability * 0.15 + best_practices * 0.2))
return {
'total': total,
'maintainability': maintainability,
'security': security_score,
'readability': readability,
'best_practices': best_practices,
'details': {
'error_count': error_count,
'warning_count': warning_count,
'info_count': info_count,
'security_issues': security_issues
}
}
def generate_fix_suggestion(self, file_path: str, line: int, message: str, code: str) -> Optional[str]:
"""
对指定问题生成修复建议代码
"""
prompt = f"""你是一位代码修复专家。请根据以下问题，生成修复后的代码。
问题描述：{message}
问题所在行号：{line}
原始代码：
```
{code}
```
请以 JSON 格式输出修复建议：
```json
{{
"fixed_code": "修复后的完整代码或关键片段",
"explanation": "修复说明（50字以内）",
"confidence": "high/medium/low 修复把握度"
}}
```
如果无法修复，请返回：{{"fixed_code": "", "explanation": "无法自动修复", "confidence": "low"}}"""
try:
response = self._call_ai(prompt)
if response and response.get('fixed_code'):
return response
except Exception as e:
logger.warning(f'生成修复建议失败: {e}')
return None
def _get_code_files(self, clone_dir: str, language: str, changed_files: Optional[List[str]] = None) -> List[str]:
"""获取代码文件列表"""
@@ -174,6 +275,8 @@ class AIReviewer(BaseScanner):
def _review_file(self, file_path: str, language: str, clone_dir: str = None) -> Optional[Dict[str, Any]]:
"""审查单个文件"""
issues = []
try:
with open(file_path, 'r', encoding='utf-8') as f:
code = f.read()
@@ -186,22 +289,46 @@ class AIReviewer(BaseScanner):
else:
truncated = False
# 构建 prompt
prompt = self._build_prompt(code, language)
# 给代码加行号再发给模型，便于模型返回准确行号
code_with_lines = self._code_with_line_numbers(code)
prompt = self._build_prompt(code_with_lines, language)
# 调用 AI
response = self._call_ai(prompt)
if not response:
return None
# 解析响应
# 获取相对路径
rel_path = os.path.relpath(file_path, clone_dir) if (clone_dir and file_path) else file_path
if not response:
return {
'file': rel_path,
'path': file_path,
'truncated': truncated,
'issues': []
}
# 解析 AI 响应，转换为标准 issues 格式，并校正行号
ai_issues = response.get('issues', [])
for issue in ai_issues:
self._correct_issue_line(issue, code)
issues.append({
'tool': 'ai_reviewer',
'type': issue.get('type', 'info'),
'severity': issue.get('severity', 'Info'),
'message': issue.get('message', ''),
'file': rel_path,
'line': issue.get('line', 0),
'column': issue.get('column', 0),
'symbol': issue.get('symbol', ''),
'code_context': issue.get('code_context', ''),
'defect_reason': issue.get('defect_reason', '')
})
return {
'file': rel_path,
'path': file_path,
'truncated': truncated,
'review': response
'issues': issues
}
except Exception as e:
@@ -217,29 +344,83 @@ class AIReviewer(BaseScanner):
else:
lang_name = language
prompt = f"""你是一位资深的 {lang_name} 代码审查专家。请审查以下代码，并给出：
prompt = f"""你是一位资深的 {lang_name} 代码审查专家。请审查以下代码，找出潜在的问题和缺陷。
1. **代码优点** - 写得好地方
2. **问题建议** - 需要改进的地方
3. **优化建议** - 如何让代码更好
请用中文回复，保持简洁，每个文件审查不超过 3 点建议。
以下是代码：
```{language}
{code}
```
请以 JSON 格式输出：
请以 JSON 格式输出审查结果，必须包含以下字段：
```json
{{
"优点": ["..."],
"问题": ["..."],
"优化": ["..."]
"issues": [
{{
"line": 行号,
"column": 列号,
"message": "问题描述",
"type": "error/warning/info 之一",
"severity": "Error/Warning/Info 之一",
"symbol": "错误标识符如 unused-variable, syntax-error 等",
"code_context": "问题代码的上下文（包含问题的那行或几行代码）",
"defect_reason": "缺陷原因分析（30字以内简洁描述）"
}}
]
}}
```
注意：
1. line 和 column 是问题所在的行号和列号（从 1 开始）
2. type: error=错误, warning=警告, info=信息
3. severity: Error=严重, Warning=一般, Info=提示
4. code_context: 包含问题代码的那一行或相邻的几行
5. defect_reason: 精简描述，30字以内，说明问题原因和风险
如果代码没有问题，返回空数组: {{"issues": []}}
重要：以下代码每行前已标注行号（格式为 "行号|"），请根据问题实际出现的代码行，严格使用该行前的行号填写 issues 中的 line 字段，不要猜测或使用错误行号。
以下是待审查的代码（行号已标注）：
```{language}
{code}
```"""
return prompt
def _code_with_line_numbers(self, code: str) -> str:
"""给代码每行前加上行号，便于模型返回准确行号"""
lines = code.split('\n')
width = len(str(len(lines)))
return '\n'.join(f'{i:>{width}}| {line}' for i, line in enumerate(lines, 1))
def _correct_issue_line(self, issue: Dict[str, Any], code: str) -> None:
"""
根据 message/symbol 在源码中搜索，尽量把 issue 的 line 校正到真实出现位置。
AI 返回的行号常不准确，通过匹配问题相关的标识符（如 'unused_module'）修正行号。
"""
line = issue.get('line')
if not line or not code:
return
lines = code.split('\n')
if line < 1 or line > len(lines):
return
# 从 message 中提取被引用的标识符（如 'unused_module' -> unused_module）
message = (issue.get('message') or '')
symbol = (issue.get('symbol') or '').strip()
candidates = []
if symbol:
candidates.append(symbol)
for m in re.finditer(r"['\"]([a-zA-Z_][a-zA-Z0-9_]*)['\"]", message or ''):
candidates.append(m.group(1))
# 若 message 里没有引号标识符，取首段英文/数字/下划线作为关键词
if not candidates:
first_word = re.search(r'\b([a-zA-Z_][a-zA-Z0-9_]*)\b', message)
if first_word:
candidates.append(first_word.group(1))
for token in candidates:
if not token:
continue
for i, code_line in enumerate(lines):
if token in code_line:
issue['line'] = i + 1
return
def _call_ai(self, prompt: str) -> Optional[Dict[str, Any]]:
"""调用 AI 服务"""
try:
@@ -255,6 +436,87 @@ class AIReviewer(BaseScanner):
logger.error(f'AI 调用失败: {str(e)}')
return None
def _extract_json_obj(self, content: Any) -> Optional[Dict[str, Any]]:
"""
从模型输出中尽可能提取 JSON 对象(dict)。
兼容场景：
- content 已经是 dict
- content 是 JSON 字符串
- content 被 ```json ... ``` 或 ``` ... ``` 包裹
- content 前后夹杂说明文字，只要包含一个最外层 { ... } 就尝试解析
"""
if content is None:
logger.debug("_extract_json_obj: content is None")
return None
# 如果已经是 dict，直接返回
if isinstance(content, dict):
logger.debug("_extract_json_obj: content is already dict")
return content
if not isinstance(content, str):
content = str(content)
text = content.strip()
logger.debug(f"_extract_json_obj: 原始内容长度 = {len(text)}")
logger.debug(f"_extract_json_obj: 原始内容前100字符: {text[:100]}")
# 去掉代码块包裹（兼容 ```json / ``` json / ```JSON 等）
lowered = text.lower()
fence_start = lowered.find('```')
if fence_start != -1:
logger.debug(f"_extract_json_obj: 发现代码块 fence_start={fence_start}")
# 找到第一段 fence
after = text[fence_start + 3:]
after_l = after.lower()
# 如果 fence 后紧跟语言标识（json 或其他），跳过这一行直到换行
newline_idx = after.find('\n')
if newline_idx != -1:
lang_header = after_l[:newline_idx].strip()
logger.debug(f"_extract_json_obj: 语言标识: {lang_header}")
body = after[newline_idx + 1:]
# 截取到下一个 fence 结束
end_idx = body.lower().find('```')
if end_idx != -1:
candidate = body[:end_idx].strip()
else:
# 没有结束 fence，直接用 body 作为候选（可能是截断的 JSON）
candidate = body.strip()
# 只有在确实像 json 的情况下才替换，避免误伤普通文本
if '{' in candidate and '}' in candidate:
text = candidate
logger.debug(f"_extract_json_obj: 提取代码块内容成功，长度={len(text)}")
else:
# 没有换行就按旧逻辑尽量截取
pass
# 直接解析
try:
obj = json.loads(text)
logger.debug("_extract_json_obj: 直接解析成功")
return obj if isinstance(obj, dict) else None
except Exception as e:
logger.debug(f"_extract_json_obj: 直接解析失败: {e}")
# 兜底：截取最外层 { ... } 再解析
start = text.find('{')
end = text.rfind('}')
logger.debug(f"_extract_json_obj: 查找大括号 start={start}, end={end}")
if start != -1 and end != -1 and end > start:
candidate = text[start:end + 1].strip()
logger.debug(f"_extract_json_obj: 候选内容长度={len(candidate)}, 前50字符: {candidate[:50]}")
try:
obj = json.loads(candidate)
logger.debug("_extract_json_obj: 兜底解析成功")
return obj if isinstance(obj, dict) else None
except Exception as e:
logger.debug(f"_extract_json_obj: 兜底解析失败: {e}")
return None
logger.debug("_extract_json_obj: 未能提取到有效的 JSON 对象")
return None
def _call_ollama(self, prompt: str) -> Optional[Dict[str, Any]]:
"""调用 Ollama 本地模型"""
import requests
@@ -267,24 +529,16 @@ class AIReviewer(BaseScanner):
"format": "json"
}
logger.info(f"调用 Ollama: {url}, model={self.model}")
response = requests.post(url, json=payload, timeout=120)
if response.status_code == 200:
result = response.json()
content = result.get('response', '')
# 尝试解析 JSON
try:
# 提取 JSON 部分
if '```json' in content:
content = content.split('```json')[1].split('```')[0]
elif '```' in content:
content = content.split('```')[1].split('```')[0]
return json.loads(content.strip())
except json.JSONDecodeError:
# 如果不是 JSON，直接返回文本
return {'raw_review': content}
logger.info(f"Ollama 返回内容长度: {len(content) if content else 0}")
logger.debug(f"Ollama 返回内容预览: {content[:200] if content else 'empty'}")
parsed = self._extract_json_obj(content)
return parsed
logger.warning(f'Ollama 返回错误: {response.status_code}')
return None
@@ -306,7 +560,7 @@ class AIReviewer(BaseScanner):
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024,
"max_tokens": 1024*5,
"temperature": 0.7
}
elif 'deepseek' in self.api_url:
@@ -314,7 +568,7 @@ class AIReviewer(BaseScanner):
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024,
"max_tokens": 1024*5,
"temperature": 0.7
}
else:
@@ -322,7 +576,7 @@ class AIReviewer(BaseScanner):
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024,
"max_tokens": 1024*5,
"temperature": 0.7
}
@@ -331,34 +585,8 @@ class AIReviewer(BaseScanner):
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
try:
if '```json' in content:
content = content.split('```json')[1].split('```')[0]
elif '```' in content:
content = content.split('```')[1].split('```')[0]
return json.loads(content.strip())
except json.JSONDecodeError:
return {'raw_review': content}
parsed = self._extract_json_obj(content)
return parsed
logger.warning(f'API 返回错误: {response.status_code}')
return None
def _generate_summary(self, reviews: List[Dict[str, Any]]) -> str:
"""生成审查总结"""
if not reviews:
return '未找到需要审查的代码'
total_issues = sum(
len(r.get('review', {}).get('问题', [])) +
len(r.get('review', {}).get('优化', []))
for r in reviews
)
files_count = len(reviews)
if total_issues == 0:
return f'✅ AI 审查通过！审查了 {files_count} 个文件，未发现问题'
return f'🤖 AI 审查了 {files_count} 个文件，发现 {total_issues} 个改进建议'