Yb-QaAutomationHub/scripts/governance_audit.py

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())