572 lines
24 KiB
Python
572 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
项目初始化与部署工作流管理器
|
|
提供状态管理、步骤执行和错误恢复功能
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple
|
|
from dataclasses import dataclass, asdict
|
|
from enum import Enum
|
|
|
|
class StepStatus(Enum):
|
|
PENDING = "⏭️ 未开始"
|
|
IN_PROGRESS = "⏸️ 进行中"
|
|
COMPLETED = "✅ 已完成"
|
|
FAILED = "❌ 失败"
|
|
SKIPPED = "⏭️ 跳过"
|
|
|
|
class ProjectType(Enum):
|
|
DOCKER_WEB = "docker_web"
|
|
DOCKER_API = "docker_api"
|
|
DOCKER_MICROSERVICES = "docker_microservices"
|
|
DESKTOP = "desktop"
|
|
MOBILE = "mobile"
|
|
CLI = "cli"
|
|
UNKNOWN = "unknown"
|
|
|
|
class TechStack(Enum):
|
|
JAVA_SPRING = "java_spring"
|
|
NODE_EXPRESS = "node_express"
|
|
NODE_NEST = "node_nest"
|
|
PYTHON_DJANGO = "python_django"
|
|
PYTHON_FASTAPI = "python_fastapi"
|
|
GO_GIN = "go_gin"
|
|
GO_FIBER = "go_fiber"
|
|
REACT = "react"
|
|
VUE = "vue"
|
|
ANGULAR = "angular"
|
|
UNKNOWN = "unknown"
|
|
|
|
@dataclass
|
|
class ProjectInfo:
|
|
"""项目信息"""
|
|
name: str
|
|
directory: str
|
|
git_url: str
|
|
port_range: Optional[str] = None
|
|
project_type: ProjectType = ProjectType.UNKNOWN
|
|
tech_stack: List[TechStack] = None
|
|
|
|
def __post_init__(self):
|
|
if self.tech_stack is None:
|
|
self.tech_stack = []
|
|
|
|
@dataclass
|
|
class Step:
|
|
"""工作流步骤"""
|
|
id: int
|
|
name: str
|
|
description: str
|
|
applicable_types: List[ProjectType]
|
|
status: StepStatus = StepStatus.PENDING
|
|
start_time: Optional[str] = None
|
|
end_time: Optional[str] = None
|
|
notes: str = ""
|
|
error_info: Optional[str] = None
|
|
|
|
class WorkflowManager:
|
|
"""工作流管理器"""
|
|
|
|
# 完整的21个核心步骤
|
|
ALL_TYPES = [ProjectType.DOCKER_WEB, ProjectType.DOCKER_API, ProjectType.DOCKER_MICROSERVICES, ProjectType.DESKTOP, ProjectType.MOBILE, ProjectType.CLI]
|
|
DOCKER_TYPES = [ProjectType.DOCKER_WEB, ProjectType.DOCKER_API, ProjectType.DOCKER_MICROSERVICES]
|
|
|
|
# 关键时序：初始基线启动(3，内部) → 初始界面截图(4，内部) → 面板改造(6-11) → 产品构建启动(15)
|
|
# → 健康验证(16) → 产品截图与录屏(17) → README/归档(18) → 交付核查(19) → 交付树双存档(20，授权推送时)
|
|
CORE_STEPS = [
|
|
Step(0, "克隆项目", "从Git仓库克隆项目到本地，不修改任何业务代码", ALL_TYPES),
|
|
Step(1, "项目完整度评估", "评估项目完整度、依赖情况与可运行性", ALL_TYPES),
|
|
Step(2, "项目整体分析", "分析项目结构、技术栈、启动方式和部署策略", ALL_TYPES),
|
|
Step(3, "初始版本基线启动（内部）", "在改动任何代码前，用项目现有启动方式跑通初始版本并确认可访问；仅用于内部质量核对", ALL_TYPES),
|
|
Step(4, "初始界面基线截图（内部）", "代码零改动状态下采集初始界面作为内部对比基线；严禁进入交付包或使用 before 等对外文件名", ALL_TYPES),
|
|
Step(5, "Docker配置生成", "生成/完善 Docker Compose 配置文件", DOCKER_TYPES),
|
|
Step(6, "页面信息与品牌名称修改", "替换页面品牌信息并建立科技感面板基础样式", ALL_TYPES),
|
|
Step(7, "Logo生成与替换", "生成并替换项目 Logo、Favicon，适配深色面板", ALL_TYPES),
|
|
Step(8, "版权与第三方标识规范", "清理依法可移除的公司名称、LOGO、商标、水印、联系方式等专属标识；MIT 等许可证要求保留的版权与许可文本必须随源码保留", ALL_TYPES),
|
|
Step(9, "页脚、导航与社区链接", "导航、页脚改为半透明悬浮面板风格，并放置开源社区与 GCC AI Coding 链接", ALL_TYPES),
|
|
Step(10, "登录页面", "改成居中玻璃面板、发光输入框、动态背景", ALL_TYPES),
|
|
Step(11, "AI对话功能", "实现前端AI对话界面与指定服务商API对接及使用教程", ALL_TYPES),
|
|
Step(12, "示例数据", "生成和导入示例数据，数据卡片/列表面板承载", DOCKER_TYPES),
|
|
Step(13, "CORS配置检查", "检查并配置 CORS 设置", DOCKER_TYPES),
|
|
Step(14, "端口检查与分配", "检查端口占用并分配端口", DOCKER_TYPES),
|
|
Step(15, "改造后构建与启动", "Docker项目构建镜像启动容器；非Docker项目手动构建并启动改造版", ALL_TYPES),
|
|
Step(16, "健康验证", "验证服务健康、面板UI审计0违规、全功能走查无报错", ALL_TYPES),
|
|
Step(17, "产品截图与全功能录屏", "采集当前产品全套高清界面截图（中性文件名）和全功能演示录屏；内部基线素材不进入交付物", ALL_TYPES),
|
|
Step(18, "标准化README与版本归档", "编写标准化 README（引用最终素材、含架构图与部署教程）；不主动写来源/致谢宣传，并按授权决定是否提交", ALL_TYPES),
|
|
Step(19, "交付物核查", "按7项交付物清单与7条验收标准做交付前阻断式核查", ALL_TYPES),
|
|
Step(20, "上传交付树双存档", "经用户授权后克隆交付树仓库，同时上传交付zip与同名解压目录（README/截图/录屏/文档），放行mp4、更新registry/sidecar/交付登记/README，推送并做本地远程HEAD比对与Gitea API双重验证", ALL_TYPES),
|
|
]
|
|
|
|
# 旧版本步骤名 → 新版本步骤名（用于历史状态迁移）
|
|
STEP_NAME_ALIASES = {
|
|
"构建与启动": "改造后构建与启动",
|
|
"素材录制归档": "产品截图与全功能录屏",
|
|
"原始项目基线启动": "初始版本基线启动（内部）",
|
|
"改造前基线截图": "初始界面基线截图（内部）",
|
|
"原作者信息与版权痕迹合规处理": "版权与第三方标识规范",
|
|
"页脚与导航增强": "页脚、导航与社区链接",
|
|
"改造后截图与全功能录屏": "产品截图与全功能录屏",
|
|
"标准化README与提交": "标准化README与版本归档",
|
|
}
|
|
|
|
def __init__(self, project_dir: str):
|
|
self.project_dir = Path(project_dir)
|
|
self.claude_dir = self.project_dir / ".Codex"
|
|
self.state_file = self.claude_dir / "workflow_state.json"
|
|
self.error_file = self.claude_dir / "error_log.md"
|
|
self.project_info: Optional[ProjectInfo] = None
|
|
self.steps: List[Step] = []
|
|
|
|
def init_project(self, name: str, git_url: str, port_range: Optional[str] = None):
|
|
"""初始化项目信息"""
|
|
self.project_info = ProjectInfo(
|
|
name=name,
|
|
directory=str(self.project_dir),
|
|
git_url=git_url,
|
|
port_range=port_range
|
|
)
|
|
self.steps = [Step(s.id, s.name, s.description, s.applicable_types) for s in self.CORE_STEPS]
|
|
self._ensure_directories()
|
|
self._save_state()
|
|
|
|
def _ensure_directories(self):
|
|
"""确保必要的目录存在"""
|
|
self.claude_dir.mkdir(exist_ok=True)
|
|
(self.claude_dir / 'reports').mkdir(exist_ok=True)
|
|
|
|
def _save_state(self):
|
|
"""保存状态到JSON文件"""
|
|
def enum_to_value(obj):
|
|
if isinstance(obj, Enum):
|
|
return obj.value
|
|
if isinstance(obj, list):
|
|
return [enum_to_value(item) for item in obj]
|
|
if isinstance(obj, dict):
|
|
return {k: enum_to_value(v) for k, v in obj.items()}
|
|
return obj
|
|
|
|
project_info_dict = None
|
|
if self.project_info:
|
|
project_info_dict = asdict(self.project_info)
|
|
project_info_dict["project_type"] = project_info_dict["project_type"].value
|
|
project_info_dict["tech_stack"] = [ts.value for ts in project_info_dict["tech_stack"]]
|
|
|
|
state = {
|
|
"project_info": project_info_dict,
|
|
"steps": [
|
|
{
|
|
"id": s.id,
|
|
"name": s.name,
|
|
"status": s.status.value,
|
|
"start_time": s.start_time,
|
|
"end_time": s.end_time,
|
|
"notes": s.notes,
|
|
"error_info": s.error_info
|
|
}
|
|
for s in self.steps
|
|
],
|
|
"last_updated": datetime.now().isoformat()
|
|
}
|
|
with open(self.state_file, 'w', encoding='utf-8') as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
|
|
def load_state(self) -> bool:
|
|
"""从JSON文件加载状态"""
|
|
if not self.state_file.exists():
|
|
return False
|
|
|
|
try:
|
|
with open(self.state_file, 'r', encoding='utf-8') as f:
|
|
state = json.load(f)
|
|
|
|
if state.get("project_info"):
|
|
info = state["project_info"]
|
|
self.project_info = ProjectInfo(
|
|
name=info["name"],
|
|
directory=info["directory"],
|
|
git_url=info["git_url"],
|
|
port_range=info.get("port_range"),
|
|
project_type=ProjectType(info.get("project_type", "unknown")),
|
|
tech_stack=[TechStack(ts) for ts in info.get("tech_stack", [])]
|
|
)
|
|
|
|
# 以新 CORE_STEPS 为权威骨架；旧状态按“步骤名（含别名）”映射进度，
|
|
# 避免编号错位导致已完成的步骤被重置。
|
|
old_steps = state.get("steps", [])
|
|
name_to_state: Dict[str, dict] = {}
|
|
for s in old_steps:
|
|
canon = self.STEP_NAME_ALIASES.get(s.get("name", ""), s.get("name", ""))
|
|
name_to_state[canon] = s
|
|
|
|
self.steps = []
|
|
for cs in self.CORE_STEPS:
|
|
prev = name_to_state.get(cs.name)
|
|
if prev is not None:
|
|
self.steps.append(Step(
|
|
id=cs.id,
|
|
name=cs.name,
|
|
description=cs.description,
|
|
applicable_types=cs.applicable_types,
|
|
status=StepStatus(prev.get("status", StepStatus.PENDING.value)),
|
|
start_time=prev.get("start_time"),
|
|
end_time=prev.get("end_time"),
|
|
notes=prev.get("notes", ""),
|
|
error_info=prev.get("error_info"),
|
|
))
|
|
else:
|
|
# 新增步骤默认未开始
|
|
self.steps.append(Step(cs.id, cs.name, cs.description, cs.applicable_types))
|
|
|
|
return True
|
|
except Exception as e:
|
|
self._log_error(f"加载状态失败: {e}")
|
|
return False
|
|
|
|
def get_current_step(self) -> Optional[Step]:
|
|
"""获取当前步骤"""
|
|
for step in self.steps:
|
|
if step.status == StepStatus.IN_PROGRESS:
|
|
return step
|
|
if step.status == StepStatus.PENDING:
|
|
return step
|
|
return None
|
|
|
|
def start_step(self, step_id: int) -> bool:
|
|
"""开始执行步骤"""
|
|
step = next((s for s in self.steps if s.id == step_id), None)
|
|
if not step:
|
|
return False
|
|
|
|
# 检查步骤是否适用于当前项目类型
|
|
if self.project_info and self.project_info.project_type != ProjectType.UNKNOWN:
|
|
if self.project_info.project_type not in step.applicable_types:
|
|
step.status = StepStatus.SKIPPED
|
|
step.notes = f"步骤不适用于项目类型: {self.project_info.project_type.value}"
|
|
self._save_state()
|
|
return True
|
|
|
|
step.status = StepStatus.IN_PROGRESS
|
|
step.start_time = datetime.now().isoformat()
|
|
self._save_state()
|
|
return True
|
|
|
|
def complete_step(self, step_id: int, notes: str = ""):
|
|
"""完成步骤"""
|
|
step = next((s for s in self.steps if s.id == step_id), None)
|
|
if step:
|
|
step.status = StepStatus.COMPLETED
|
|
step.end_time = datetime.now().isoformat()
|
|
step.notes = notes
|
|
self._save_state()
|
|
|
|
def fail_step(self, step_id: int, error_info: str):
|
|
"""标记步骤失败"""
|
|
step = next((s for s in self.steps if s.id == step_id), None)
|
|
if step:
|
|
step.status = StepStatus.FAILED
|
|
step.end_time = datetime.now().isoformat()
|
|
step.error_info = error_info
|
|
self._save_state()
|
|
self._log_error(f"步骤 {step_id} ({step.name}) 失败: {error_info}")
|
|
|
|
def skip_step(self, step_id: int, reason: str = ""):
|
|
"""跳过步骤"""
|
|
step = next((s for s in self.steps if s.id == step_id), None)
|
|
if step:
|
|
step.status = StepStatus.SKIPPED
|
|
step.notes = reason
|
|
self._save_state()
|
|
|
|
def _log_error(self, message: str):
|
|
"""记录错误到错误日志"""
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
error_entry = f"\n## [{timestamp}]\n{message}\n\n"
|
|
|
|
with open(self.error_file, 'a', encoding='utf-8') as f:
|
|
f.write(error_entry)
|
|
|
|
def get_progress(self) -> Dict:
|
|
"""获取进度信息"""
|
|
total = len(self.steps)
|
|
completed = sum(1 for s in self.steps if s.status == StepStatus.COMPLETED)
|
|
failed = sum(1 for s in self.steps if s.status == StepStatus.FAILED)
|
|
skipped = sum(1 for s in self.steps if s.status == StepStatus.SKIPPED)
|
|
pending = sum(1 for s in self.steps if s.status == StepStatus.PENDING)
|
|
|
|
return {
|
|
"total": total,
|
|
"completed": completed,
|
|
"failed": failed,
|
|
"skipped": skipped,
|
|
"pending": pending,
|
|
"percentage": (completed / total * 100) if total > 0 else 0
|
|
}
|
|
|
|
def can_resume(self) -> Tuple[bool, Optional[int]]:
|
|
"""检查是否可以恢复执行"""
|
|
if not self.state_file.exists():
|
|
return False, None
|
|
|
|
self.load_state()
|
|
|
|
# 检查是否有失败的步骤
|
|
failed_step = next((s for s in self.steps if s.status == StepStatus.FAILED), None)
|
|
if failed_step:
|
|
return True, failed_step.id
|
|
|
|
# 检查是否有进行中的步骤
|
|
in_progress = next((s for s in self.steps if s.status == StepStatus.IN_PROGRESS), None)
|
|
if in_progress:
|
|
return True, in_progress.id
|
|
|
|
# 检查是否有未开始的步骤
|
|
pending = next((s for s in self.steps if s.status == StepStatus.PENDING), None)
|
|
if pending:
|
|
return True, pending.id
|
|
|
|
return False, None
|
|
|
|
def generate_report(self) -> str:
|
|
"""生成执行报告"""
|
|
progress = self.get_progress()
|
|
|
|
report = f"""# 工作流执行报告
|
|
|
|
## 项目信息
|
|
- **项目名称**: {self.project_info.name if self.project_info else 'N/A'}
|
|
- **项目目录**: {self.project_info.directory if self.project_info else 'N/A'}
|
|
- **Git地址**: {self.project_info.git_url if self.project_info else 'N/A'}
|
|
- **项目类型**: {self.project_info.project_type.value if self.project_info else 'N/A'}
|
|
|
|
## 执行进度
|
|
- **总步骤**: {progress['total']}
|
|
- **已完成**: {progress['completed']}
|
|
- **失败**: {progress['failed']}
|
|
- **跳过**: {progress['skipped']}
|
|
- **待执行**: {progress['pending']}
|
|
- **完成度**: {progress['percentage']:.1f}%
|
|
|
|
## 步骤详情
|
|
"""
|
|
for step in self.steps:
|
|
report += f"\n### 步骤 {step.id}: {step.name}\n"
|
|
report += f"- **状态**: {step.status.value}\n"
|
|
if step.start_time:
|
|
report += f"- **开始时间**: {step.start_time}\n"
|
|
if step.end_time:
|
|
report += f"- **完成时间**: {step.end_time}\n"
|
|
if step.notes:
|
|
report += f"- **说明**: {step.notes}\n"
|
|
if step.error_info:
|
|
report += f"- **错误**: {step.error_info}\n"
|
|
|
|
return report
|
|
|
|
# 技术栈检测工具
|
|
class TechStackDetector:
|
|
"""技术栈检测器"""
|
|
|
|
DETECTION_RULES = {
|
|
TechStack.JAVA_SPRING: ["pom.xml", "build.gradle", "src/main/java"],
|
|
TechStack.NODE_EXPRESS: ["package.json", "app.js", "server.js"],
|
|
TechStack.NODE_NEST: ["nest-cli.json", "@nestjs"],
|
|
TechStack.PYTHON_DJANGO: ["manage.py", "django", "requirements.txt"],
|
|
TechStack.PYTHON_FASTAPI: ["fastapi", "main.py"],
|
|
TechStack.GO_GIN: ["go.mod", "gin"],
|
|
TechStack.GO_FIBER: ["go.mod", "fiber"],
|
|
TechStack.REACT: ["package.json", "react", "src/App.js", "src/App.tsx", "vite.config.ts", "vite.config.js"],
|
|
TechStack.VUE: ["package.json", "vue", "vue.config.js", "vite.config.ts", "vite.config.js"],
|
|
TechStack.ANGULAR: ["angular.json", "@angular"],
|
|
}
|
|
|
|
@classmethod
|
|
def detect(cls, project_dir: str) -> List[TechStack]:
|
|
"""检测项目技术栈"""
|
|
detected = []
|
|
project_path = Path(project_dir)
|
|
|
|
for tech, indicators in cls.DETECTION_RULES.items():
|
|
found = False
|
|
for indicator in indicators:
|
|
# 递归检查文件存在
|
|
for _ in project_path.rglob(indicator):
|
|
found = True
|
|
break
|
|
if found: break
|
|
|
|
# 递归检查文件内容
|
|
for file in project_path.rglob("*.json"):
|
|
try:
|
|
with open(file, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
if indicator in content:
|
|
found = True
|
|
break
|
|
except:
|
|
pass
|
|
if found: break
|
|
|
|
if found and tech not in detected:
|
|
detected.append(tech)
|
|
|
|
return detected if detected else [TechStack.UNKNOWN]
|
|
|
|
# 项目类型判断器
|
|
class ProjectTypeDetector:
|
|
"""项目类型检测器"""
|
|
|
|
@classmethod
|
|
def detect(cls, project_dir: str, tech_stacks: List[TechStack]) -> ProjectType:
|
|
"""检测项目类型"""
|
|
project_path = Path(project_dir)
|
|
|
|
# 检查桌面应用特征
|
|
desktop_indicators = ["electron", "qt", "swing", "wpf", ".desktop"]
|
|
for indicator in desktop_indicators:
|
|
if any(indicator in str(tech.value).lower() for tech in tech_stacks):
|
|
return ProjectType.DESKTOP
|
|
|
|
# 检查移动应用特征
|
|
mobile_indicators = ["react-native", "flutter", "android", "ios"]
|
|
for indicator in mobile_indicators:
|
|
if any(indicator in str(tech.value).lower() for tech in tech_stacks):
|
|
return ProjectType.MOBILE
|
|
|
|
# 检查Docker配置
|
|
if (project_path / "Dockerfile").exists() or (project_path / "docker-compose.yml").exists():
|
|
# 检查微服务特征
|
|
if (project_path / "services").exists() or len(list(project_path.glob("**/Dockerfile"))) > 1:
|
|
return ProjectType.DOCKER_MICROSERVICES
|
|
return ProjectType.DOCKER_WEB
|
|
|
|
# 检查API服务特征
|
|
if any(tech in [TechStack.NODE_EXPRESS, TechStack.NODE_NEST,
|
|
TechStack.PYTHON_FASTAPI, TechStack.GO_GIN] for tech in tech_stacks):
|
|
return ProjectType.DOCKER_API
|
|
|
|
return ProjectType.UNKNOWN
|
|
|
|
def main():
|
|
"""命令行入口"""
|
|
if len(sys.argv) < 2:
|
|
print("Usage: workflow_manager.py <command> [args...]")
|
|
print("Commands: init, status, resume, report")
|
|
sys.exit(1)
|
|
|
|
command = sys.argv[1]
|
|
|
|
if command == "init":
|
|
if len(sys.argv) < 5:
|
|
print("Usage: workflow_manager.py init <project_dir> <name> <git_url> [port_range]")
|
|
sys.exit(1)
|
|
project_dir = sys.argv[2]
|
|
name = sys.argv[3]
|
|
git_url = sys.argv[4]
|
|
port_range = sys.argv[5] if len(sys.argv) > 5 else None
|
|
|
|
manager = WorkflowManager(project_dir)
|
|
manager.init_project(name, git_url, port_range)
|
|
print(f"项目已初始化: {name}")
|
|
|
|
elif command == "start_step":
|
|
if len(sys.argv) < 4:
|
|
print("Usage: workflow_manager.py start_step <project_dir> <step_id>")
|
|
sys.exit(1)
|
|
project_dir = sys.argv[2]
|
|
step_id = int(sys.argv[3])
|
|
|
|
manager = WorkflowManager(project_dir)
|
|
if manager.load_state():
|
|
if manager.start_step(step_id):
|
|
print(f"步骤 {step_id} 已开始")
|
|
else:
|
|
print(f"步骤 {step_id} 开始失败")
|
|
else:
|
|
print("未找到状态文件")
|
|
|
|
elif command == "complete_step":
|
|
if len(sys.argv) < 4:
|
|
print("Usage: workflow_manager.py complete_step <project_dir> <step_id> [notes]")
|
|
sys.exit(1)
|
|
project_dir = sys.argv[2]
|
|
step_id = int(sys.argv[3])
|
|
notes = sys.argv[4] if len(sys.argv) > 4 else ""
|
|
|
|
manager = WorkflowManager(project_dir)
|
|
if manager.load_state():
|
|
manager.complete_step(step_id, notes)
|
|
print(f"步骤 {step_id} 已完成")
|
|
else:
|
|
print("未找到状态文件")
|
|
|
|
elif command == "detect":
|
|
if len(sys.argv) < 3:
|
|
print("Usage: workflow_manager.py detect <project_dir>")
|
|
sys.exit(1)
|
|
project_dir = sys.argv[2]
|
|
|
|
manager = WorkflowManager(project_dir)
|
|
if manager.load_state():
|
|
tech_stack = TechStackDetector.detect(project_dir)
|
|
project_type = ProjectTypeDetector.detect(project_dir, tech_stack)
|
|
|
|
manager.project_info.tech_stack = tech_stack
|
|
manager.project_info.project_type = project_type
|
|
manager.start_step(2)
|
|
manager.complete_step(2, f"Detected tech stack: {[ts.value for ts in tech_stack]}, project type: {project_type.value}")
|
|
print(f"Tech stack: {[ts.value for ts in tech_stack]}, Project type: {project_type.value}")
|
|
else:
|
|
print("未找到状态文件")
|
|
|
|
elif command == "status":
|
|
if len(sys.argv) < 3:
|
|
print("Usage: workflow_manager.py status <project_dir>")
|
|
sys.exit(1)
|
|
project_dir = sys.argv[2]
|
|
|
|
manager = WorkflowManager(project_dir)
|
|
if manager.load_state():
|
|
progress = manager.get_progress()
|
|
print(f"进度: {progress['completed']}/{progress['total']} ({progress['percentage']:.1f}%)")
|
|
current = manager.get_current_step()
|
|
if current:
|
|
print(f"当前步骤: 步骤 {current.id} - {current.name}")
|
|
else:
|
|
print("未找到状态文件")
|
|
|
|
elif command == "resume":
|
|
if len(sys.argv) < 3:
|
|
print("Usage: workflow_manager.py resume <project_dir>")
|
|
sys.exit(1)
|
|
project_dir = sys.argv[2]
|
|
|
|
manager = WorkflowManager(project_dir)
|
|
can_resume, step_id = manager.can_resume()
|
|
if can_resume:
|
|
print(f"可以从步骤 {step_id} 恢复执行")
|
|
else:
|
|
print("无法恢复，请重新初始化")
|
|
|
|
elif command == "report":
|
|
if len(sys.argv) < 3:
|
|
print("Usage: workflow_manager.py report <project_dir>")
|
|
sys.exit(1)
|
|
project_dir = sys.argv[2]
|
|
|
|
manager = WorkflowManager(project_dir)
|
|
if manager.load_state():
|
|
print(manager.generate_report())
|
|
else:
|
|
print("未找到状态文件")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|