xst/QaAutomationHub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 安徽运八需求全流水线输出同步 + 知识库/agents/资源文件更新

Browse Source
This commit is contained in:
xst
2026-07-14 15:50:17 +08:00
parent e34a3c64ce
commit 2bc036c867
79 changed files with 15673 additions and 2722 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/fleet_runner.py
+242 -30
View File
@@ -179,7 +179,7 @@ def build_all_output_dirs(base_name: str) -> None:
# ── 编排核心 ──────────────────────────────────────────────────────────────
def run_zone(zone: str, base_name: str, requirement_path: Path, config: dict[str, Any]) -> dict[str, Any]:
def run_zone(zone: str, base_name: str, requirement_path: Path, config: dict[str, Any], scaffold_only: bool = False) -> dict[str, Any]:
"""运行一个战区，返回该战区的 manifest 数据。"""
zone_config = config.get("battle_zones", {}).get(zone, {})
@@ -214,7 +214,7 @@ def run_zone(zone: str, base_name: str, requirement_path: Path, config: dict[str
if handler is None:
raise ValueError(f"未知战区: {zone}")
result = handler(base_name, requirement_path, context, config, agents)
result = handler(base_name, requirement_path, context, config, agents, scaffold_only=scaffold_only)
# 标记完成
if result:
@@ -284,6 +284,7 @@ def _run_prepare_zone(
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
scaffold_only: bool = False,
) -> dict[str, Any]:
"""执行 Prepare 战区: document-parser → knowledge-activator。"""
@@ -315,6 +316,20 @@ def _run_prepare_zone(
requirement_body = safe_read_text(normalized_requirement_file)
confidence = _estimate_document_confidence(requirement_body, requirement_path.suffix.lower())
# 附加文件检测：从需求文本中提取引用的文件路径和URL
attached_source_files = _detect_attached_files(requirement_body)
# 构建多源注册表
sources_registry = _build_sources_registry(
requirement_path=requirement_path,
technical_solution_files=technical_solution_files,
attached_files=attached_source_files,
)
if attached_source_files:
print(f"📎 检测到 {len(attached_source_files)} 个附加源文件")
for af in attached_source_files:
print(f" - {af['path']} ({af.get('exists', 'unknown')})")
# 2. 知识激活
activated_knowledge = _activate_knowledge(requirement_path, config)
@@ -334,6 +349,8 @@ def _run_prepare_zone(
"document_confidence": confidence,
"activated_knowledge": activated_knowledge,
"knowledge_gaps": knowledge_gaps,
"attached_source_files": attached_source_files,
"sources_registry": sources_registry,
"agent_notes": {
"document-parser": f"解析完成，置信度 {confidence.get('requirement', 0):.0%}",
"knowledge-activator": f"激活 {len(activated_knowledge.get('terminology', {}).get('permanent', []))} 常驻 + "
@@ -447,6 +464,130 @@ def _detect_knowledge_gaps(activated_knowledge: dict[str, Any], requirement_text
return gaps
def _detect_attached_files(requirement_text: str) -> list[dict[str, str]]:
"""从需求文本中提取引用的文件路径和URL。
支持三种模式:
1. Windows 绝对路径 (E:\\Downloads\\xxx.pdf)
2. URL (https://www.showdoc.com.cn/...)
3. 相对项目路径 (output/prototype/xxx.md)
"""
import re
attached: list[dict[str, str]] = []
# Pattern 1: Windows absolute paths with Chinese support
win_path_pattern = re.compile(
r'([A-Za-z]:[\\/](?:[^\\/:*?"<>|\r\n]+[\\/])*[^\\/:*?"<>|\r\n]+\.(?:pdf|docx?|xlsx?|html?|txt|md|json|xml|csv))',
re.IGNORECASE,
)
for match in win_path_pattern.finditer(requirement_text):
path = match.group(1)
line_start = requirement_text.rfind('\n', 0, match.start()) + 1
line_end = requirement_text.find('\n', match.end())
context_line = requirement_text[line_start:line_end if line_end != -1 else len(requirement_text)]
attached.append({
"path": path,
"type": "local_file",
"source": "embedded_path",
"exists": str(Path(path).exists()).lower() if path else "false",
"context_line": context_line.strip()[:200],
})
# Pattern 2: Documentation URLs (Showdoc, Confluence, etc.)
url_pattern = re.compile(r'(https?://[^\s\n\r一-鿿]+)')
doc_domains = ['showdoc', 'confluence', 'wiki', 'yuque', 'notion', 'figma', 'lanhu', 'axure', 'modao']
for match in url_pattern.finditer(requirement_text):
url = match.group(1).rstrip('.,;:;）)')
if any(domain in url.lower() for domain in doc_domains):
attached.append({
"path": url,
"type": "url",
"source": "embedded_url",
"exists": "unknown",
"context_line": "",
})
# Pattern 3: Project-relative paths with context keywords
rel_pattern = re.compile(
r'(?:(?:原型文件|接口文档|技术方案|设计稿|原型|文档|文件|路径)[\s：:]*)?'
r'([a-zA-Z0-9_\-\.]+/[a-zA-Z0-9_\-\./]+\.(?:html?|pdf|docx?|md|json))',
re.IGNORECASE,
)
for match in rel_pattern.finditer(requirement_text):
rel_path = match.group(1)
if '/' not in rel_path:
continue
abs_path = REPO_ROOT / rel_path
attached.append({
"path": str(abs_path),
"type": "project_file",
"source": "embedded_relative_path",
"exists": str(abs_path.exists()).lower(),
"context_line": "",
})
# Deduplicate by path
seen: set[str] = set()
unique: list[dict[str, str]] = []
for item in attached:
if item["path"] not in seen:
seen.add(item["path"])
unique.append(item)
return unique
def _build_sources_registry(
requirement_path: Path,
technical_solution_files: list[Path],
attached_files: list[dict[str, str]],
) -> list[dict[str, str]]:
"""构建多源注册表，跟踪所有权威数据源及其角色。"""
registry: list[dict[str, str]] = [
{
"path": str(requirement_path),
"type": "primary_requirement",
"role": "business_background",
"format": requirement_path.suffix.lower().lstrip("."),
"authority": "primary",
}
]
for tf in technical_solution_files:
registry.append({
"path": str(tf),
"type": "technical_solution",
"role": "technical_constraint",
"format": tf.suffix.lower().lstrip("."),
"authority": "technical_reference",
})
for af in attached_files:
path = af.get("path", "")
file_type = af.get("type", "")
registry.append({
"path": path,
"type": f"attached_{file_type}",
"role": _infer_source_role(af),
"format": Path(path).suffix.lower().lstrip(".") if file_type != "url" else "url",
"authority": "reference",
"exists": af.get("exists", "unknown"),
})
return registry
def _infer_source_role(attached_file: dict[str, str]) -> str:
"""推断附加源文件的角色。"""
path_lower = attached_file.get("path", "").lower()
if any(kw in path_lower for kw in ["接口", "api", "接口文档"]):
return "api_spec"
if any(kw in path_lower for kw in ["原型", "prototype", "html"]):
return "ui_prototype"
if any(kw in path_lower for kw in ["showdoc", "confluence", "wiki"]):
return "documentation"
if any(kw in path_lower for kw in ["技术方案", "technical", "设计"]):
return "technical_design"
return "reference"
# ── Analyze 战区 ──────────────────────────────────────────────────────────
def _run_analyze_zone(
@@ -455,6 +596,7 @@ def _run_analyze_zone(
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
scaffold_only: bool = False,
) -> dict[str, Any]:
"""执行 Analyze 战区: requirement-analyzer + conflict-detector → risk-assessor。"""
@@ -640,6 +782,7 @@ def _run_design_zone(
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
scaffold_only: bool = False,
) -> dict[str, Any]:
"""执行 Design 战区: strategist → (testpoint-designer + data-builder) → case-designer。"""
@@ -672,13 +815,36 @@ def _run_design_zone(
"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": {
}
if scaffold_only:
design_manifest["agent_notes"] = {
"test-strategist": {"status": "completed", "message": f"策略已生成，{p0_count} 个 P0 风险需 100% 覆盖"},
"testpoint-designer": {
"status": "pending_ai",
"message": "待 AI Agent 生成测试点",
"prompt_file": "agents/design/testpoint_designer.md",
"output_file": str(test_points_path),
},
"case-designer": {
"status": "pending_ai",
"message": "待 AI Agent 生成用例",
"prompt_file": "agents/design/case_designer.md",
"output_file": str(test_cases_path),
},
"data-builder": {
"status": "pending_ai",
"message": "待 AI Agent 完善测试数据",
"prompt_file": "agents/design/data_builder.md",
"output_file": str(data_path),
},
}
else:
design_manifest["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
@@ -773,6 +939,7 @@ def _run_execute_zone(
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
scaffold_only: bool = False,
) -> dict[str, Any]:
"""执行 Execute 战区: web-executor + mobile-executor → result-reporter。"""
@@ -1376,6 +1543,7 @@ def _run_review_zone(
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
scaffold_only: bool = False,
) -> dict[str, Any]:
"""执行 Review 战区: case-reviewer + coverage-auditor → quality-gatekeeper。"""
@@ -1406,32 +1574,71 @@ def _run_review_zone(
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 "测试用例文件尚未生成或为空"
)
if scaffold_only:
verdict = "PENDING_AI"
verdict_reason = "等待 AI Agent 评审（review 战区: case-reviewer → coverage-auditor → quality-gatekeeper）"
elif case_count > 0:
verdict = "PASS"
verdict_reason = f"用例数量: {case_count}，覆盖率达标"
else:
verdict = "BLOCKED"
verdict_reason = "测试用例文件尚未生成或为空"
_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}",
},
}
if scaffold_only:
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": {
"status": "pending_ai",
"message": "待 AI Agent 评审用例",
"prompt_file": "agents/review/case_reviewer.md",
"output_file": str(review_report_path),
},
"coverage-auditor": {
"status": "pending_ai",
"message": "待 AI Agent 审计覆盖率",
"prompt_file": "agents/review/coverage_auditor.md",
"output_file": str(coverage_report_path),
},
"quality-gatekeeper": {
"status": "pending_ai",
"message": "待 AI Agent 质量裁决",
"prompt_file": "agents/review/quality_gatekeeper.md",
"output_file": str(verdict_path),
},
},
}
else:
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
@@ -1472,6 +1679,7 @@ def _run_monitor_zone(
context: dict[str, Any],
config: dict[str, Any],
agents: list[str],
scaffold_only: bool = False,
) -> dict[str, Any]:
"""执行 Monitor 战区: execution-analyst → knowledge-curator。"""
@@ -1581,6 +1789,7 @@ def main() -> None:
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 导出")
run_parser.add_argument("--scaffold-only", action="store_true", help="仅生成脚手架（模板+清单+脚本），AI内容由SKILL.md调用Agent生成")
# prepare
prepare_parser = subparsers.add_parser("prepare", help="仅准备战区")
@@ -1672,6 +1881,7 @@ def main() -> None:
build_all_output_dirs(base_name)
# 按战区顺序执行
scaffold_only = getattr(args, 'scaffold_only', False)
for zone in zones_to_run:
# 检查确认门禁
if zone in ("design", "review") and "analyze" in zones_to_run:
@@ -1686,7 +1896,7 @@ def main() -> None:
return
try:
run_zone(zone, base_name, requirement_path, config)
run_zone(zone, base_name, requirement_path, config, scaffold_only=scaffold_only)
except Exception as exc:
print(f"\n❌ {zone.upper()} 战区执行失败: {exc}")
raise
@@ -1700,9 +1910,11 @@ def main() -> None:
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"):
if verdict in ("PASS", "PASS_WITH_FIX") and not scaffold_only:
print(f"\n📦 质量裁决 {verdict}，自动导出 Excel + XMind...")
_run_export(requirement_path, skip_xmind=getattr(args, "skip_xmind", False))
elif verdict == "PENDING_AI":
print(f"\n⏳ 质量裁决 {verdict}，AI 内容生成未完成。请通过 SKILL.md 调用 AI Agent。")
else:
print(f"\n🛑 质量裁决 {verdict}，跳过导出。请先解决阻断项。")