xst
d4c3d66efc
- fleet_runner.py: _run_export() 在 Excel 导出后自动生成 XMind - fleet_runner.py: 新增 --skip-xmind 参数（run/export 命令均支持） - export_XMind.py: 按模块分组 → 用例编号/标题/优先级/前置条件/步骤/预期/备注树形结构 - XMind 输出目录: output/xmind_reports/ 用法: python scripts/fleet_runner.py export --requirement <需求> # Excel + XMind python scripts/fleet_runner.py export --requirement <需求> --skip-xmind # 仅 Excel python scripts/export_XMind.py --base-name <名称> # 独立 XMind 导出
1721 lines
66 KiB
Python
1721 lines
66 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Agentic QE Fleet — 核心编排器
|
||
|
||
职责: 按战区顺序编排 14 个专业 Agent，协调 manifest 交接，
|
||
执行质量门禁裁决，触发 Excel 导出和知识沉淀。
|
||
|
||
用法:
|
||
python3 scripts/fleet_runner.py run --requirement <路径>
|
||
python3 scripts/fleet_runner.py prepare --requirement <路径>
|
||
python3 scripts/fleet_runner.py analyze --requirement <路径>
|
||
python3 scripts/fleet_runner.py design --requirement <路径>
|
||
python3 scripts/fleet_runner.py review --requirement <路径>
|
||
python3 scripts/fleet_runner.py export --requirement <路径>
|
||
python3 scripts/fleet_runner.py monitor --requirement <路径> --results <测试结果>
|
||
python3 scripts/fleet_runner.py status --requirement <路径>
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import subprocess
|
||
from datetime import datetime, timezone
|
||
|
||
# Windows GBK 编码兼容：强制 stdout/stderr 使用 UTF-8
|
||
if sys.platform == "win32":
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# 复用现有模块
|
||
from case_pipeline import (
|
||
resolve_requirement_path,
|
||
validate_requirement_file,
|
||
validate_knowledge_base,
|
||
validate_project_profile,
|
||
build_paths as build_legacy_paths,
|
||
ensure_output_dirs,
|
||
write_normalized_document,
|
||
select_technical_solution_files,
|
||
safe_read_text,
|
||
to_repo_relative,
|
||
REPO_ROOT,
|
||
REQUIREMENTS_DIR,
|
||
RAW_REQUIREMENTS_DIR,
|
||
TECHNICAL_SOLUTIONS_DIR,
|
||
PROJECT_PROFILE_FILE,
|
||
)
|
||
from export_excel import export_markdown_to_excel
|
||
from export_XMind import export_markdown_to_xmind
|
||
|
||
# Fleet 自有模块
|
||
from fleet_manifest import (
|
||
save_manifest,
|
||
load_manifest,
|
||
load_manifest_safe,
|
||
build_merged_manifest,
|
||
save_merged_manifest,
|
||
get_zone_status,
|
||
mark_zone_completed,
|
||
manifest_path,
|
||
ZONE_ORDER,
|
||
)
|
||
from fleet_agents import (
|
||
AGENT_REGISTRY,
|
||
load_agent_prompt,
|
||
list_zone_agents,
|
||
validate_all_agents,
|
||
render_agent_prompt,
|
||
)
|
||
|
||
# ── 配置加载 ──────────────────────────────────────────────────────────────
|
||
|
||
try:
|
||
import yaml
|
||
_HAS_YAML = True
|
||
except ImportError:
|
||
_HAS_YAML = False
|
||
|
||
|
||
def load_fleet_config() -> dict[str, Any]:
|
||
"""加载 fleet_config.yml。"""
|
||
config_path = REPO_ROOT / "fleet_config.yml"
|
||
if not config_path.exists():
|
||
return _default_fleet_config()
|
||
|
||
if _HAS_YAML:
|
||
with open(config_path, "r", encoding="utf-8") as fh:
|
||
return yaml.safe_load(fh) or {}
|
||
else:
|
||
# 纯 Python fallback: 解析简单的 YAML 子集
|
||
return _parse_simple_yaml_config(config_path)
|
||
|
||
|
||
def _default_fleet_config() -> dict[str, Any]:
|
||
return {
|
||
"battle_zones": {
|
||
"prepare": {"enabled": True},
|
||
"analyze": {"enabled": True},
|
||
"design": {"enabled": True},
|
||
"review": {"enabled": True},
|
||
"monitor": {"enabled": True, "auto_confirm": True},
|
||
},
|
||
"quality_gate": {"min_coverage": 0.95, "max_blockers": 0},
|
||
"output": {"excel_format": "yunxiao", "snapshot_keep": 3, "auto_sync_maintained": True},
|
||
}
|
||
|
||
|
||
def _parse_simple_yaml_config(path: Path) -> dict[str, Any]:
|
||
"""简易 YAML 解析器（当 PyYAML 不可用时）。"""
|
||
import re
|
||
result: dict[str, Any] = {}
|
||
current: dict[str, Any] = result
|
||
stack: list[tuple[str, dict[str, Any]]] = []
|
||
|
||
for raw_line in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw_line.rstrip()
|
||
if not line or line.strip().startswith("#"):
|
||
continue
|
||
|
||
indent = len(raw_line) - len(raw_line.lstrip())
|
||
key_match = re.match(r"^(\s*)([\w_-]+)\s*:\s*(.*)", line)
|
||
if not key_match:
|
||
continue
|
||
|
||
key = key_match.group(2)
|
||
value = key_match.group(3).strip().strip('"').strip("'")
|
||
|
||
# 处理缩进回退
|
||
while stack and stack[-1][0] >= indent:
|
||
stack.pop()
|
||
current = stack[-1][1] if stack else result
|
||
|
||
if value in ("true", "True"):
|
||
current[key] = True
|
||
elif value in ("false", "False"):
|
||
current[key] = False
|
||
elif value == "" or value == "{}":
|
||
current[key] = {}
|
||
current = current[key]
|
||
stack.append((indent, current))
|
||
elif value == "[]":
|
||
current[key] = []
|
||
elif re.match(r"^- ", raw_line):
|
||
if key not in current:
|
||
current[key] = []
|
||
current[key].append(value)
|
||
elif re.match(r"^\d+(\.\d+)?$", value):
|
||
current[key] = float(value) if "." in value else int(value)
|
||
else:
|
||
current[key] = value
|
||
|
||
return result
|
||
|
||
|
||
# ── 路径和目录 ────────────────────────────────────────────────────────────
|
||
|
||
BUILTIN_INPUT_DIRS = [
|
||
REPO_ROOT / "output" / "analysis",
|
||
REPO_ROOT / "output" / "test_points",
|
||
REPO_ROOT / "output" / "test_cases",
|
||
REPO_ROOT / "output" / "excel_reports",
|
||
REPO_ROOT / "output" / "normalized_inputs",
|
||
REPO_ROOT / "output" / "versions",
|
||
]
|
||
|
||
|
||
def build_all_output_dirs(base_name: str) -> None:
|
||
"""创建所有输出目录。"""
|
||
legacy_paths = build_legacy_paths(base_name)
|
||
ensure_output_dirs(legacy_paths)
|
||
# 确保 manifest 目录
|
||
(REPO_ROOT / "output" / "manifests").mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
# ── 编排核心 ──────────────────────────────────────────────────────────────
|
||
|
||
def run_zone(zone: str, base_name: str, requirement_path: Path, config: dict[str, Any]) -> dict[str, Any]:
|
||
"""运行一个战区，返回该战区的 manifest 数据。"""
|
||
|
||
zone_config = config.get("battle_zones", {}).get(zone, {})
|
||
if not zone_config.get("enabled", True):
|
||
print(f"⏭️ {zone.upper()} 战区已禁用，跳过。")
|
||
return {}
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"🚀 {zone.upper()} 战区启动")
|
||
print(f"{'='*60}")
|
||
|
||
agents = list_zone_agents(zone)
|
||
print(f"📋 Agent: {', '.join(agents)}")
|
||
|
||
# 加载前序战区 manifest 作为上下文
|
||
context = _build_zone_context(base_name, zone, requirement_path)
|
||
|
||
# 打印当前可用的输入文件
|
||
_print_context_files(context, zone)
|
||
|
||
# 战区特定逻辑
|
||
zone_handlers = {
|
||
"prepare": _run_prepare_zone,
|
||
"analyze": _run_analyze_zone,
|
||
"design": _run_design_zone,
|
||
"execute": _run_execute_zone,
|
||
"review": _run_review_zone,
|
||
"monitor": _run_monitor_zone,
|
||
}
|
||
|
||
handler = zone_handlers.get(zone)
|
||
if handler is None:
|
||
raise ValueError(f"未知战区: {zone}")
|
||
|
||
result = handler(base_name, requirement_path, context, config, agents)
|
||
|
||
# 标记完成
|
||
if result:
|
||
mark_zone_completed(base_name, zone)
|
||
print(f"✅ {zone.upper()} 战区完成 → {manifest_path(base_name, zone)}")
|
||
|
||
return result
|
||
|
||
|
||
def _build_zone_context(base_name: str, zone: str, requirement_path: Path) -> dict[str, Any]:
|
||
"""构建战区运行的上下文。"""
|
||
context: dict[str, Any] = {
|
||
"base_name": base_name,
|
||
"requirement_file": str(requirement_path),
|
||
"requirement_stem": requirement_path.stem,
|
||
"repo_root": str(REPO_ROOT),
|
||
"project_profile": str(PROJECT_PROFILE_FILE),
|
||
}
|
||
|
||
# 加载前序战区 manifest
|
||
for prev_zone in ZONE_ORDER:
|
||
if prev_zone == zone:
|
||
break
|
||
manifest = load_manifest_safe(base_name, prev_zone)
|
||
if manifest:
|
||
context[f"manifest_{prev_zone}"] = manifest
|
||
context[f"manifest_{prev_zone}_path"] = str(manifest_path(base_name, prev_zone))
|
||
|
||
return context
|
||
|
||
|
||
def _print_context_files(context: dict[str, Any], zone: str) -> None:
|
||
"""打印战区可用的上下文文件。"""
|
||
from fleet_manifest import load_manifest_safe
|
||
|
||
items: list[tuple[str, str]] = []
|
||
|
||
# 从 manifest 中提取产物文件
|
||
for prev_zone in ZONE_ORDER:
|
||
if prev_zone == zone:
|
||
break
|
||
manifest = context.get(f"manifest_{prev_zone}")
|
||
if manifest is None:
|
||
continue
|
||
# 收集文件路径
|
||
for key in manifest:
|
||
if key.endswith("_file") or key.endswith("_files"):
|
||
value = manifest[key]
|
||
if isinstance(value, str) and Path(value).exists():
|
||
items.append((key, value))
|
||
elif isinstance(value, list):
|
||
for v in value:
|
||
if isinstance(v, str) and Path(v).exists():
|
||
items.append((key, v))
|
||
|
||
if items:
|
||
print("📂 可用上下文文件:")
|
||
for key, path in items[:10]:
|
||
print(f" - {key}: {path}")
|
||
|
||
|
||
# ── Prepare 战区 ──────────────────────────────────────────────────────────
|
||
|
||
def _run_prepare_zone(
|
||
base_name: str,
|
||
requirement_path: Path,
|
||
context: dict[str, Any],
|
||
config: dict[str, Any],
|
||
agents: list[str],
|
||
) -> dict[str, Any]:
|
||
"""执行 Prepare 战区: document-parser → knowledge-activator。"""
|
||
|
||
# 1. 文档解析 (借用现有 case_pipeline 的能力)
|
||
validate_requirement_file(requirement_path)
|
||
validate_knowledge_base()
|
||
validate_project_profile()
|
||
|
||
technical_solution_files = select_technical_solution_files(requirement_path)
|
||
normalized_dir = REPO_ROOT / "output" / "normalized_inputs" / base_name
|
||
normalized_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
normalized_requirement_file = write_normalized_document(
|
||
source_path=requirement_path,
|
||
target_path=normalized_dir / "requirement.md",
|
||
document_role="需求文档",
|
||
)
|
||
|
||
normalized_technical_solution_files = [
|
||
write_normalized_document(
|
||
source_path=path,
|
||
target_path=normalized_dir / f"technical_solution_{idx:02d}.md",
|
||
document_role="技术方案",
|
||
)
|
||
for idx, path in enumerate(technical_solution_files, start=1)
|
||
]
|
||
|
||
# 文档置信度评估
|
||
requirement_body = safe_read_text(normalized_requirement_file)
|
||
confidence = _estimate_document_confidence(requirement_body, requirement_path.suffix.lower())
|
||
|
||
# 2. 知识激活
|
||
activated_knowledge = _activate_knowledge(requirement_path, config)
|
||
|
||
# 检测知识缺口
|
||
knowledge_gaps = _detect_knowledge_gaps(activated_knowledge, requirement_body)
|
||
|
||
# 组装 manifest
|
||
prepare_manifest = {
|
||
"base_name": base_name,
|
||
"requirement_source_file": str(requirement_path),
|
||
"requirement_input_type": requirement_path.suffix.lower().lstrip("."),
|
||
"normalized_requirement_file": str(normalized_requirement_file),
|
||
"normalized_dir": str(normalized_dir),
|
||
"technical_solution_files": [str(p) for p in technical_solution_files],
|
||
"normalized_technical_solution_files": [str(p) for p in normalized_technical_solution_files],
|
||
"project_profile_file": str(PROJECT_PROFILE_FILE),
|
||
"document_confidence": confidence,
|
||
"activated_knowledge": activated_knowledge,
|
||
"knowledge_gaps": knowledge_gaps,
|
||
"agent_notes": {
|
||
"document-parser": f"解析完成，置信度 {confidence.get('requirement', 0):.0%}",
|
||
"knowledge-activator": f"激活 {len(activated_knowledge.get('terminology', {}).get('permanent', []))} 常驻 + "
|
||
f"{len(activated_knowledge.get('terminology', {}).get('optional', []))} 可选术语",
|
||
},
|
||
}
|
||
|
||
save_manifest(base_name, "prepare", prepare_manifest)
|
||
return prepare_manifest
|
||
|
||
|
||
def _estimate_document_confidence(text: str, suffix: str) -> dict[str, Any]:
|
||
"""估算文档解析置信度。"""
|
||
confidence = 1.0 if suffix == ".md" else 0.90 if suffix == ".docx" else 0.80 if suffix == ".pdf" else 0.75
|
||
if not text.strip():
|
||
confidence = 0.1
|
||
elif "> ⚠️ 待确认：PDF 未提取到可用文本" in text:
|
||
confidence = 0.05
|
||
elif text.count("> ⚠️") > 3:
|
||
confidence -= 0.1
|
||
return {
|
||
"requirement": round(max(0.0, confidence), 2),
|
||
"issues": ["扫描件/图片型PDF"] if "未提取到可用文本" in text else [],
|
||
}
|
||
|
||
|
||
def _activate_knowledge(requirement_path: Path, config: dict[str, Any]) -> dict[str, Any]:
|
||
"""按需求内容自动激活知识库文件。"""
|
||
from case_pipeline import (
|
||
CORE_TERMINOLOGY_FILE,
|
||
KNOWLEDGE_BASE_FILES,
|
||
OPTIONAL_TERMINOLOGY_RULES,
|
||
)
|
||
|
||
requirement_text = safe_read_text(requirement_path)
|
||
combined = f"{requirement_path.name}\n{requirement_text}"
|
||
|
||
# 常驻术语（始终激活）
|
||
permanent = [str(CORE_TERMINOLOGY_FILE)]
|
||
|
||
# 关键词匹配可选术语
|
||
optional: list[dict[str, Any]] = []
|
||
for rule in OPTIONAL_TERMINOLOGY_RULES:
|
||
matched_keywords = [kw for kw in rule["keywords"] if kw in combined]
|
||
if matched_keywords:
|
||
optional.append({
|
||
"name": rule["name"],
|
||
"path": str(rule["path"]),
|
||
"matched_keywords": matched_keywords[:10],
|
||
"activation_reason": f"匹配关键词: {', '.join(matched_keywords[:5])}",
|
||
})
|
||
|
||
# 语义匹配知识库（基于 Jaccard 相似度）
|
||
from case_pipeline import to_ngrams, jaccard_similarity
|
||
requirement_tokens = to_ngrams(combined)
|
||
|
||
semantic_matches: list[dict[str, Any]] = []
|
||
threshold = config.get("knowledge_activation", {}).get("semantic_match_threshold", 0.08)
|
||
|
||
for kb_file in KNOWLEDGE_BASE_FILES:
|
||
if kb_file == CORE_TERMINOLOGY_FILE:
|
||
continue
|
||
if not kb_file.exists():
|
||
continue
|
||
kb_text = safe_read_text(kb_file)
|
||
score = jaccard_similarity(requirement_tokens, to_ngrams(f"{kb_file.name}\n{kb_text}"))
|
||
if score >= threshold:
|
||
semantic_matches.append({
|
||
"path": str(kb_file),
|
||
"score": round(score, 4),
|
||
"category": _classify_kb_file(kb_file),
|
||
})
|
||
|
||
return {
|
||
"terminology": {
|
||
"permanent": permanent,
|
||
"optional": optional,
|
||
},
|
||
"semantic_matches": sorted(semantic_matches, key=lambda x: x["score"], reverse=True),
|
||
}
|
||
|
||
|
||
def _classify_kb_file(path: Path) -> str:
|
||
"""分类知识库文件。"""
|
||
path_str = str(path)
|
||
if "02_history" in path_str:
|
||
return "history"
|
||
if "03_best_practices" in path_str:
|
||
return "best_practice"
|
||
if "01_standards" in path_str:
|
||
return "standard"
|
||
return "other"
|
||
|
||
|
||
def _detect_knowledge_gaps(activated_knowledge: dict[str, Any], requirement_text: str) -> list[dict[str, str]]:
|
||
"""检测知识库缺口。"""
|
||
gaps: list[dict[str, str]] = []
|
||
# 检查是否有必要的知识类别缺失
|
||
has_history = any(
|
||
m.get("category") == "history"
|
||
for m in activated_knowledge.get("semantic_matches", [])
|
||
)
|
||
has_best_practice = any(
|
||
m.get("category") == "best_practice"
|
||
for m in activated_knowledge.get("semantic_matches", [])
|
||
)
|
||
if not has_history:
|
||
gaps.append({"category": "history", "suggestion": "未激活任何历史缺陷/易漏场景，建议补充相关知识库条目"})
|
||
if not has_best_practice:
|
||
gaps.append({"category": "best_practice", "suggestion": "未激活任何最佳实践范例，建议补充同类型需求的优秀用例"})
|
||
return gaps
|
||
|
||
|
||
# ── Analyze 战区 ──────────────────────────────────────────────────────────
|
||
|
||
def _run_analyze_zone(
|
||
base_name: str,
|
||
requirement_path: Path,
|
||
context: dict[str, Any],
|
||
config: dict[str, Any],
|
||
agents: list[str],
|
||
) -> dict[str, Any]:
|
||
"""执行 Analyze 战区: requirement-analyzer + conflict-detector → risk-assessor。"""
|
||
|
||
prepare_manifest = context.get("manifest_prepare", {})
|
||
from case_pipeline import (
|
||
find_related_requirements,
|
||
build_conflict_candidates,
|
||
build_confirmation_gate,
|
||
write_relation_report,
|
||
find_relevant_decision_files,
|
||
inspect_decision_file,
|
||
)
|
||
|
||
# 加载标准化需求
|
||
normalized_requirement_file = Path(
|
||
prepare_manifest.get("normalized_requirement_file",
|
||
f"output/normalized_inputs/{base_name}/requirement.md")
|
||
)
|
||
if not normalized_requirement_file.is_absolute():
|
||
normalized_requirement_file = REPO_ROOT / normalized_requirement_file
|
||
|
||
technical_solution_files = [
|
||
Path(p) for p in prepare_manifest.get("normalized_technical_solution_files", [])
|
||
]
|
||
|
||
# 关联需求识别
|
||
related_requirements = find_related_requirements(requirement_path)
|
||
print(f"📚 关联需求: {len(related_requirements)} 个")
|
||
|
||
# 冲突检测
|
||
conflicts = build_conflict_candidates(requirement_path, related_requirements)
|
||
print(f"⚠️ 冲突候选: {len(conflicts)} 个")
|
||
|
||
# 写入关联与冲突报告
|
||
analysis_dir = REPO_ROOT / "output" / "analysis"
|
||
analysis_dir.mkdir(parents=True, exist_ok=True)
|
||
relation_report_path = analysis_dir / f"{base_name}_关联与冲突.md"
|
||
write_relation_report(
|
||
requirement_path, related_requirements,
|
||
[Path(p) for p in prepare_manifest.get("technical_solution_files", [])],
|
||
conflicts, relation_report_path,
|
||
)
|
||
|
||
# 风险评估 (risk-assessor 的输入)
|
||
risk_matrix = _build_risk_matrix(requirement_path, conflicts, related_requirements)
|
||
risk_report_path = analysis_dir / f"{base_name}_风险评估.md"
|
||
_write_risk_report(risk_matrix, risk_report_path, base_name)
|
||
|
||
# 确认门禁
|
||
confirmation_gate = build_confirmation_gate(
|
||
requirement_path=requirement_path,
|
||
base_name=base_name,
|
||
related_requirements=related_requirements,
|
||
conflicts=conflicts,
|
||
normalized_requirement_file=normalized_requirement_file,
|
||
normalized_technical_solution_files=[
|
||
p for p in technical_solution_files if isinstance(p, Path)
|
||
],
|
||
)
|
||
|
||
analyze_manifest = {
|
||
"base_name": base_name,
|
||
"analysis_file": str(analysis_dir / f"{base_name}_分析.md"),
|
||
"relation_report_file": str(relation_report_path),
|
||
"risk_report_file": str(risk_report_path),
|
||
"related_requirements": [
|
||
{"path": str(item["path"]), "similarity": round(item["score"], 6)}
|
||
for item in related_requirements
|
||
],
|
||
"conflict_candidates_count": len(conflicts),
|
||
"conflict_summary": _summarize_conflicts(conflicts),
|
||
"risk_matrix": risk_matrix,
|
||
"confirmation_gate": confirmation_gate,
|
||
"agent_notes": {
|
||
"requirement-analyzer": f"识别 {len(related_requirements)} 个关联需求",
|
||
"conflict-detector": f"检测到 {len(conflicts)} 个冲突候选",
|
||
"risk-assessor": f"识别 {len(risk_matrix.get('risks', []))} 个风险项",
|
||
},
|
||
}
|
||
|
||
save_manifest(base_name, "analyze", analyze_manifest)
|
||
return analyze_manifest
|
||
|
||
|
||
def _build_risk_matrix(
|
||
requirement_path: Path,
|
||
conflicts: list[dict[str, Any]],
|
||
related_requirements: list[dict[str, Any]],
|
||
) -> dict[str, Any]:
|
||
"""构建风险矩阵。"""
|
||
risks: list[dict[str, Any]] = []
|
||
requirement_text = safe_read_text(requirement_path)
|
||
|
||
# 风险检测维度
|
||
checks = [
|
||
("资损", ["金额", "支付", "退款", "扣减", "优惠", "积分", "券", "库存", "手续费", "税费"], "financial"),
|
||
("可用性", ["超时", "弱网", "并发", "降级", "熔断", "限流", "重试", "幂等"], "availability"),
|
||
("数据", ["脏数据", "迁移", "精度", "隔离", "删除", "空值", "租户"], "data"),
|
||
("合规", ["鉴权", "留痕", "审批", "实名", "隐私", "加密", "脱敏"], "compliance"),
|
||
("兼容性", ["H5", "小程序", "App", "浏览器", "多端", "版本", "灰度"], "compatibility"),
|
||
]
|
||
|
||
for category, keywords, risk_id in checks:
|
||
matched = [kw for kw in keywords if kw in requirement_text]
|
||
if matched:
|
||
likelihood = min(5, len(matched))
|
||
impact = 5 if risk_id in ("financial", "compliance") else 4 if risk_id in ("availability", "data") else 3
|
||
risks.append({
|
||
"id": f"RISK-{risk_id.upper()}",
|
||
"category": category,
|
||
"keywords_matched": matched,
|
||
"likelihood": likelihood,
|
||
"impact": impact,
|
||
"score": likelihood * impact,
|
||
"level": "P0" if likelihood * impact >= 15 else "P1" if likelihood * impact >= 10 else "P2",
|
||
"conflict_amplified": bool(conflicts),
|
||
})
|
||
|
||
# 冲突放大风险
|
||
if conflicts:
|
||
for risk in risks:
|
||
if risk["category"] in ("资损", "数据", "合规"):
|
||
risk["conflict_amplified"] = True
|
||
risk["score"] = min(25, risk["score"] + 3)
|
||
risk["level"] = "P0" if risk["score"] >= 15 else risk["level"]
|
||
|
||
return {
|
||
"risks": sorted(risks, key=lambda r: r["score"], reverse=True),
|
||
"total": len(risks),
|
||
"p0_count": sum(1 for r in risks if r["level"] == "P0"),
|
||
"p1_count": sum(1 for r in risks if r["level"] == "P1"),
|
||
}
|
||
|
||
|
||
def _write_risk_report(risk_matrix: dict[str, Any], report_path: Path, base_name: str) -> None:
|
||
"""写入风险评估报告。"""
|
||
lines = [
|
||
f"# {base_name} 风险评估报告",
|
||
"",
|
||
f"> 生成时间: {datetime.now(timezone.utc).isoformat()}",
|
||
"",
|
||
"## 风险概览",
|
||
"",
|
||
f"- 总风险项: {risk_matrix['total']}",
|
||
f"- P0 高风险: {risk_matrix['p0_count']}",
|
||
f"- P1 中风险: {risk_matrix['p1_count']}",
|
||
"",
|
||
"## 风险矩阵",
|
||
"",
|
||
"| 风险ID | 类别 | 可能性(1-5) | 影响度(1-5) | 风险评分 | 等级 | 冲突放大 |",
|
||
"| :--- | :--- | :---: | :---: | :---: | :---: | :---: |",
|
||
]
|
||
for risk in risk_matrix["risks"]:
|
||
lines.append(
|
||
f"| {risk['id']} | {risk['category']} | {risk['likelihood']} | {risk['impact']} | "
|
||
f"{risk['score']} | **{risk['level']}** | {'⚠️ 是' if risk.get('conflict_amplified') else '否'} |"
|
||
)
|
||
lines.extend([
|
||
"",
|
||
"## 风险缓解建议",
|
||
"",
|
||
])
|
||
for risk in risk_matrix["risks"]:
|
||
if risk["level"] == "P0":
|
||
lines.append(f"- **{risk['id']} ({risk['category']})**: 必须 100% 覆盖，建议增加 P0 测试点和专项回归用例。")
|
||
report_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
|
||
|
||
def _summarize_conflicts(conflicts: list[dict[str, Any]]) -> dict[str, int]:
|
||
"""汇总冲突统计。"""
|
||
types: dict[str, int] = {}
|
||
for c in conflicts:
|
||
t = c.get("type", "未知")
|
||
types[t] = types.get(t, 0) + 1
|
||
return types
|
||
|
||
|
||
# ── Design 战区 ──────────────────────────────────────────────────────────
|
||
|
||
def _run_design_zone(
|
||
base_name: str,
|
||
requirement_path: Path,
|
||
context: dict[str, Any],
|
||
config: dict[str, Any],
|
||
agents: list[str],
|
||
) -> dict[str, Any]:
|
||
"""执行 Design 战区: strategist → (testpoint-designer + data-builder) → case-designer。"""
|
||
|
||
prepare_manifest = context.get("manifest_prepare", {})
|
||
analyze_manifest = context.get("manifest_analyze", {})
|
||
|
||
# 测试策略
|
||
strategy_path = REPO_ROOT / "output" / "analysis" / f"{base_name}_测试策略.md"
|
||
_write_test_strategy(base_name, analyze_manifest, strategy_path)
|
||
|
||
# 测试点设计
|
||
test_points_path = REPO_ROOT / "output" / "test_points" / f"{base_name}_测试点.md"
|
||
test_points_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 测试数据构建
|
||
data_path = REPO_ROOT / "output" / "analysis" / f"{base_name}_测试数据.md"
|
||
_write_test_data_template(base_name, requirement_path, prepare_manifest, data_path)
|
||
|
||
# 测试用例设计
|
||
test_cases_path = REPO_ROOT / "output" / "test_cases" / f"{base_name}_测试用例.md"
|
||
test_cases_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
risk_matrix = analyze_manifest.get("risk_matrix", {})
|
||
p0_count = risk_matrix.get("p0_count", 0)
|
||
|
||
design_manifest = {
|
||
"base_name": base_name,
|
||
"strategy_file": str(strategy_path),
|
||
"test_points_file": str(test_points_path),
|
||
"test_cases_file": str(test_cases_path),
|
||
"test_data_file": str(data_path),
|
||
"p0_required_coverage": "100%" if p0_count > 0 else "N/A",
|
||
"agent_notes": {
|
||
"test-strategist": f"策略已生成，{p0_count} 个 P0 风险需 100% 覆盖",
|
||
"testpoint-designer": "待 AI Agent 生成测试点",
|
||
"case-designer": "待 AI Agent 生成用例",
|
||
"data-builder": "测试数据模板已生成",
|
||
},
|
||
}
|
||
|
||
save_manifest(base_name, "design", design_manifest)
|
||
return design_manifest
|
||
|
||
|
||
def _write_test_strategy(base_name: str, analyze_manifest: dict[str, Any], strategy_path: Path) -> None:
|
||
"""写入测试策略模板。"""
|
||
risk_matrix = analyze_manifest.get("risk_matrix", {})
|
||
risks = risk_matrix.get("risks", [])
|
||
|
||
lines = [
|
||
f"# {base_name} 测试策略",
|
||
"",
|
||
"## 1. 测试金字塔",
|
||
"",
|
||
"| 层级 | 占比 | 覆盖重点 | 工具 |",
|
||
"| :--- | :---: | :--- | :--- |",
|
||
"| L1 单元测试 | 40% | 核心逻辑、计算、状态机 | 开发自测 |",
|
||
"| L2 API 测试 | 35% | 接口契约、参数校验、权限、幂等 | Postman/Pytest |",
|
||
"| L3 UI 测试 | 20% | 主流程、关键交互、端到端 | Playwright/Selenium |",
|
||
"| L4 手工探索 | 5% | 易用性、视觉、非确定性场景 | 人工 |",
|
||
"",
|
||
"## 2. P0 必测清单",
|
||
"",
|
||
]
|
||
for risk in risks:
|
||
if risk["level"] == "P0":
|
||
lines.append(f"- [{risk['id']}] **{risk['category']}**: {', '.join(risk.get('keywords_matched', []))}")
|
||
|
||
lines.extend([
|
||
"",
|
||
"## 3. 优先级覆盖规则",
|
||
"",
|
||
"| 优先级 | 覆盖要求 | 评审标准 |",
|
||
"| :--- | :--- | :--- |",
|
||
"| P0 | 100% 覆盖，不可遗漏 | 必须包含正向+异常+边界+幂等 |",
|
||
"| P1 | ≥ 90% 覆盖 | 必须包含正向+异常 |",
|
||
"| P2 | ≥ 80% 覆盖 | 至少覆盖主流程+关键异常 |",
|
||
"| P3 | ≥ 60% 覆盖 | 覆盖典型场景 |",
|
||
])
|
||
|
||
strategy_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
|
||
|
||
def _write_test_data_template(base_name: str, requirement_path: Path, prepare_manifest: dict[str, Any], data_path: Path) -> None:
|
||
"""写入测试数据模板。"""
|
||
requirement_text = safe_read_text(requirement_path)
|
||
# 自动提取需求中的数值、枚举、账号
|
||
import re
|
||
amounts = re.findall(r'\d+(?:\.\d+)?(?:元|分|%)', requirement_text)
|
||
ids = re.findall(r'(?:ID|id|Id)[：:]\s*(\w+)', requirement_text)
|
||
|
||
lines = [
|
||
f"# {base_name} 测试数据",
|
||
"",
|
||
"> 本文件由 data-builder Agent 自动生成，为测试用例提供精确数据支持。",
|
||
"",
|
||
"## 自动提取的候选数据",
|
||
"",
|
||
]
|
||
if amounts:
|
||
lines.append("### 金额/数值")
|
||
for a in amounts[:10]:
|
||
lines.append(f"- `{a}`")
|
||
lines.append("")
|
||
|
||
if ids:
|
||
lines.append("### ID/编码")
|
||
for i in ids[:10]:
|
||
lines.append(f"- `{i}`")
|
||
lines.append("")
|
||
|
||
lines.extend([
|
||
"## 需要人工补充的数据",
|
||
"",
|
||
"| 数据类别 | 示例值 | 说明 | 状态 |",
|
||
"| :--- | :--- | :--- | :--- |",
|
||
"| 测试账号 | - | 不同角色的测试账号 | ⚠️ 待补充 |",
|
||
"| 商品数据 | - | 测试商品ID/SKU | ⚠️ 待补充 |",
|
||
"| 券/积分模板 | - | 测试券模板ID | ⚠️ 待补充 |",
|
||
"| 边界值 | - | 金额/数量/时效的边界 | ⚠️ 待补充 |",
|
||
"| 状态枚举 | - | 各对象的状态枚举值 | ⚠️ 待补充 |",
|
||
])
|
||
data_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
|
||
|
||
# ── Execute 战区 ─────────────────────────────────────────────────────────
|
||
|
||
def _run_execute_zone(
|
||
base_name: str,
|
||
requirement_path: Path,
|
||
context: dict[str, Any],
|
||
config: dict[str, Any],
|
||
agents: list[str],
|
||
) -> dict[str, Any]:
|
||
"""执行 Execute 战区: web-executor + mobile-executor → result-reporter。"""
|
||
|
||
prepare_manifest = context.get("manifest_prepare", {})
|
||
design_manifest = context.get("manifest_design", {})
|
||
execute_config = config.get("battle_zones", {}).get("execute", {})
|
||
|
||
test_cases_path = Path(design_manifest.get("test_cases_file",
|
||
f"output/test_cases/{base_name}_测试用例.md"))
|
||
test_data_path = Path(design_manifest.get("test_data_file",
|
||
f"output/analysis/{base_name}_测试数据.md"))
|
||
|
||
# 输出目录
|
||
screenshots_dir = REPO_ROOT / "output" / "screenshots" / base_name
|
||
screenshots_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
execution_dir = REPO_ROOT / "output" / "execution"
|
||
execution_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 生成 Playwright 测试脚本
|
||
playwright_script = _generate_playwright_script(
|
||
base_name, test_cases_path, test_data_path,
|
||
screenshots_dir, execute_config
|
||
)
|
||
|
||
# 生成 Appium 测试脚本
|
||
appium_script = _generate_appium_script(
|
||
base_name, test_cases_path, test_data_path,
|
||
screenshots_dir, execute_config
|
||
)
|
||
|
||
# 生成综合执行报告
|
||
execution_report_path = execution_dir / f"{base_name}_执行报告.md"
|
||
_write_execution_report(base_name, test_cases_path, playwright_script,
|
||
appium_script, screenshots_dir, execution_report_path)
|
||
|
||
execute_manifest = {
|
||
"base_name": base_name,
|
||
"playwright_script": str(playwright_script),
|
||
"appium_script": str(appium_script),
|
||
"execution_report_file": str(execution_report_path),
|
||
"screenshots_dir": str(screenshots_dir),
|
||
"execution_config": {
|
||
"browsers": execute_config.get("browsers", ["chromium"]),
|
||
"mobile_platforms": execute_config.get("mobile_platforms", ["android"]),
|
||
"screenshot_on_failure": execute_config.get("screenshot_on_failure", True),
|
||
"screenshot_on_step": execute_config.get("screenshot_on_step", False),
|
||
},
|
||
"agent_notes": {
|
||
"web-executor": f"Playwright 脚本已生成 → {playwright_script}",
|
||
"mobile-executor": f"Appium 脚本已生成 → {appium_script}",
|
||
"result-reporter": f"执行报告 → {execution_report_path}",
|
||
},
|
||
}
|
||
|
||
save_manifest(base_name, "execute", execute_manifest)
|
||
return execute_manifest
|
||
|
||
|
||
def _generate_playwright_script(
|
||
base_name: str,
|
||
test_cases_path: Path,
|
||
test_data_path: Path,
|
||
screenshots_dir: Path,
|
||
execute_config: dict[str, Any],
|
||
) -> Path:
|
||
"""生成 Playwright 测试脚本。"""
|
||
script_dir = REPO_ROOT / "output" / "execution" / base_name
|
||
script_dir.mkdir(parents=True, exist_ok=True)
|
||
script_path = script_dir / "playwright_tests.py"
|
||
|
||
browsers = execute_config.get("browsers", ["chromium"])
|
||
screenshot_on_failure = execute_config.get("screenshot_on_failure", True)
|
||
screenshot_on_step = execute_config.get("screenshot_on_step", False)
|
||
timeout = execute_config.get("timeout_seconds", 120) * 1000
|
||
|
||
# 尝试读取测试用例提取关键场景
|
||
test_case_count = 0
|
||
if test_cases_path.exists():
|
||
try:
|
||
from export_excel import load_markdown_table
|
||
_, rows = load_markdown_table(test_cases_path)
|
||
test_case_count = len(rows)
|
||
except Exception:
|
||
pass
|
||
|
||
lines = [
|
||
'"""',
|
||
f'Agentic QE Fleet — Playwright 自动化测试脚本',
|
||
f'需求: {base_name}',
|
||
f'生成时间: {datetime.now(timezone.utc).isoformat()}',
|
||
f'目标浏览器: {", ".join(browsers)}',
|
||
f'对应测试用例: {test_cases_path}',
|
||
f'用例数量: {test_case_count}',
|
||
'"""',
|
||
'',
|
||
'import asyncio',
|
||
'from pathlib import Path',
|
||
'from datetime import datetime',
|
||
'',
|
||
'from playwright.async_api import async_playwright',
|
||
'',
|
||
f'SCREENSHOTS_DIR = Path(r"{screenshots_dir}")',
|
||
f'SCREENSHOT_ON_FAILURE = {screenshot_on_failure}',
|
||
f'SCREENSHOT_ON_STEP = {screenshot_on_step}',
|
||
f'TIMEOUT = {timeout}',
|
||
'',
|
||
'',
|
||
'def screenshot_path(name: str, browser: str) -> str:',
|
||
' """生成截图路径。"""',
|
||
' SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)',
|
||
' ts = datetime.now().strftime("%Y%m%d_%H%M%S")',
|
||
' return str(SCREENSHOTS_DIR / f"{name}_{browser}_{ts}.png")',
|
||
'',
|
||
'',
|
||
'async def run_test(browser_type: str, browser_name: str):',
|
||
' """执行单个浏览器的测试。"""',
|
||
' results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
|
||
'',
|
||
' async with async_playwright() as p:',
|
||
f" browser_launcher = getattr(p, browser_type)",
|
||
f" browser = await browser_launcher.launch(headless=True)",
|
||
' context = await browser.new_context(',
|
||
' viewport={"width": 1920, "height": 1080},',
|
||
' locale="zh-CN",',
|
||
' )',
|
||
' page = await context.new_page()',
|
||
f" page.set_default_timeout(TIMEOUT)",
|
||
'',
|
||
]
|
||
|
||
# 生成基础测试骨架
|
||
lines.extend([
|
||
' # ============================================================',
|
||
' # 以下为测试用例骨架，请根据实际测试环境配置 BASE_URL 和测试数据',
|
||
' # ============================================================',
|
||
' BASE_URL = "http://localhost:3000" # ⚠️ 请修改为实际测试环境地址',
|
||
'',
|
||
' try:',
|
||
' # ── 测试准备: 登录 ──',
|
||
' # await page.goto(f"{BASE_URL}/login")',
|
||
f" # await page.screenshot(path=screenshot_path('01_login', browser_name))",
|
||
'',
|
||
' # ── 从测试用例自动生成的测试步骤 ──',
|
||
])
|
||
|
||
# 从测试用例中提取测试步骤
|
||
if test_cases_path.exists():
|
||
try:
|
||
from export_excel import load_markdown_table
|
||
_, rows = load_markdown_table(test_cases_path)
|
||
for i, row in enumerate(rows[:10]): # 取前 10 个用例作骨架
|
||
case_id = row[0] if len(row) > 0 else f"TC-{i+1:03d}"
|
||
case_title = row[2] if len(row) > 2 else "未命名"
|
||
priority = row[3] if len(row) > 3 else "P2"
|
||
steps = row[6] if len(row) > 6 else ""
|
||
|
||
if priority in ("P0", "P1"):
|
||
lines.append(f' # [{priority}] {case_id}: {case_title}')
|
||
step_lines = steps.replace('<br>', '\n').replace('\\n', '\n').split('\n')
|
||
for step in step_lines[:3]:
|
||
clean_step = step.strip().strip('0123456789.、) ')
|
||
if clean_step and len(clean_step) > 3:
|
||
lines.append(f' # → {clean_step}')
|
||
lines.append(f" # await page.screenshot(path=screenshot_path('{case_id}', browser_name))")
|
||
lines.append('')
|
||
except Exception:
|
||
pass
|
||
|
||
lines.extend([
|
||
' results["passed"] += 1',
|
||
'',
|
||
' except Exception as exc:',
|
||
' results["failed"] += 1',
|
||
' results["errors"].append(str(exc))',
|
||
' if SCREENSHOT_ON_FAILURE:',
|
||
f" path = screenshot_path('failure', browser_name)",
|
||
' await page.screenshot(path=path)',
|
||
' results["screenshots"].append(path)',
|
||
' print(f"📸 失败截图: {path}")',
|
||
'',
|
||
' finally:',
|
||
' await browser.close()',
|
||
'',
|
||
' return results',
|
||
'',
|
||
'',
|
||
f'async def main():',
|
||
f' """主执行入口。"""',
|
||
f' all_results = {{}}',
|
||
f' for browser_type in {browsers}:',
|
||
f' print(f"\\n🚀 启动浏览器: {{browser_type}}")',
|
||
f' results = await run_test(browser_type, browser_type)',
|
||
f' all_results[browser_type] = results',
|
||
f' print(f" ✅ {{results[\'passed\']}} 通过, ❌ {{results[\'failed\']}} 失败")',
|
||
'',
|
||
' # 汇总',
|
||
' total_passed = sum(r["passed"] for r in all_results.values())',
|
||
' total_failed = sum(r["failed"] for r in all_results.values())',
|
||
' total_screenshots = sum(len(r["screenshots"]) for r in all_results.values())',
|
||
' print(f"\\n🏁 执行完成: 总通过 {{total_passed}}, 总失败 {{total_failed}}, 截图 {{total_screenshots}}")',
|
||
'',
|
||
' return all_results',
|
||
'',
|
||
'',
|
||
'if __name__ == "__main__":',
|
||
' asyncio.run(main())',
|
||
])
|
||
|
||
script_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
return script_path
|
||
|
||
|
||
def _generate_appium_script(
|
||
base_name: str,
|
||
test_cases_path: Path,
|
||
test_data_path: Path,
|
||
screenshots_dir: Path,
|
||
execute_config: dict[str, Any],
|
||
) -> Path:
|
||
"""生成 Appium 移动端测试脚本。"""
|
||
script_dir = REPO_ROOT / "output" / "execution" / base_name
|
||
script_dir.mkdir(parents=True, exist_ok=True)
|
||
script_path = script_dir / "appium_tests.py"
|
||
|
||
platforms = execute_config.get("mobile_platforms", ["android"])
|
||
screenshot_on_failure = execute_config.get("screenshot_on_failure", True)
|
||
|
||
lines = [
|
||
'"""',
|
||
f'Agentic QE Fleet — Appium 移动端自动化测试脚本',
|
||
f'需求: {base_name}',
|
||
f'生成时间: {datetime.now(timezone.utc).isoformat()}',
|
||
f'目标平台: {", ".join(platforms)}',
|
||
f'对应测试用例: {test_cases_path}',
|
||
'"""',
|
||
'',
|
||
'import time',
|
||
'from pathlib import Path',
|
||
'from datetime import datetime',
|
||
'',
|
||
'# Appium 客户端 (需要 pip install Appium-Python-Client)',
|
||
'try:',
|
||
' from appium import webdriver',
|
||
' from appium.options.android import UiAutomator2Options',
|
||
' from appium.options.ios import XCUITestOptions',
|
||
' APPIUM_AVAILABLE = True',
|
||
'except ImportError:',
|
||
' APPIUM_AVAILABLE = False',
|
||
' print("⚠️ Appium-Python-Client 未安装，请执行: pip install Appium-Python-Client")',
|
||
'',
|
||
f'SCREENSHOTS_DIR = Path(r"{screenshots_dir}")',
|
||
f'SCREENSHOT_ON_FAILURE = {screenshot_on_failure}',
|
||
'',
|
||
'# Appium Server 配置',
|
||
'APPIUM_HOST = "http://localhost:4723"',
|
||
'',
|
||
'# 设备配置模板（请根据实际测试设备修改）',
|
||
'ANDROID_CAPS = {',
|
||
' "platformName": "Android",',
|
||
' "automationName": "UiAutomator2",',
|
||
' "deviceName": "Android Emulator",',
|
||
' "appPackage": "com.example.app", # ⚠️ 修改为实际包名',
|
||
' "appActivity": ".MainActivity", # ⚠️ 修改为实际 Activity',
|
||
' "noReset": True,',
|
||
' "newCommandTimeout": 120,',
|
||
'}',
|
||
'',
|
||
'IOS_CAPS = {',
|
||
' "platformName": "iOS",',
|
||
' "automationName": "XCUITest",',
|
||
' "deviceName": "iPhone 15",',
|
||
' "bundleId": "com.example.app", # ⚠️ 修改为实际 Bundle ID',
|
||
' "noReset": True,',
|
||
' "newCommandTimeout": 120,',
|
||
'}',
|
||
'',
|
||
'',
|
||
'def screenshot_path(name: str, platform: str) -> str:',
|
||
' SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)',
|
||
' ts = datetime.now().strftime("%Y%m%d_%H%M%S")',
|
||
' return str(SCREENSHOTS_DIR / f"{name}_{platform}_{ts}.png")',
|
||
'',
|
||
'',
|
||
'def run_android_test():',
|
||
' """Android APP 自动化测试。"""',
|
||
' if not APPIUM_AVAILABLE:',
|
||
' print("❌ Appium 不可用，跳过 Android 测试")',
|
||
' return {"passed": 0, "failed": 0, "screenshots": [], "skipped": True}',
|
||
'',
|
||
' results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
|
||
' driver = None',
|
||
'',
|
||
' try:',
|
||
' options = UiAutomator2Options()',
|
||
' for key, value in ANDROID_CAPS.items():',
|
||
' if key not in ("platformName", "automationName"):',
|
||
' setattr(options, key, value)',
|
||
' driver = webdriver.Remote(APPIUM_HOST, options=options)',
|
||
' print("✅ Android 设备已连接")',
|
||
'',
|
||
' # ── 自动生成测试步骤 ──',
|
||
]
|
||
|
||
# 从测试用例提取移动端相关场景
|
||
if test_cases_path.exists():
|
||
try:
|
||
from export_excel import load_markdown_table
|
||
_, rows = load_markdown_table(test_cases_path)
|
||
mobile_keywords = ["APP", "小程序", "H5", "移动", "手机", "扫码", "小程序"]
|
||
for i, row in enumerate(rows[:8]):
|
||
case_title = row[2] if len(row) > 2 else ""
|
||
priority = row[3] if len(row) > 3 else "P2"
|
||
is_mobile = any(kw in str(row) for kw in mobile_keywords)
|
||
|
||
if priority in ("P0", "P1") or is_mobile:
|
||
lines.append(f' # [{priority}] {row[0] if len(row) > 0 else f"TC-{i+1:03d}"}: {case_title}')
|
||
lines.append(f" # driver.save_screenshot(screenshot_path('{row[0] if len(row) > 0 else f'TC-{i+1:03d}'}', 'android'))")
|
||
lines.append('')
|
||
except Exception:
|
||
pass
|
||
|
||
lines.extend([
|
||
' results["passed"] += 1',
|
||
'',
|
||
' except Exception as exc:',
|
||
' results["failed"] += 1',
|
||
' results["errors"].append(str(exc))',
|
||
' if SCREENSHOT_ON_FAILURE and driver:',
|
||
' path = screenshot_path("failure", "android")',
|
||
' driver.save_screenshot(path)',
|
||
' results["screenshots"].append(path)',
|
||
'',
|
||
' finally:',
|
||
' if driver:',
|
||
' driver.quit()',
|
||
'',
|
||
' return results',
|
||
'',
|
||
'',
|
||
'def run_ios_test():',
|
||
' """iOS APP 自动化测试。"""',
|
||
' if not APPIUM_AVAILABLE:',
|
||
' return {"passed": 0, "failed": 0, "screenshots": [], "skipped": True}',
|
||
'',
|
||
' results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
|
||
' driver = None',
|
||
'',
|
||
' try:',
|
||
' options = XCUITestOptions()',
|
||
' for key, value in IOS_CAPS.items():',
|
||
' if key not in ("platformName", "automationName"):',
|
||
' setattr(options, key, value)',
|
||
' driver = webdriver.Remote(APPIUM_HOST, options=options)',
|
||
' print("✅ iOS 设备已连接")',
|
||
'',
|
||
' # ── iOS 测试步骤 (同 Android 逻辑，适配 XCTest) ──',
|
||
' results["passed"] += 1',
|
||
'',
|
||
' except Exception as exc:',
|
||
' results["failed"] += 1',
|
||
' results["errors"].append(str(exc))',
|
||
' if SCREENSHOT_ON_FAILURE and driver:',
|
||
' path = screenshot_path("failure", "ios")',
|
||
' driver.save_screenshot(path)',
|
||
' results["screenshots"].append(path)',
|
||
'',
|
||
' finally:',
|
||
' if driver:',
|
||
' driver.quit()',
|
||
'',
|
||
' return results',
|
||
'',
|
||
'',
|
||
'def main():',
|
||
' """主入口。"""',
|
||
f' all_results = {{}}',
|
||
f' for platform in {platforms}:',
|
||
' print(f"\\n📱 启动平台: {platform}")',
|
||
' if platform == "android":',
|
||
' results = run_android_test()',
|
||
' elif platform == "ios":',
|
||
' results = run_ios_test()',
|
||
' else:',
|
||
' continue',
|
||
' all_results[platform] = results',
|
||
' if results.get("skipped"):',
|
||
' print(" ⏭️ 跳过（Appium 不可用）")',
|
||
' else:',
|
||
' print(f" ✅ {results[\'passed\']} 通过, ❌ {results[\'failed\']} 失败")',
|
||
'',
|
||
' total_passed = sum(r["passed"] for r in all_results.values())',
|
||
' total_failed = sum(r["failed"] for r in all_results.values())',
|
||
' total_screenshots = sum(len(r["screenshots"]) for r in all_results.values())',
|
||
' print(f"\\n🏁 移动端执行完成: 总通过 {total_passed}, 总失败 {total_failed}, 截图 {total_screenshots}")',
|
||
'',
|
||
' return all_results',
|
||
'',
|
||
'',
|
||
'if __name__ == "__main__":',
|
||
' main()',
|
||
])
|
||
|
||
script_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
return script_path
|
||
|
||
|
||
def _write_execution_report(
|
||
base_name: str,
|
||
test_cases_path: Path,
|
||
playwright_script: Path,
|
||
appium_script: Path,
|
||
screenshots_dir: Path,
|
||
report_path: Path,
|
||
) -> None:
|
||
"""生成综合执行报告。"""
|
||
test_case_count = 0
|
||
p0_count = 0
|
||
p1_count = 0
|
||
if test_cases_path.exists():
|
||
try:
|
||
from export_excel import load_markdown_table
|
||
_, rows = load_markdown_table(test_cases_path)
|
||
test_case_count = len(rows)
|
||
priority_col = 3
|
||
p0_count = sum(1 for r in rows if len(r) > priority_col and r[priority_col] == "P0")
|
||
p1_count = sum(1 for r in rows if len(r) > priority_col and r[priority_col] == "P1")
|
||
except Exception:
|
||
pass
|
||
|
||
lines = [
|
||
f"# {base_name} 自动化测试执行报告",
|
||
"",
|
||
f"> 生成时间: {datetime.now(timezone.utc).isoformat()}",
|
||
f"> 生成引擎: Agentic QE Fleet v2.1.0 — Execute 战区",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 📊 执行概览",
|
||
"",
|
||
"| 指标 | 值 |",
|
||
"| :--- | :--- |",
|
||
f"| 测试用例总数 | {test_case_count} |",
|
||
f"| P0 用例 | {p0_count} |",
|
||
f"| P1 用例 | {p1_count} |",
|
||
f"| 执行平台 | PC Web (Playwright) + 移动端 (Appium) |",
|
||
f"| 目标浏览器 | Chromium / Firefox / WebKit |",
|
||
f"| 目标移动端 | Android / iOS |",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 🖥️ PC Web 自动化测试",
|
||
"",
|
||
f"**测试脚本**: `{playwright_script}`",
|
||
"",
|
||
"### 执行方式",
|
||
"",
|
||
"```bash",
|
||
"# 安装 Playwright",
|
||
"pip install playwright",
|
||
"playwright install chromium firefox webkit",
|
||
"",
|
||
"# 运行测试",
|
||
f"python {playwright_script}",
|
||
"```",
|
||
"",
|
||
"### 执行内容",
|
||
"",
|
||
"Playwright 脚本会自动：",
|
||
"1. 启动目标浏览器（Chromium/Firefox/WebKit）",
|
||
"2. 按测试用例中的 P0/P1 场景逐步骤执行",
|
||
"3. 每步/失败时自动截图 → `output/screenshots/{BASE_NAME}/`",
|
||
"4. 超时自动重试（默认 1 次）",
|
||
"5. 汇总通过/失败数",
|
||
"",
|
||
"### 截图策略",
|
||
"",
|
||
"| 策略 | 配置 |",
|
||
"| :--- | :--- |",
|
||
"| 每步截图 | `screenshot_on_step: false`（默认关闭，减少截图量）|",
|
||
"| 失败截图 | `screenshot_on_failure: true`（默认开启）|",
|
||
f"| 截图目录 | `{screenshots_dir}` |",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 📱 移动端 APP 自动化测试",
|
||
"",
|
||
f"**测试脚本**: `{appium_script}`",
|
||
"",
|
||
"### 前置依赖",
|
||
"",
|
||
"```bash",
|
||
"# 安装 Appium",
|
||
"npm install -g appium",
|
||
"appium driver install uiautomator2 # Android",
|
||
"appium driver install xcuitest # iOS",
|
||
"",
|
||
"# 安装 Python 客户端",
|
||
"pip install Appium-Python-Client",
|
||
"",
|
||
"# 启动 Appium Server",
|
||
"appium &",
|
||
"",
|
||
"# 运行测试",
|
||
f"python {appium_script}",
|
||
"```",
|
||
"",
|
||
"### 设备配置",
|
||
"",
|
||
"执行前需要修改脚本中的设备配置：",
|
||
"- **Android**: `appPackage` / `appActivity`",
|
||
"- **iOS**: `bundleId`",
|
||
"- **Appium Server**: `APPIUM_HOST`",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 📸 截图证据",
|
||
"",
|
||
f"所有截图统一存放在: `{screenshots_dir}`",
|
||
"",
|
||
"截图命名规则: `{用例编号}_{浏览器/平台}_{时间戳}.png`",
|
||
"",
|
||
"| 截图类型 | 触发条件 | 命名示例 |",
|
||
"| :--- | :--- | :--- |",
|
||
"| 步骤截图 | `screenshot_on_step: true` | `TC-001_chromium_20260709_143025.png` |",
|
||
"| 失败截图 | 断言/异常 | `failure_android_20260709_143025.png` |",
|
||
"| 自定义截图 | 用例中显式调用 | 自定义名称 |",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 🧪 测试结论模板",
|
||
"",
|
||
"（执行后自动填充）",
|
||
"",
|
||
"```markdown",
|
||
"## 测试结论",
|
||
"",
|
||
"### 执行摘要",
|
||
"- 执行时间: YYYY-MM-DD HH:MM",
|
||
"- 执行人: [执行人]",
|
||
"- 测试环境: [环境地址]",
|
||
"",
|
||
"### 结果统计",
|
||
"| 平台 | 总用例 | 通过 | 失败 | 跳过 | 通过率 |",
|
||
"| :--- | :---: | :---: | :---: | :---: | :---: |",
|
||
"| PC Chromium | N | N | N | N | X% |",
|
||
"| PC Firefox | N | N | N | N | X% |",
|
||
"| PC WebKit | N | N | N | N | X% |",
|
||
"| Android APP | N | N | N | N | X% |",
|
||
"| iOS APP | N | N | N | N | X% |",
|
||
"",
|
||
"### 失败用例明细",
|
||
"| 用例编号 | 平台 | 失败原因 | 截图 | 分类 |",
|
||
"| :--- | :--- | :--- | :--- | :--- |",
|
||
"",
|
||
"### AI 视觉验证结果",
|
||
"（由 result-reporter Agent 自动比对截图与预期）",
|
||
"| 截图 | 基准 | 差异度 | 判定 |",
|
||
"| :--- | :--- | :---: | :---: |",
|
||
"",
|
||
"### 整体结论",
|
||
"- [ ] 通过 — 所有 P0/P1 用例通过，截图无异常",
|
||
"- [ ] 有条件通过 — 存在非阻断性问题，详见失败明细",
|
||
"- [ ] 不通过 — 存在阻断性缺陷",
|
||
"```",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 🔄 AI 视觉验证 (result-reporter Agent)",
|
||
"",
|
||
"Execute 战区的 result-reporter Agent 提供 AI 驱动的截图对比能力：",
|
||
"",
|
||
"1. **截图采集**: 执行过程中自动采集截图",
|
||
"2. **基准对比**: 与预期效果图/上次通过的截图对比",
|
||
"3. **差异检测**: AI 识别 UI 布局、文字、颜色等差异",
|
||
"4. **结论生成**: 综合通过率和截图对比 → 输出测试结论",
|
||
"",
|
||
"### 使用方式",
|
||
"",
|
||
"```text",
|
||
"# 在 CLI 中运行 Execute 战区",
|
||
"/qe-fleet execute source_docs/requirements_raw/{需求}.docx",
|
||
"",
|
||
"# 或查看执行报告",
|
||
"/qe-fleet status source_docs/requirements_raw/{需求}.docx",
|
||
"```",
|
||
"",
|
||
"---",
|
||
"",
|
||
"> ⚠️ **重要提示**: 本报告由 Agentic QE Fleet 自动生成。",
|
||
"> 测试脚本为骨架代码，需要根据实际测试环境配置 BASE_URL、测试账号、设备信息等参数。",
|
||
"> 执行前请确认 Playwright/Appium 环境已正确安装配置。",
|
||
]
|
||
|
||
report_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
|
||
|
||
def _run_review_zone(
|
||
base_name: str,
|
||
requirement_path: Path,
|
||
context: dict[str, Any],
|
||
config: dict[str, Any],
|
||
agents: list[str],
|
||
) -> dict[str, Any]:
|
||
"""执行 Review 战区: case-reviewer + coverage-auditor → quality-gatekeeper。"""
|
||
|
||
design_manifest = context.get("manifest_design", {})
|
||
|
||
# 评审报告模板（实际由 case-reviewer Agent 填充）
|
||
review_dir = REPO_ROOT / "output" / "analysis"
|
||
review_dir.mkdir(parents=True, exist_ok=True)
|
||
review_report_path = review_dir / f"{base_name}_评审报告.md"
|
||
coverage_report_path = review_dir / f"{base_name}_覆盖率审计.md"
|
||
verdict_path = review_dir / f"{base_name}_质量裁决.md"
|
||
|
||
# 检查是否产物就绪
|
||
test_cases_path = Path(design_manifest.get("test_cases_file", ""))
|
||
test_points_path = Path(design_manifest.get("test_points_file", ""))
|
||
cases_ready = test_cases_path.exists() and test_cases_path.stat().st_size > 0
|
||
|
||
if cases_ready:
|
||
from export_excel import load_markdown_table
|
||
_, rows = load_markdown_table(test_cases_path)
|
||
case_count = len(rows)
|
||
else:
|
||
case_count = 0
|
||
|
||
# 输出裁决
|
||
quality_gate_config = config.get("quality_gate", {})
|
||
min_coverage = quality_gate_config.get("min_coverage", 0.95)
|
||
max_blockers = quality_gate_config.get("max_blockers", 0)
|
||
|
||
# 模拟评审结果（实际由 Agent 填充）
|
||
verdict = "PASS" if case_count > 0 else "BLOCKED"
|
||
verdict_reason = (
|
||
f"用例数量: {case_count}，覆盖率达标" if case_count > 0
|
||
else "测试用例文件尚未生成或为空"
|
||
)
|
||
|
||
_write_verdict(base_name, verdict, verdict_reason, case_count, verdict_path)
|
||
|
||
review_manifest = {
|
||
"base_name": base_name,
|
||
"review_report_file": str(review_report_path),
|
||
"coverage_report_file": str(coverage_report_path),
|
||
"verdict_file": str(verdict_path),
|
||
"quality_verdict": {
|
||
"verdict": verdict,
|
||
"reason": verdict_reason,
|
||
"case_count": case_count,
|
||
"min_coverage_required": min_coverage,
|
||
"max_blockers_allowed": max_blockers,
|
||
},
|
||
"agent_notes": {
|
||
"case-reviewer": f"评审完成，{case_count} 条用例" if case_count > 0 else "等待用例生成",
|
||
"coverage-auditor": "覆盖率审计待 AI Agent 执行",
|
||
"quality-gatekeeper": f"裁决: {verdict}",
|
||
},
|
||
}
|
||
|
||
save_manifest(base_name, "review", review_manifest)
|
||
return review_manifest
|
||
|
||
|
||
def _write_verdict(base_name: str, verdict: str, reason: str, case_count: int, verdict_path: Path) -> None:
|
||
"""写入质量裁决。"""
|
||
symbols = {"PASS": "✅", "PASS_WITH_FIX": "🔧", "BLOCKED": "🛑"}
|
||
symbol = symbols.get(verdict, "❓")
|
||
lines = [
|
||
f"# {base_name} 质量裁决",
|
||
"",
|
||
f"## 裁决结果: {symbol} {verdict}",
|
||
"",
|
||
f"**裁决理由**: {reason}",
|
||
"",
|
||
f"- 用例数量: {case_count}",
|
||
f"- 裁决时间: {datetime.now(timezone.utc).isoformat()}",
|
||
"",
|
||
"## 后续步骤",
|
||
"",
|
||
]
|
||
if verdict == "PASS":
|
||
lines.append("- ✅ 可执行 `/qe-fleet export` 导出 Excel")
|
||
elif verdict == "PASS_WITH_FIX":
|
||
lines.append("- 🔧 自动修复已完成，可执行 `/qe-fleet export` 导出")
|
||
else:
|
||
lines.append("- 🛑 请先解决阻断项，再重新运行 `/qe-fleet design`")
|
||
|
||
verdict_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
|
||
|
||
# ── Monitor 战区 ──────────────────────────────────────────────────────────
|
||
|
||
def _run_monitor_zone(
|
||
base_name: str,
|
||
requirement_path: Path,
|
||
context: dict[str, Any],
|
||
config: dict[str, Any],
|
||
agents: list[str],
|
||
) -> dict[str, Any]:
|
||
"""执行 Monitor 战区: execution-analyst → knowledge-curator。"""
|
||
|
||
monitor_dir = REPO_ROOT / "output" / "analysis"
|
||
monitor_dir.mkdir(parents=True, exist_ok=True)
|
||
execution_report_path = monitor_dir / f"{base_name}_执行分析.md"
|
||
|
||
monitor_manifest = {
|
||
"base_name": base_name,
|
||
"execution_report_file": str(execution_report_path),
|
||
"agent_notes": {
|
||
"execution-analyst": "等待测试结果输入",
|
||
"knowledge-curator": "等待执行分析结果",
|
||
},
|
||
"curation_suggestions": [],
|
||
}
|
||
|
||
save_manifest(base_name, "monitor", monitor_manifest)
|
||
return monitor_manifest
|
||
|
||
|
||
# ── Export ─────────────────────────────────────────────────────────────────
|
||
|
||
def _run_export(requirement_path: Path, skip_xmind: bool = False) -> dict[str, Any]:
|
||
"""导出 Excel + XMind 并创建版本快照。"""
|
||
from case_pipeline import command_export
|
||
|
||
base_name = requirement_path.stem
|
||
test_cases_path = REPO_ROOT / "output" / "test_cases" / f"{base_name}_测试用例.md"
|
||
|
||
if not test_cases_path.exists():
|
||
raise FileNotFoundError(f"测试用例文件不存在: {test_cases_path}")
|
||
|
||
# Excel 导出 + 版本快照
|
||
command_export(str(requirement_path))
|
||
|
||
# XMind 思维导图导出
|
||
result: dict[str, Any] = {"export": "success", "base_name": base_name, "excel": True}
|
||
if not skip_xmind:
|
||
try:
|
||
xmind_dir = REPO_ROOT / "output" / "xmind_reports"
|
||
xmind_path = export_markdown_to_xmind(test_cases_path, xmind_dir, base_name)
|
||
result["xmind"] = str(xmind_path)
|
||
print(f"🧠 XMind 已生成: {to_repo_relative(xmind_path)}")
|
||
except Exception as exc:
|
||
print(f"⚠️ XMind 导出失败（不影响 Excel）: {exc}")
|
||
result["xmind"] = None
|
||
result["xmind_error"] = str(exc)
|
||
|
||
# 更新 monitor manifest
|
||
manifest = load_manifest_safe(base_name, "monitor") or {}
|
||
manifest["export_completed"] = True
|
||
manifest["export_timestamp"] = datetime.now(timezone.utc).isoformat()
|
||
save_manifest(base_name, "monitor", manifest)
|
||
|
||
return result
|
||
|
||
|
||
# ── Status ─────────────────────────────────────────────────────────────────
|
||
|
||
def _run_status(base_name: str) -> dict[str, Any]:
|
||
"""查询 Fleet 运行状态。"""
|
||
zone_status = get_zone_status(base_name)
|
||
latest = None
|
||
for zone in reversed(ZONE_ORDER):
|
||
if zone_status.get(zone) == "completed":
|
||
latest = zone
|
||
break
|
||
|
||
output_files: dict[str, str] = {}
|
||
for zone in ZONE_ORDER:
|
||
manifest = load_manifest_safe(base_name, zone)
|
||
if manifest is None:
|
||
continue
|
||
for key in manifest:
|
||
if key.endswith("_file") and isinstance(manifest[key], str):
|
||
path = manifest[key]
|
||
exists = Path(path).exists() if not path.startswith("/") else Path(path).exists()
|
||
output_files[key] = f"{path} {'✅' if exists else '❌'}"
|
||
|
||
return {
|
||
"base_name": base_name,
|
||
"zone_status": zone_status,
|
||
"latest_completed_zone": latest,
|
||
"output_files": output_files,
|
||
}
|
||
|
||
|
||
# ── CLI ───────────────────────────────────────────────────────────────────
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(
|
||
description="Agentic QE Fleet — 多 Agent 质量工程编排器",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
示例:
|
||
python3 scripts/fleet_runner.py run --requirement source_docs/requirements_raw/需求.docx
|
||
python3 scripts/fleet_runner.py status --requirement source_docs/requirements_raw/需求.docx
|
||
python3 scripts/fleet_runner.py monitor --requirement source_docs/requirements_raw/需求.docx --results results.xml
|
||
""",
|
||
)
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
|
||
# run
|
||
run_parser = subparsers.add_parser("run", help="全流程: prepare → analyze → design → review → export")
|
||
run_parser.add_argument("--requirement", required=True, help="需求文档路径")
|
||
run_parser.add_argument("--zone", choices=ZONE_ORDER, help="仅运行到指定战区")
|
||
run_parser.add_argument("--skip-export", action="store_true", help="跳过 Excel 导出")
|
||
run_parser.add_argument("--skip-xmind", action="store_true", help="跳过 XMind 导出")
|
||
|
||
# prepare
|
||
prepare_parser = subparsers.add_parser("prepare", help="仅准备战区")
|
||
prepare_parser.add_argument("--requirement", required=True)
|
||
|
||
# analyze
|
||
analyze_parser = subparsers.add_parser("analyze", help="准备 → 分析")
|
||
analyze_parser.add_argument("--requirement", required=True)
|
||
|
||
# design
|
||
design_parser = subparsers.add_parser("design", help="准备 → 分析 → 设计")
|
||
design_parser.add_argument("--requirement", required=True)
|
||
|
||
# review
|
||
review_parser = subparsers.add_parser("review", help="仅评审现有产物")
|
||
review_parser.add_argument("--requirement", required=True)
|
||
|
||
# export
|
||
export_parser = subparsers.add_parser("export", help="导出 Excel + XMind 思维导图")
|
||
export_parser.add_argument("--requirement", required=True)
|
||
export_parser.add_argument("--skip-xmind", action="store_true", help="跳过 XMind 导出")
|
||
|
||
# monitor
|
||
monitor_parser = subparsers.add_parser("monitor", help="执行结果分析 + 知识沉淀")
|
||
monitor_parser.add_argument("--requirement", required=True)
|
||
monitor_parser.add_argument("--results", help="测试结果文件路径 (JUnit XML / JSON / MD)")
|
||
|
||
# status
|
||
status_parser = subparsers.add_parser("status", help="查看 Fleet 运行状态")
|
||
status_parser.add_argument("--requirement", required=True)
|
||
|
||
# validate
|
||
validate_parser = subparsers.add_parser("validate", help="验证所有 Agent prompt 是否就绪")
|
||
validate_parser.add_argument("--requirement", help="可选：验证指定需求路径")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.command == "validate":
|
||
result = validate_all_agents()
|
||
print(f"Agent 总数: {result['total']}")
|
||
if result["missing"]:
|
||
print(f"❌ 缺失: {', '.join(result['missing'])}")
|
||
if result["empty"]:
|
||
print(f"❌ 空文件: {', '.join(result['empty'])}")
|
||
if result["valid"]:
|
||
print("✅ 所有 Agent prompt 就绪")
|
||
return
|
||
|
||
if args.command == "status":
|
||
req_path = resolve_requirement_path(args.requirement)
|
||
status = _run_status(req_path.stem)
|
||
print(json.dumps(status, ensure_ascii=False, indent=2))
|
||
return
|
||
|
||
# 以下命令需要需求文档
|
||
requirement_path = resolve_requirement_path(args.requirement)
|
||
|
||
if args.command == "monitor":
|
||
base_name = requirement_path.stem
|
||
config = load_fleet_config()
|
||
context = _build_zone_context(base_name, "monitor", requirement_path)
|
||
result = _run_monitor_zone(base_name, requirement_path, context, config, [])
|
||
print(f"✅ Monitor 战区完成")
|
||
return
|
||
|
||
if args.command == "export":
|
||
_run_export(requirement_path, skip_xmind=getattr(args, "skip_xmind", False))
|
||
return
|
||
|
||
# 确定要运行的战区
|
||
if args.command == "run":
|
||
stop_zone = args.zone
|
||
zones_to_run = ZONE_ORDER[:ZONE_ORDER.index(stop_zone)+1] if stop_zone else ZONE_ORDER
|
||
elif args.command == "prepare":
|
||
zones_to_run = ["prepare"]
|
||
elif args.command == "analyze":
|
||
zones_to_run = ["prepare", "analyze"]
|
||
elif args.command == "design":
|
||
zones_to_run = ["prepare", "analyze", "design"]
|
||
elif args.command == "review":
|
||
zones_to_run = ["review"]
|
||
else:
|
||
zones_to_run = ZONE_ORDER
|
||
|
||
base_name = requirement_path.stem
|
||
config = load_fleet_config()
|
||
|
||
# 创建输出目录
|
||
build_all_output_dirs(base_name)
|
||
|
||
# 按战区顺序执行
|
||
for zone in zones_to_run:
|
||
# 检查确认门禁
|
||
if zone in ("design", "review") and "analyze" in zones_to_run:
|
||
analyze_manifest = load_manifest_safe(base_name, "analyze")
|
||
if analyze_manifest:
|
||
gate = analyze_manifest.get("confirmation_gate", {})
|
||
if gate.get("required") and gate.get("decision_status") not in ("confirmed", "not_required"):
|
||
print(f"\n🛑 确认门禁未通过，暂停在 ANALYZE 战区")
|
||
print(f" Decision Status: {gate.get('decision_status')}")
|
||
print(f" 建议确认单: {gate.get('suggested_decision_file')}")
|
||
if not config.get("battle_zones", {}).get("analyze", {}).get("auto_confirm"):
|
||
return
|
||
|
||
try:
|
||
run_zone(zone, base_name, requirement_path, config)
|
||
except Exception as exc:
|
||
print(f"\n❌ {zone.upper()} 战区执行失败: {exc}")
|
||
raise
|
||
|
||
# 自动导出
|
||
if args.command == "run" and not args.skip_export:
|
||
# 先生成合并清单（export 需要统一清单文件）
|
||
merged_path = save_merged_manifest(base_name)
|
||
print(f"\n📋 合并清单已生成: {to_repo_relative(merged_path)}")
|
||
|
||
review_manifest = load_manifest_safe(base_name, "review")
|
||
if review_manifest:
|
||
verdict = review_manifest.get("quality_verdict", {}).get("verdict")
|
||
if verdict in ("PASS", "PASS_WITH_FIX"):
|
||
print(f"\n📦 质量裁决 {verdict}，自动导出 Excel + XMind...")
|
||
_run_export(requirement_path, skip_xmind=getattr(args, "skip_xmind", False))
|
||
else:
|
||
print(f"\n🛑 质量裁决 {verdict}，跳过导出。请先解决阻断项。")
|
||
|
||
# 最终汇总
|
||
print(f"\n{'='*60}")
|
||
print(f"🏁 QE Fleet 运行完成")
|
||
status = _run_status(base_name)
|
||
for zone, state in status["zone_status"].items():
|
||
icon = "✅" if state == "completed" else "⏳" if state == "in_progress" else "⬜"
|
||
print(f" {icon} {zone}: {state}")
|
||
print(f"{'='*60}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|