import argparse import subprocess import sys from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent WORKFLOW_TRIGGER_PATHS = { "AGENTS.md", "README.md", "操作手册.md", "scripts/case_pipeline.py", "scripts/export_excel.py", "scripts/governance_audit.py", "knowledge_base/01_standards/test_case_template.md", "knowledge_base/01_standards/definition_of_done.md", "knowledge_base/01_standards/review_checklist.md", "knowledge_base/01_standards/terminology.md", "knowledge_base/01_standards/terminology_optional_saas.md", } WORKFLOW_TRIGGER_PREFIXES = ( ".claude/", "agents/", "scripts/", "knowledge_base/01_standards/", ) STRUCTURE_TRIGGER_PREFIXES = ( ".claude/", "agents/", "scripts/", "knowledge_base/01_standards/", ) IGNORED_PREFIXES = ( "output/", "requirements/", "knowledge_base/02_history/", "knowledge_base/03_best_practices/", ) CONSISTENCY_EXPECTATIONS = { "AGENTS.md": [ "fleet_runner.py", "case_pipeline.py", "output/manifests/{BASE_NAME}.json", "knowledge_base/00_project/project_profile.md", "knowledge_base/01_standards/review_checklist.md", "normalized_requirement_file", ], ".claude/commands/case_generate.md": [ "fleet_runner.py", "qe-fleet run", "case_generate", "output/manifests/{BASE_NAME}.json", ], ".claude/skills/case_generate/SKILL.md": [ "fleet_runner.py", "qe-fleet run", "case_generate", "output/manifests/{BASE_NAME}.json", ], ".claude/instructions.md": [ "scripts/case_pipeline.py", "effective_terminology_files", "knowledge_base/00_project/project_profile.md", "knowledge_base/01_standards/review_checklist.md", "normalized_requirement_file", ], "README.md": [ "output/manifests/{BASE_NAME}.json", "术语文件按需求内容自动识别生效范围", "knowledge_base/00_project/project_profile.md", "review_checklist.md", "source_docs/requirements_raw", "output/versions/{BASE_NAME}/v1/", "VERSION_INDEX.md", "最近 `3` 个版本", ], "操作手册.md": [ "terminology_optional_saas.md", "自动识别", "project_profile.md", "review_checklist.md", "source_docs/", "output/versions/{BASE_NAME}/v1/", "snapshot_meta.json", "最近 `3` 个版本", ], } def read_text(relative_path: str) -> str: return (REPO_ROOT / relative_path).read_text(encoding="utf-8") def git_status_paths() -> list[tuple[str, str]]: try: result = subprocess.run( ["git", "status", "--porcelain"], cwd=REPO_ROOT, check=True, capture_output=True, text=True, ) except subprocess.CalledProcessError as exc: raise RuntimeError(exc.stderr.strip() or "无法读取 git status。") from exc items: list[tuple[str, str]] = [] for raw_line in result.stdout.splitlines(): if not raw_line: continue status = raw_line[:2] payload = raw_line[3:] path = payload.split(" -> ")[-1].strip() if path: items.append((status, path)) return items def matches_any_prefix(path: str, prefixes: tuple[str, ...]) -> bool: return any(path.startswith(prefix) for prefix in prefixes) def should_ignore(path: str) -> bool: return matches_any_prefix(path, IGNORED_PREFIXES) def is_workflow_trigger(path: str) -> bool: return path in WORKFLOW_TRIGGER_PATHS or matches_any_prefix(path, WORKFLOW_TRIGGER_PREFIXES) def is_structure_trigger(path: str, status: str) -> bool: if "?" in status or "A" in status or "D" in status or "R" in status: return matches_any_prefix(path, STRUCTURE_TRIGGER_PREFIXES) return False def classify_changed_files(items: list[tuple[str, str]]) -> dict[str, list[str]]: workflow_hits: list[str] = [] structure_hits: list[str] = [] ignored_hits: list[str] = [] for status, path in items: if should_ignore(path): ignored_hits.append(path) continue if is_workflow_trigger(path): workflow_hits.append(path) if is_structure_trigger(path, status): structure_hits.append(path) return { "workflow_hits": sorted(set(workflow_hits)), "structure_hits": sorted(set(structure_hits)), "ignored_hits": sorted(set(ignored_hits)), } def audit_consistency() -> list[str]: issues: list[str] = [] for relative_path, snippets in CONSISTENCY_EXPECTATIONS.items(): content = read_text(relative_path) for snippet in snippets: if snippet not in content: issues.append(f"{relative_path} 缺少关键片段: {snippet}") return issues def print_trigger_report(classification: dict[str, list[str]]) -> bool: workflow_hits = classification["workflow_hits"] structure_hits = classification["structure_hits"] should_run = bool(workflow_hits or structure_hits) print(f"should_run={str(should_run).lower()}") if workflow_hits: print("workflow_hits:") for path in workflow_hits: print(f" - {path}") if structure_hits: print("structure_hits:") for path in structure_hits: print(f" - {path}") return should_run def command_should_run() -> int: classification = classify_changed_files(git_status_paths()) should_run = print_trigger_report(classification) return 0 if should_run else 1 def command_audit() -> int: issues = audit_consistency() if issues: print("audit_status=fail") for issue in issues: print(f"- {issue}") return 1 print("audit_status=pass") return 0 def command_auto() -> int: classification = classify_changed_files(git_status_paths()) should_run = print_trigger_report(classification) if not should_run: print("audit_skipped=true") return 0 issues = audit_consistency() if issues: print("audit_status=fail") for issue in issues: print(f"- {issue}") return 1 print("audit_status=pass") return 0 def main() -> int: parser = argparse.ArgumentParser(description="流程一致性与文档滞后审计。") subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("should-run", help="根据当前 git 变更判断是否需要触发治理审计。") subparsers.add_parser("audit", help="执行一致性与文档滞后检查。") subparsers.add_parser("auto", help="自动判断是否触发，并在需要时执行审计。") args = parser.parse_args() if args.command == "should-run": return command_should_run() if args.command == "audit": return command_audit() if args.command == "auto": return command_auto() return 1 if __name__ == "__main__": sys.exit(main())