开发code-quality

This commit is contained in:
2026-01-30 00:42:33 +08:00
parent f95b5f5b13
commit 3367268ace
14 changed files with 1636 additions and 0 deletions
@@ -0,0 +1,193 @@
#!/usr/bin/env python3
"""
自动完成检查 Hook
拦截任务结束,执行代码质量检查
"""
import json
import sys
import os
from pathlib import Path
# 导入模块
sys.path.insert(0, str(Path(__file__).parent))
from requirements_tracker import generate_requirement_checklist, save_checklist
from code_analyzer import analyze_code_quality, calculate_completeness_score
from notification_sender import notify_check_result
def load_iteration_tracker(project_root):
"""加载迭代计数器"""
tracker_file = Path(project_root) / '.claude' / 'iteration-tracker.json'
if tracker_file.exists():
with open(tracker_file, 'r', encoding='utf-8') as f:
return json.load(f)
return {}
def update_iteration_count(project_root, task_id, increment=True):
"""更新迭代计数"""
tracker = load_iteration_tracker(project_root)
if task_id not in tracker:
tracker[task_id] = 0
if increment:
tracker[task_id] += 1
# 保存
tracker_file = Path(project_root) / '.claude' / 'iteration-tracker.json'
tracker_file.parent.mkdir(parents=True, exist_ok=True)
with open(tracker_file, 'w', encoding='utf-8') as f:
json.dump(tracker, f, ensure_ascii=False, indent=2)
return tracker[task_id]
def reset_iteration_count(project_root, task_id):
"""重置迭代计数(任务完成时)"""
return update_iteration_count(project_root, task_id, increment=False)
def generate_fix_requirements(issues, requirements_checklist):
"""生成修复要求"""
fix_requirements = []
# 从代码问题生成修复要求
if issues['todo']:
fix_requirements.append(f"- 处理 {len(issues['todo'])} 个 TODO/FIXME 项")
if issues['empty_functions']:
fix_requirements.append(f"- 实现以下空函数: {', '.join([f'{i[\"file\"]}:{i[\"line\"]}' for i in issues['empty_functions'][:3]])}")
if issues['placeholders']:
fix_requirements.append(f"- 移除或实现占位符代码 ({len(issues['placeholders'])} 处)")
if issues['commented_code']:
fix_requirements.append(f"- 清理或取消注释代码 ({len(issues['commented_code'])} 处)")
# 从需求清单生成修复要求
unverified = [r for r in requirements_checklist.get('requirements', []) if not r.get('verified', False)]
if unverified:
fix_requirements.append(f"- 验证以下需求已完成:\n " + "\n ".join([f"{i+1}. {r['description']}" for i, r in enumerate(unverified[:5])]))
return "\n".join(fix_requirements) if fix_requirements else "未发现明显问题,建议人工审查"
def perform_quick_check(project_root, task_id, conversation_context=''):
"""执行快速检查"""
# 1. 生成需求检查清单
requirements_checklist = generate_requirement_checklist(project_root, conversation_context)
save_checklist(requirements_checklist, task_id, project_root)
# 2. 代码质量分析
analysis_result = analyze_code_quality(project_root)
score = analysis_result['score']
issues = analysis_result['issues']
# 3. 检查逻辑
passed = score >= 80
critical_issues = len(issues['empty_functions']) + len(issues['placeholders'])
# 4. 获取迭代计数
iteration_count = load_iteration_tracker(project_root).get(task_id, 0)
requires_review = iteration_count >= 3
# 5. 生成结果
result = {
'passed': passed,
'score': score,
'iteration_count': iteration_count,
'requires_review': requires_review,
'issues': analysis_result['summary'],
'total_issues': analysis_result['total_issues']
}
# 6. 决定是否阻止完成
if not passed and not requires_review:
# 增加迭代计数
new_count = update_iteration_count(project_root, task_id)
result['iteration_count'] = new_count
# 生成修复要求
fix_requirements = generate_fix_requirements(issues, requirements_checklist)
result['fix_requirements'] = fix_requirements
result['block_completion'] = True
result['message'] = f'代码质量检查未通过 (评分: {score}/100,迭代: {new_count}/3)'
elif requires_review:
result['block_completion'] = True
result['message'] = f'已达最大迭代次数 ({iteration_count}/3),需要深度审查'
result['trigger_review_agent'] = True
else:
# 通过,重置计数
reset_iteration_count(project_root, task_id)
result['block_completion'] = False
result['message'] = f'代码质量检查通过 (评分: {score}/100)'
return result
def main():
"""主函数"""
try:
# 加载输入
input_data = json.load(sys.stdin)
except json.JSONDecodeError:
input_data = {}
# 获取参数
project_root = input_data.get('projectRoot', os.getcwd())
task_id = input_data.get('taskId', 'default')
conversation_context = input_data.get('conversationContext', '')
# 执行快速检查
check_result = perform_quick_check(project_root, task_id, conversation_context)
# 发送通知(异步,不阻塞)
if check_result.get('passed') or check_result.get('requires_review'):
try:
notify_check_result(check_result)
except Exception:
pass # 通知失败不影响主流程
# 显示终端信息
if not check_result.get('passed'):
print(f"\n{'='*60}", file=sys.stderr)
print(f"⚠️ 代码质量检查未通过", file=sys.stderr)
print(f"{'='*60}", file=sys.stderr)
print(f"评分: {check_result['score']}/100", file=sys.stderr)
print(f"迭代次数: {check_result['iteration_count']}/3", file=sys.stderr)
print(f"\n发现的问题:", file=sys.stderr)
for key, count in check_result['issues'].items():
if count > 0:
print(f" - {key}: {count}", file=sys.stderr)
print(f"\n修复要求:", file=sys.stderr)
print(check_result.get('fix_requirements', ''), file=sys.stderr)
print(f"{'='*60}\n", file=sys.stderr)
# 构造 Hook 输出
output = {
'hookSpecificOutput': {
'hookEventName': 'TaskEnd',
'blockCompletion': check_result.get('block_completion', False),
'additionalContext': check_result.get('fix_requirements', ''),
'triggerReviewAgent': check_result.get('trigger_review_agent', False),
'checkResult': check_result
}
}
# 如果通过,添加成功消息
if check_result.get('passed'):
output['hookSpecificOutput']['message'] = f"✅ 代码质量检查通过 (评分: {check_result['score']}/100)"
# 输出 JSON
print(json.dumps(output, ensure_ascii=False))
if __name__ == '__main__':
main()
+175
View File
@@ -0,0 +1,175 @@
#!/usr/bin/env python3
"""
代码分析器
扫描代码文件,检测常见问题
"""
import json
import sys
import os
import re
from pathlib import Path
def analyze_file(file_path):
"""分析单个代码文件"""
issues = []
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
lines = content.split('\n')
# 检测模式
patterns = {
'TODO': r'TODO|FIXME|XXX|HACK',
'empty_function': r'def\s+\w+\s*\([^)]*\)\s*:\s*(pass|#\s*.*|.{0})',
'empty_method': r'(def|async\s+def)\s+\w+\s*\([^)]*\)\s*:\s*(pass|.{0})',
'commented_code': r'^\s*#.*=[^#\n]|^\s*#.*\([^)]*\)',
'placeholder': r'(NotImplementedError|raise\s+NotImplemented|pass\s*#.*placeholder)',
'debug_print': r'print\([^)]*\)|console\.log\([^)]*\)',
}
for line_num, line in enumerate(lines, 1):
for issue_type, pattern in patterns.items():
if re.search(pattern, line, re.IGNORECASE):
issues.append({
'type': issue_type,
'line': line_num,
'content': line.strip(),
'file': str(file_path)
})
except Exception as e:
issues.append({
'type': 'read_error',
'message': str(e),
'file': str(file_path)
})
return issues
def detect_project_changes(project_root):
"""检测项目中的代码变更"""
import subprocess
try:
# 获取 Git 变更的文件
result = subprocess.run(
['git', 'diff', '--name-only', 'HEAD'],
cwd=project_root,
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
changed_files = result.stdout.strip().split('\n')
changed_files = [f for f in changed_files if f]
# 过滤代码文件
code_extensions = {'.py', '.js', '.ts', '.jsx', '.tsx', '.java', '.go', '.rs', '.c', '.cpp', '.h'}
code_files = [
Path(project_root) / f
for f in changed_files
if Path(f).suffix.lower() in code_extensions
]
return code_files
except Exception:
pass
# 如果 Git 不可用,扫描所有代码文件
code_extensions = {'.py', '.js', '.ts', '.jsx', '.tsx', '.java', '.go', '.rs', '.c', '.cpp', '.h'}
project_path = Path(project_root)
code_files = []
for ext in code_extensions:
code_files.extend(project_path.rglob(f'*{ext}'))
return code_files
def analyze_code_quality(project_root):
"""分析代码质量"""
all_issues = {
'todo': [],
'empty_functions': [],
'commented_code': [],
'placeholders': [],
'debug_statements': [],
'other': []
}
# 获取需要分析的文件
files_to_analyze = detect_project_changes(project_root)
for file_path in files_to_analyze:
if not file_path.exists():
continue
issues = analyze_file(file_path)
for issue in issues:
issue_type = issue['type']
if issue_type == 'TODO':
all_issues['todo'].append(issue)
elif 'empty' in issue_type:
all_issues['empty_functions'].append(issue)
elif issue_type == 'commented_code':
all_issues['commented_code'].append(issue)
elif issue_type == 'placeholder':
all_issues['placeholders'].append(issue)
elif issue_type == 'debug_print':
all_issues['debug_statements'].append(issue)
else:
all_issues['other'].append(issue)
return all_issues
def calculate_completeness_score(issues):
"""计算完整度评分"""
total_issues = sum(len(v) for v in issues.values())
# 严重问题权重
critical_issues = len(issues['empty_functions']) + len(issues['placeholders'])
minor_issues = len(issues['todo']) + len(issues['commented_code']) + len(issues['debug_statements'])
# 计算分数(0-100)
score = 100 - (critical_issues * 15) - (minor_issues * 5)
return max(0, min(100, score))
def main():
"""主函数"""
# 从命令行参数或 stdin 获取项目路径
if len(sys.argv) > 1:
project_root = sys.argv[1]
else:
project_root = os.getcwd()
# 执行分析
issues = analyze_code_quality(project_root)
score = calculate_completeness_score(issues)
# 输出结果
result = {
'score': score,
'issues': issues,
'total_issues': sum(len(v) for v in issues.values()),
'summary': {
'todo_count': len(issues['todo']),
'empty_functions_count': len(issues['empty_functions']),
'commented_code_count': len(issues['commented_code']),
'placeholder_count': len(issues['placeholders']),
'debug_statements_count': len(issues['debug_statements'])
}
}
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == '__main__':
main()
@@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""
通知发送器
发送 Windows 桌面通知和播放声音
"""
import json
import sys
import os
import platform
from pathlib import Path
def send_notification(title, message, icon_path=None, duration=10):
"""发送 Windows 桌面通知"""
try:
from plyer import notification
notification.notify(
title=title,
message=message,
app_name='Claude Code',
app_icon=icon_path,
timeout=duration
)
return True
except ImportError:
# 如果 plyer 不可用,尝试使用 Windows 原生 API
try:
import win10toast
toaster = win10toast.ToastNotifier()
toaster.show_toast(
title,
message,
icon_path=icon_path,
duration=duration
)
return True
except ImportError:
# 最后的回退:输出到 stderr(会显示在终端)
print(f'\n📢 {title}: {message}', file=sys.stderr)
return False
except Exception as e:
print(f'\n📢 {title}: {message}', file=sys.stderr)
return False
def play_sound(sound_type='success', volume=0.7):
"""播放提示音"""
sound_files = {
'success': 'success.wav',
'warning': 'warning.wav',
'attention': 'attention.wav'
}
sound_file = sound_files.get(sound_type, 'success.wav')
# 获取声音文件路径
script_dir = Path(__file__).parent
sounds_dir = script_dir.parent / 'sounds'
sound_path = sounds_dir / sound_file
try:
# 尝试使用 pygame
import pygame
pygame.mixer.init()
pygame.mixer.music.set_volume(volume)
pygame.mixer.music.load(str(sound_path))
pygame.mixer.music.play()
return True
except ImportError:
# 如果 pygame 不可用,尝试系统 beep
try:
import winsound
if sound_type == 'success':
winsound.Beep(800, 200)
elif sound_type == 'warning':
winsound.Beep(400, 300)
else:
winsound.Beep(600, 400)
return True
except Exception:
# 最后的回退:终端提示
return False
except Exception:
return False
def notify_check_result(check_result):
"""根据检查结果发送通知"""
if check_result.get('passed'):
send_notification(
title='✅ 代码质量检查通过',
message='所有功能已实现,代码规范符合要求',
duration=5
)
play_sound('success')
elif check_result.get('requires_review'):
send_notification(
title='🔔 需要深度审查',
message='自动修复已达上限,请查看详细报告',
duration=15
)
play_sound('attention')
else:
issue_count = check_result.get('issue_count', 0)
send_notification(
title='⚠️ 代码质量检查未通过',
message=f'发现 {issue_count} 个问题需要修复',
duration=10
)
play_sound('warning')
def main():
"""主函数 - 可以独立运行或作为模块导入"""
# 从 stdin 或命令行参数读取
if len(sys.argv) > 1:
# 命令行模式
if sys.argv[1] == 'test':
# 测试通知
send_notification('测试通知', '代码质量保障系统正在运行')
play_sound('success')
elif sys.argv[1] == 'success':
send_notification('✅ 检查通过', '所有功能已实现,代码规范符合要求')
play_sound('success')
elif sys.argv[1] == 'warning':
send_notification('⚠️ 检查未通过', '发现问题需要修复')
play_sound('warning')
elif sys.argv[1] == 'attention':
send_notification('🔔 需要审查', '请查看详细报告')
play_sound('attention')
else:
# 从 stdin 读取 JSON
try:
input_data = json.load(sys.stdin)
notify_check_result(input_data)
except json.JSONDecodeError:
pass
if __name__ == '__main__':
main()
@@ -0,0 +1,154 @@
#!/usr/bin/env python3
"""
需求追踪器
从对话历史和计划文件中提取需求,生成检查清单
"""
import json
import sys
import os
import re
from pathlib import Path
def load_input():
"""加载 Hook 输入数据"""
try:
return json.load(sys.stdin)
except json.JSONDecodeError:
return {}
def extract_requirements_from_conversation(conversation_text):
"""从对话中提取需求"""
requirements = []
# 需求关键词模式
patterns = {
'功能需求': r'(需要|实现|添加|创建|开发)\s*([^,。;\n]{5,50})',
'技术栈': r'(使用|基于|框架)\s*([^,。;\n]{2,30})',
'约束条件': r'(必须|应该|不能|禁止)\s*([^,。;\n]{5,50})',
}
for category, pattern in patterns.items():
matches = re.findall(pattern, conversation_text)
for match in matches:
if isinstance(match, tuple):
requirements.append({
'category': category,
'description': match[-1].strip(),
'verified': False
})
return requirements
def read_plan_files(project_root):
"""读取项目计划文件"""
plan_files = []
docs_path = Path(project_root) / 'docs' / 'plans'
if docs_path.exists():
for plan_file in docs_path.glob('*.md'):
try:
with open(plan_file, 'r', encoding='utf-8') as f:
content = f.read()
plan_files.append({
'path': str(plan_file),
'content': content
})
except Exception as e:
pass
return plan_files
def parse_plan_to_checklist(plan_content):
"""解析计划文件为检查清单"""
checklist = []
# 提取任务/功能项
lines = plan_content.split('\n')
for line in lines:
line = line.strip()
# 匹配 markdown 列表项
if line.startswith('- ') or line.startswith('* '):
item = line[2:].strip()
# 移除 [ ] 或 [x] 标记
item = re.sub(r'^\[[ x]\]\s*', '', item)
if item:
checklist.append({
'description': item,
'verified': False,
'source': 'plan_file'
})
return checklist
def generate_requirement_checklist(project_root, conversation_context=''):
"""生成需求检查清单"""
checklist = {
'requirements': [],
'source': 'unknown'
}
# 1. 尝试从计划文件读取
plan_files = read_plan_files(project_root)
if plan_files:
for plan in plan_files:
items = parse_plan_to_checklist(plan['content'])
checklist['requirements'].extend(items)
checklist['source'] = 'plan_files'
checklist['plan_files'] = [p['path'] for p in plan_files]
# 2. 如果没有计划文件,从对话提取
if not checklist['requirements'] and conversation_context:
items = extract_requirements_from_conversation(conversation_context)
checklist['requirements'] = items
checklist['source'] = 'conversation'
return checklist
def save_checklist(checklist, task_id, project_root):
"""保存检查清单到临时文件"""
cache_dir = Path(project_root) / '.claude' / 'cache'
cache_dir.mkdir(parents=True, exist_ok=True)
checklist_file = cache_dir / f'requirements_{task_id}.json'
with open(checklist_file, 'w', encoding='utf-8') as f:
json.dump(checklist, f, ensure_ascii=False, indent=2)
return str(checklist_file)
def main():
"""主函数"""
input_data = load_input()
# 获取项目根目录
project_root = input_data.get('projectRoot', os.getcwd())
# 获取对话上下文(如果有)
conversation_context = input_data.get('conversationContext', '')
# 生成检查清单
checklist = generate_requirement_checklist(project_root, conversation_context)
# 保存到缓存
task_id = input_data.get('taskId', 'default')
checklist_file = save_checklist(checklist, task_id, project_root)
# 输出结果
output = {
'checklist_generated': True,
'requirements_count': len(checklist['requirements']),
'source': checklist['source'],
'checklist_file': checklist_file
}
print(json.dumps(output, ensure_ascii=False))
if __name__ == '__main__':
main()
@@ -0,0 +1,2 @@
plyer>=2.1.0
pygame>=2.5.0