starmen1265
d19062fc1e
- 创建 SKILL.md 定义9阶段流程 - 定义触发词: '开始开发...' - 创建状态文件结构 state.json - 创建需求收集模板 requirements.json - 创建状态管理工具 state_manager.py - 支持跨会话状态恢复
266 lines
8.7 KiB
Python
266 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
dev-workflow 状态管理工具
|
|
|
|
用于读写和更新 dev-workflow 的状态文件。
|
|
|
|
Usage:
|
|
python3 state_manager.py --init
|
|
python3 state_manager.py --status
|
|
python3 state_manager.py --phase 3
|
|
python3 state_manager.py --complete-phase 2
|
|
python3 state_manager.py --update-phase 3 --data '{"agents": ["A", "B", "C"]}'
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
|
|
STATE_FILE = Path(".planning/dev-workflow/state.json")
|
|
REQUIREMENTS_FILE = Path(".planning/dev-workflow/requirements.json")
|
|
|
|
|
|
def get_state_path() -> Path:
|
|
"""获取状态文件路径"""
|
|
# 从当前目录向上查找项目根目录
|
|
cwd = Path.cwd()
|
|
while cwd != cwd.parent:
|
|
if (cwd / ".planning").exists():
|
|
return cwd / STATE_FILE
|
|
cwd = cwd.parent
|
|
return Path.cwd() / STATE_FILE
|
|
|
|
|
|
def load_state() -> dict:
|
|
"""加载状态文件"""
|
|
state_path = get_state_path()
|
|
if not state_path.exists():
|
|
print(f"❌ 状态文件不存在: {state_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
with open(state_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def save_state(state: dict) -> None:
|
|
"""保存状态文件"""
|
|
state_path = get_state_path()
|
|
state["updated_at"] = datetime.now().isoformat()
|
|
|
|
state_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(state_path, "w", encoding="utf-8") as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"✅ 状态已保存: {state_path}")
|
|
|
|
|
|
def init_state(project_name: str = None, trigger_message: str = None) -> dict:
|
|
"""初始化状态文件"""
|
|
state_path = get_state_path()
|
|
|
|
state = {
|
|
"schema": "dev-workflow-state-v1",
|
|
"project_name": project_name,
|
|
"current_phase": 1,
|
|
"current_stage": "头脑风暴",
|
|
"started_at": datetime.now().isoformat(),
|
|
"updated_at": datetime.now().isoformat(),
|
|
"retry_count": 0,
|
|
"max_retries": 3,
|
|
"phases": {
|
|
"1": {"name": "头脑风暴", "status": "in_progress", "started_at": datetime.now().isoformat()},
|
|
"2": {"name": "方案设计", "status": "pending"},
|
|
"3": {"name": "方案审核", "status": "pending"},
|
|
"4": {"name": "方案汇报", "status": "pending"},
|
|
"5": {"name": "Spec编排", "status": "pending"},
|
|
"6": {"name": "并行开发", "status": "pending"},
|
|
"7": {"name": "测试", "status": "pending"},
|
|
"8": {"name": "代码审查", "status": "pending"},
|
|
"9": {"name": "修复验证", "status": "pending"}
|
|
},
|
|
"context": {
|
|
"trigger_message": trigger_message,
|
|
"working_directory": str(Path.cwd()),
|
|
"project_type": None
|
|
}
|
|
}
|
|
|
|
save_state(state)
|
|
return state
|
|
|
|
|
|
def get_status() -> str:
|
|
"""获取当前状态摘要"""
|
|
state = load_state()
|
|
|
|
phase = state["current_phase"]
|
|
stage = state["current_stage"]
|
|
started = state.get("started_at", "N/A")
|
|
|
|
completed = sum(1 for p in state["phases"].values() if p.get("status") == "completed")
|
|
total = len(state["phases"])
|
|
|
|
status = f"""
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
DEV-WORKFLOW STATUS
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
|
|
Project: {state.get("project_name", "未命名")}
|
|
Phase: {phase}/9 - {stage}
|
|
Progress: {completed}/{total} phases completed
|
|
Started: {started}
|
|
Retries: {state.get("retry_count", 0)}/{state.get("max_retries", 3)}
|
|
|
|
Phases:
|
|
"""
|
|
|
|
for num, phase_data in state["phases"].items():
|
|
status_icon = {
|
|
"completed": "✓",
|
|
"in_progress": "→",
|
|
"pending": "○",
|
|
"failed": "✗"
|
|
}.get(phase_data.get("status", "pending"), "?")
|
|
|
|
status += f" {status_icon} Phase {num}: {phase_data['name']}\n"
|
|
|
|
status += "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
|
|
|
|
return status
|
|
|
|
|
|
def set_phase(phase_num: int) -> None:
|
|
"""设置当前阶段"""
|
|
state = load_state()
|
|
|
|
if str(phase_num) not in state["phases"]:
|
|
print(f"❌ 无效的阶段: {phase_num}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# 标记前一个阶段为完成
|
|
current = state["current_phase"]
|
|
if current > 0 and state["phases"][str(current)].get("status") == "in_progress":
|
|
state["phases"][str(current)]["status"] = "completed"
|
|
state["phases"][str(current)]["completed_at"] = datetime.now().isoformat()
|
|
|
|
# 设置新阶段
|
|
state["current_phase"] = phase_num
|
|
state["current_stage"] = state["phases"][str(phase_num)]["name"]
|
|
state["phases"][str(phase_num)]["status"] = "in_progress"
|
|
state["phases"][str(phase_num)]["started_at"] = datetime.now().isoformat()
|
|
|
|
save_state(state)
|
|
print(f"✅ 已进入阶段 {phase_num}: {state['current_stage']}")
|
|
|
|
|
|
def complete_phase(phase_num: int) -> None:
|
|
"""标记阶段完成"""
|
|
state = load_state()
|
|
|
|
if str(phase_num) not in state["phases"]:
|
|
print(f"❌ 无效的阶段: {phase_num}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
state["phases"][str(phase_num)]["status"] = "completed"
|
|
state["phases"][str(phase_num)]["completed_at"] = datetime.now().isoformat()
|
|
|
|
# 自动进入下一阶段
|
|
next_phase = phase_num + 1
|
|
if str(next_phase) in state["phases"]:
|
|
state["current_phase"] = next_phase
|
|
state["current_stage"] = state["phases"][str(next_phase)]["name"]
|
|
state["phases"][str(next_phase)]["status"] = "pending"
|
|
|
|
save_state(state)
|
|
print(f"✅ 阶段 {phase_num} 已完成")
|
|
|
|
|
|
def update_phase_data(phase_num: int, data: dict) -> None:
|
|
"""更新阶段数据"""
|
|
state = load_state()
|
|
|
|
if str(phase_num) not in state["phases"]:
|
|
print(f"❌ 无效的阶段: {phase_num}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if "data" not in state["phases"][str(phase_num)]:
|
|
state["phases"][str(phase_num)]["data"] = {}
|
|
|
|
state["phases"][str(phase_num)]["data"].update(data)
|
|
|
|
save_state(state)
|
|
print(f"✅ 阶段 {phase_num} 数据已更新")
|
|
|
|
|
|
def increment_retry() -> bool:
|
|
"""增加重试计数，返回是否还可以继续重试"""
|
|
state = load_state()
|
|
state["retry_count"] = state.get("retry_count", 0) + 1
|
|
|
|
max_retries = state.get("max_retries", 3)
|
|
can_retry = state["retry_count"] < max_retries
|
|
|
|
save_state(state)
|
|
|
|
if can_retry:
|
|
print(f"⚠️ 重试 {state['retry_count']}/{max_retries}")
|
|
else:
|
|
print(f"❌ 已达到最大重试次数 ({max_retries})")
|
|
|
|
return can_retry
|
|
|
|
|
|
def reset_retry() -> None:
|
|
"""重置重试计数"""
|
|
state = load_state()
|
|
state["retry_count"] = 0
|
|
save_state(state)
|
|
print("✅ 重试计数已重置")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="dev-workflow 状态管理工具")
|
|
|
|
parser.add_argument("--init", action="store_true", help="初始化状态文件")
|
|
parser.add_argument("--status", action="store_true", help="显示当前状态")
|
|
parser.add_argument("--phase", type=int, metavar="N", help="设置当前阶段")
|
|
parser.add_argument("--complete-phase", type=int, metavar="N", help="标记阶段完成")
|
|
parser.add_argument("--update-phase", type=int, metavar="N", help="更新阶段数据")
|
|
parser.add_argument("--data", type=str, help="JSON 数据 (与 --update-phase 配合)")
|
|
parser.add_argument("--retry", action="store_true", help="增加重试计数")
|
|
parser.add_argument("--reset-retry", action="store_true", help="重置重试计数")
|
|
parser.add_argument("--project-name", type=str, help="项目名称 (与 --init 配合)")
|
|
parser.add_argument("--trigger", type=str, help="触发消息 (与 --init 配合)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.init:
|
|
init_state(args.project_name, args.trigger)
|
|
elif args.status:
|
|
print(get_status())
|
|
elif args.phase:
|
|
set_phase(args.phase)
|
|
elif args.complete_phase:
|
|
complete_phase(args.complete_phase)
|
|
elif args.update_phase:
|
|
if not args.data:
|
|
print("❌ 需要提供 --data 参数", file=sys.stderr)
|
|
sys.exit(1)
|
|
update_phase_data(args.update_phase, json.loads(args.data))
|
|
elif args.retry:
|
|
increment_retry()
|
|
elif args.reset_retry:
|
|
reset_retry()
|
|
else:
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|