starmen1265
630dc09062
- 更新 SKILL.md：完整的 Cross 模式流程文档 - 更新 dev_workflow.py：新增 record-debate、record-review 命令 - 新增 debate-template.md：Cross-Debate 模板 - 用户介入点：Phase 1 方案选择 + Phase 6 最终审核 - 六条铁律：Briefing 原则、安全约束等 - 禁止 auto-merge：防止流程错误
454 lines
13 KiB
Python
454 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
dev_workflow.py - Agent Engineering 开发流程辅助工具
|
||
|
||
用法:
|
||
python3 dev_workflow.py create-spec --title "标题" --description "描述"
|
||
python3 dev_workflow.py create-branch --spec .planning/spec.md
|
||
python3 dev_workflow.py create-pr --branch feature/xxx --title "PR标题"
|
||
python3 dev_workflow.py record-debate --file .planning/debate.md
|
||
python3 dev_workflow.py record-review --round 1 --issues '[...]'
|
||
python3 dev_workflow.py status
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import argparse
|
||
import subprocess
|
||
import re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
|
||
# 配置
|
||
GITEA_URL = os.environ.get('GITEA_URL', 'http://192.168.1.22:8418')
|
||
GITEA_TOKEN = os.environ.get('GITEA_TOKEN', '')
|
||
GITEA_REPO = os.environ.get('GITEA_REPO', 'starmen1265/kb-skills')
|
||
REPO_PATH = Path.home() / 'work' / 'kb-skills'
|
||
PLANNING_DIR = REPO_PATH / '.planning'
|
||
|
||
# Cross 模式配置
|
||
CROSS_AGENTS = ['claude-code', 'codex', 'gemini']
|
||
DEBATE_FILE = PLANNING_DIR / 'debate.md'
|
||
REVIEW_FILE = PLANNING_DIR / 'review.json'
|
||
|
||
|
||
def run_cmd(cmd, cwd=None, capture=True):
|
||
"""执行 shell 命令"""
|
||
result = subprocess.run(
|
||
cmd,
|
||
shell=True,
|
||
cwd=cwd or REPO_PATH,
|
||
capture_output=capture,
|
||
text=True
|
||
)
|
||
return result
|
||
|
||
|
||
def create_spec(title: str, description: str, from_debate: bool = False) -> dict:
|
||
"""创建 spec 文件"""
|
||
PLANNING_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 生成分支名
|
||
branch_name = re.sub(r'[^a-z0-9]+', '-', title.lower()).strip('-')
|
||
|
||
# 读取模板
|
||
template_path = Path(__file__).parent / 'templates' / 'spec-template.md'
|
||
if template_path.exists():
|
||
content = template_path.read_text(encoding='utf-8')
|
||
else:
|
||
content = """# Spec: {{TITLE}}
|
||
|
||
**创建时间:** {{DATE}}
|
||
**状态:** draft
|
||
**分支:** feature/{{BRANCH_NAME}}
|
||
**来源:** {{SOURCE}}
|
||
|
||
---
|
||
|
||
## 概述
|
||
|
||
{{DESCRIPTION}}
|
||
|
||
## 需求
|
||
|
||
### 功能需求
|
||
|
||
- [ ] **REQ-01**: 待补充
|
||
|
||
## 验收标准
|
||
|
||
1. 待补充
|
||
|
||
---
|
||
|
||
## Cross-Debate 汇总
|
||
|
||
{{DEBATE_SUMMARY}}
|
||
|
||
---
|
||
|
||
## 变更日志
|
||
|
||
| 日期 | 变更 |
|
||
|------|------|
|
||
| {{DATE}} | 初始创建 |
|
||
"""
|
||
|
||
# 替换变量
|
||
now = datetime.now().strftime('%Y-%m-%d %H:%M')
|
||
content = content.replace('{{TITLE}}', title)
|
||
content = content.replace('{{DATE}}', now)
|
||
content = content.replace('{{BRANCH_NAME}}', branch_name)
|
||
content = content.replace('{{DESCRIPTION}}', description)
|
||
content = content.replace('{{SOURCE}}', 'Cross-Debate' if from_debate else '用户需求')
|
||
|
||
# 如果有 debate 文件，提取汇总
|
||
if from_debate and DEBATE_FILE.exists():
|
||
debate_content = DEBATE_FILE.read_text(encoding='utf-8')
|
||
# 提取共识点部分
|
||
match = re.search(r'## 共识点\n(.*?)(?=\n## |$)', debate_content, re.DOTALL)
|
||
if match:
|
||
content = content.replace('{{DEBATE_SUMMARY}}', match.group(1).strip())
|
||
else:
|
||
content = content.replace('{{DEBATE_SUMMARY}}', '见 .planning/debate.md')
|
||
else:
|
||
content = content.replace('{{DEBATE_SUMMARY}}', '无')
|
||
|
||
# 写入文件
|
||
spec_file = PLANNING_DIR / 'spec.md'
|
||
spec_file.write_text(content, encoding='utf-8')
|
||
|
||
return {
|
||
"success": True,
|
||
"spec_file": str(spec_file),
|
||
"branch_name": branch_name,
|
||
"title": title,
|
||
"from_debate": from_debate
|
||
}
|
||
|
||
|
||
def create_branch(spec_file: str) -> dict:
|
||
"""从 spec 创建功能分支"""
|
||
spec_path = Path(spec_file)
|
||
if not spec_path.exists():
|
||
return {"success": False, "error": f"Spec 文件不存在: {spec_file}"}
|
||
|
||
# 从 spec 提取分支名
|
||
content = spec_path.read_text(encoding='utf-8')
|
||
match = re.search(r'\*\*分支:\*\*\s*feature/([^\n]+)', content)
|
||
if match:
|
||
branch_suffix = match.group(1).strip()
|
||
else:
|
||
branch_suffix = f"feature-{datetime.now().strftime('%Y%m%d%H%M%S')}"
|
||
|
||
branch_name = f"feature/{branch_suffix}"
|
||
|
||
# 确保在 main 分支
|
||
run_cmd('git checkout main')
|
||
run_cmd('git pull origin main')
|
||
|
||
# 创建功能分支
|
||
result = run_cmd(f'git checkout -b {branch_name}')
|
||
if result.returncode != 0:
|
||
# 分支可能已存在，尝试切换
|
||
result = run_cmd(f'git checkout {branch_name}')
|
||
if result.returncode != 0:
|
||
return {"success": False, "error": f"无法创建分支: {result.stderr}"}
|
||
|
||
return {
|
||
"success": True,
|
||
"branch": branch_name,
|
||
"spec_file": spec_file
|
||
}
|
||
|
||
|
||
def create_pr(branch: str, title: str, body: str = "") -> dict:
|
||
"""调用 Gitea API 创建 PR（禁止 auto-merge）"""
|
||
if not GITEA_TOKEN:
|
||
return {"success": False, "error": "GITEA_TOKEN 未配置"}
|
||
|
||
# 获取最新提交信息作为 body（如果未提供）
|
||
if not body:
|
||
result = run_cmd('git log -1 --pretty=%B')
|
||
body = result.stdout.strip() if result.returncode == 0 else ""
|
||
|
||
# 添加 Cross 模式说明
|
||
review_info = ""
|
||
if REVIEW_FILE.exists():
|
||
try:
|
||
review_data = json.loads(REVIEW_FILE.read_text())
|
||
total_issues = sum(len(v) for v in review_data.get('rounds', {}).values())
|
||
rounds = len(review_data.get('rounds', {}))
|
||
review_info = f"\n\n**Cross-Review:** {rounds} 轮审核，修复 {total_issues} 个问题"
|
||
except:
|
||
pass
|
||
|
||
body += review_info
|
||
body += "\n\n⚠️ **禁止 auto-merge** - 等待人工最终审核"
|
||
|
||
# 调用 Gitea API
|
||
import urllib.request
|
||
import urllib.parse
|
||
import json as json_module
|
||
|
||
url = f"{GITEA_URL}/api/v1/repos/{GITEA_REPO}/pulls"
|
||
data = {
|
||
"title": title,
|
||
"head": branch,
|
||
"base": "main",
|
||
"body": body
|
||
}
|
||
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=json_module.dumps(data).encode('utf-8'),
|
||
headers={
|
||
'Authorization': f'token {GITEA_TOKEN}',
|
||
'Content-Type': 'application/json'
|
||
},
|
||
method='POST'
|
||
)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as response:
|
||
result = json_module.loads(response.read().decode('utf-8'))
|
||
return {
|
||
"success": True,
|
||
"pr_number": result.get("number"),
|
||
"pr_url": result.get("html_url"),
|
||
"branch": branch,
|
||
"warning": "禁止设置 auto-merge"
|
||
}
|
||
except urllib.error.HTTPError as e:
|
||
error_body = e.read().decode('utf-8') if e.fp else str(e)
|
||
return {"success": False, "error": f"API 错误: {e.code} - {error_body}"}
|
||
except Exception as e:
|
||
return {"success": False, "error": f"请求失败: {str(e)}"}
|
||
|
||
|
||
def record_debate(file: str = None) -> dict:
|
||
"""记录 Cross-Debate 结果"""
|
||
PLANNING_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
debate_content = None
|
||
if file:
|
||
file_path = Path(file)
|
||
if file_path.exists():
|
||
debate_content = file_path.read_text(encoding='utf-8')
|
||
|
||
if debate_content:
|
||
DEBATE_FILE.write_text(debate_content, encoding='utf-8')
|
||
return {
|
||
"success": True,
|
||
"file": str(DEBATE_FILE),
|
||
"message": "Cross-Debate 结果已记录"
|
||
}
|
||
else:
|
||
# 创建模板
|
||
template = f"""# Cross-Debate 结果
|
||
|
||
**创建时间:** {datetime.now().strftime('%Y-%m-%d %H:%M')}
|
||
**需求:** <!-- 填写需求 -->
|
||
|
||
---
|
||
|
||
## Agent 方案
|
||
|
||
### Claude Code（保守派）
|
||
|
||
**方案:**
|
||
<!-- 填写 -->
|
||
|
||
**理由:**
|
||
<!-- 填写 -->
|
||
|
||
---
|
||
|
||
### Codex（全面派）
|
||
|
||
**方案:**
|
||
<!-- 填写 -->
|
||
|
||
**理由:**
|
||
<!-- 填写 -->
|
||
|
||
---
|
||
|
||
### Gemini（快速派）
|
||
|
||
**方案:**
|
||
<!-- 填写 -->
|
||
|
||
**理由:**
|
||
<!-- 填写 -->
|
||
|
||
---
|
||
|
||
## 共识点
|
||
|
||
- ✅ <!-- 共识1 -->
|
||
- ✅ <!-- 共识2 -->
|
||
|
||
## 分歧点
|
||
|
||
| Agent | 方案 | 理由 |
|
||
|-------|------|------|
|
||
| <!-- Agent --> | <!-- 方案 --> | <!-- 理由 --> |
|
||
|
||
## 用户决策
|
||
|
||
**选择方案:** <!-- A/B/C 或调整 -->
|
||
|
||
---
|
||
"""
|
||
DEBATE_FILE.write_text(template, encoding='utf-8')
|
||
return {
|
||
"success": True,
|
||
"file": str(DEBATE_FILE),
|
||
"message": "Cross-Debate 模板已创建，请填写"
|
||
}
|
||
|
||
|
||
def record_review(round_num: int, issues: str = None) -> dict:
|
||
"""记录 Cross-Review 结果"""
|
||
PLANNING_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 读取现有数据
|
||
review_data = {"rounds": {}}
|
||
if REVIEW_FILE.exists():
|
||
try:
|
||
review_data = json.loads(REVIEW_FILE.read_text())
|
||
except:
|
||
pass
|
||
|
||
# 解析 issues
|
||
issues_list = []
|
||
if issues:
|
||
try:
|
||
issues_list = json.loads(issues)
|
||
except:
|
||
# 简单格式：每行一个问题
|
||
issues_list = [{"level": "medium", "desc": line} for line in issues.split('\n') if line.strip()]
|
||
|
||
# 添加本轮结果
|
||
round_key = f"round_{round_num}"
|
||
review_data["rounds"][round_key] = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"issues": issues_list
|
||
}
|
||
|
||
# 统计
|
||
review_data["stats"] = {
|
||
"total_rounds": len(review_data["rounds"]),
|
||
"total_issues": sum(len(v.get("issues", [])) for v in review_data["rounds"].values())
|
||
}
|
||
|
||
REVIEW_FILE.write_text(json.dumps(review_data, ensure_ascii=False, indent=2), encoding='utf-8')
|
||
|
||
return {
|
||
"success": True,
|
||
"file": str(REVIEW_FILE),
|
||
"round": round_num,
|
||
"issues_count": len(issues_list),
|
||
"stats": review_data["stats"]
|
||
}
|
||
|
||
|
||
def get_status() -> dict:
|
||
"""获取当前开发状态"""
|
||
result = {
|
||
"current_branch": None,
|
||
"spec_exists": False,
|
||
"debate_exists": False,
|
||
"review_exists": False,
|
||
"planning_exists": False,
|
||
"uncommitted_changes": False,
|
||
"cross_mode": False
|
||
}
|
||
|
||
# 当前分支
|
||
branch_result = run_cmd('git branch --show-current')
|
||
if branch_result.returncode == 0:
|
||
result["current_branch"] = branch_result.stdout.strip()
|
||
|
||
# 各种文件状态
|
||
result["spec_exists"] = (PLANNING_DIR / 'spec.md').exists()
|
||
result["debate_exists"] = DEBATE_FILE.exists()
|
||
result["review_exists"] = REVIEW_FILE.exists()
|
||
result["planning_exists"] = PLANNING_DIR.exists()
|
||
result["cross_mode"] = result["debate_exists"] or result["review_exists"]
|
||
|
||
# 未提交的更改
|
||
status_result = run_cmd('git status --porcelain')
|
||
result["uncommitted_changes"] = bool(status_result.stdout.strip())
|
||
|
||
# 如果有 review 数据，显示统计
|
||
if REVIEW_FILE.exists():
|
||
try:
|
||
review_data = json.loads(REVIEW_FILE.read_text())
|
||
result["review_stats"] = review_data.get("stats", {})
|
||
except:
|
||
pass
|
||
|
||
return result
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='dev-workflow 辅助工具（Cross 模式）')
|
||
subparsers = parser.add_subparsers(dest='command', help='命令')
|
||
|
||
# create-spec
|
||
spec_parser = subparsers.add_parser('create-spec', help='创建 spec 文件')
|
||
spec_parser.add_argument('--title', required=True, help='功能标题')
|
||
spec_parser.add_argument('--description', default='', help='功能描述')
|
||
spec_parser.add_argument('--from-debate', action='store_true', help='从 Cross-Debate 结果生成')
|
||
|
||
# create-branch
|
||
branch_parser = subparsers.add_parser('create-branch', help='创建功能分支')
|
||
branch_parser.add_argument('--spec', required=True, help='spec 文件路径')
|
||
|
||
# create-pr
|
||
pr_parser = subparsers.add_parser('create-pr', help='创建 PR（禁止 auto-merge）')
|
||
pr_parser.add_argument('--branch', required=True, help='分支名')
|
||
pr_parser.add_argument('--title', required=True, help='PR 标题')
|
||
pr_parser.add_argument('--body', default='', help='PR 描述')
|
||
|
||
# record-debate
|
||
debate_parser = subparsers.add_parser('record-debate', help='记录 Cross-Debate 结果')
|
||
debate_parser.add_argument('--file', help='输入文件路径（可选）')
|
||
|
||
# record-review
|
||
review_parser = subparsers.add_parser('record-review', help='记录 Cross-Review 结果')
|
||
review_parser.add_argument('--round', type=int, required=True, help='审核轮次')
|
||
review_parser.add_argument('--issues', help='问题列表（JSON 或简单文本）')
|
||
|
||
# status
|
||
subparsers.add_parser('status', help='查看状态')
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.command == 'create-spec':
|
||
result = create_spec(args.title, args.description, args.from_debate)
|
||
elif args.command == 'create-branch':
|
||
result = create_branch(args.spec)
|
||
elif args.command == 'create-pr':
|
||
result = create_pr(args.branch, args.title, args.body)
|
||
elif args.command == 'record-debate':
|
||
result = record_debate(args.file)
|
||
elif args.command == 'record-review':
|
||
result = record_review(args.round, args.issues)
|
||
elif args.command == 'status':
|
||
result = get_status()
|
||
else:
|
||
parser.print_help()
|
||
return
|
||
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|