272 lines
9.7 KiB
Python
272 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
||
"""AIOS V2 文档—代码匹配门禁。
|
||
|
||
仅使用 Python 标准库，可由本地、Gitea Actions、Codex 监督任务或发布验收脚本调用。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CheckResult:
|
||
level: str
|
||
message: str
|
||
|
||
|
||
REQUIRED_DOCUMENTS = (
|
||
"docs/aios-v2/README.md",
|
||
"docs/aios-v2/01-USER-REQUIREMENTS.md",
|
||
"docs/aios-v2/02-TARGET-ARCHITECTURE.md",
|
||
"docs/aios-v2/03-MIGRATION-AND-DELIVERY.md",
|
||
"docs/aios-v2/04-MODULE-CONTRACTS.md",
|
||
"docs/aios-v2/05-TRACEABILITY-MATRIX.md",
|
||
"docs/aios-v2/06-DOCUMENT-GOVERNANCE.md",
|
||
"docs/aios-v2/adr/README.md",
|
||
"docs/aios-v2/adr/ADR-001-single-control-plane.md",
|
||
"docs/aios-v2/adr/ADR-002-openclaw-hermes-decoupling.md",
|
||
"docs/aios-v2/adr/ADR-003-gitops-only-change-flow.md",
|
||
"docs/aios-v2/adr/ADR-004-multi-tenant-isolation.md",
|
||
"docs/aios-v2/adr/ADR-005-document-code-sync-gate.md",
|
||
"docs/aios-v2/adr/ADR-006-candidate-persistence-strategy.md",
|
||
)
|
||
|
||
REQUIRED_CODE_PATHS = (
|
||
"active-runtime/app/aios_v2/control_plane.py",
|
||
"active-runtime/app/aios_v2/persistence.py",
|
||
"active-runtime/app/aios_v2/gateway.py",
|
||
"active-runtime/app/aios_v2/app.py",
|
||
"active-runtime/app/aios_v2/orchestrator.py",
|
||
"active-runtime/app/aios_v2/execution_tracking.py",
|
||
"active-runtime/app/aios_v2/gitops_status.py",
|
||
"active-runtime/app/aios_v2/console.py",
|
||
"active-runtime/app/aios_openclaw/contract.py",
|
||
"active-runtime/app/aios_openclaw/adapter.py",
|
||
"active-runtime/app/aios_openclaw/transport.py",
|
||
"active-runtime/app/aios_openclaw/health.py",
|
||
"active-runtime/app/aios_openclaw/callback.py",
|
||
"active-runtime/app/aios_hermes/contract.py",
|
||
"active-runtime/app/aios_hermes/adapter.py",
|
||
"active-runtime/app/aios_hermes/worker.py",
|
||
"tests/aios_v2_candidate/test_candidate_flow.py",
|
||
"deploy/aios-v2/systemd/aios-v2.service",
|
||
"deploy/aios-v2/docker/Dockerfile",
|
||
"deploy/aios-v2/docker/compose.yaml",
|
||
"deploy/aios-v2/healthcheck/check.sh",
|
||
"deploy/aios-v2/rollback/rollback.sh",
|
||
"tasks/openclaw-production-deploy-task.yaml",
|
||
"tasks/codex-production-supervisor-task.yaml",
|
||
)
|
||
|
||
REQUIRED_APIS = (
|
||
"/api/v2/health",
|
||
"/api/v2/tenants",
|
||
"/api/v2/agents",
|
||
"/api/v2/tasks",
|
||
"/api/v2/audit",
|
||
"/api/v2/runtime",
|
||
"/api/v2/executions",
|
||
"/api/v2/gitops",
|
||
"/api/v2/hermes/results/reconcile",
|
||
)
|
||
|
||
REQUIRED_MODELS = (
|
||
"RequestContext",
|
||
"TenantRecord",
|
||
"AgentRecord",
|
||
"TaskRecord",
|
||
"TaskState",
|
||
"AuditEvent",
|
||
"OpenClawCommand",
|
||
"OpenClawResult",
|
||
"RuntimeHealth",
|
||
"HermesJob",
|
||
"HermesResult",
|
||
"HermesHealth",
|
||
"ExecutionEvidence",
|
||
)
|
||
|
||
CODE_CHANGE_PREFIXES = (
|
||
"active-runtime/app/aios_v2/",
|
||
"active-runtime/app/aios_openclaw/",
|
||
"active-runtime/app/aios_hermes/",
|
||
"deploy/aios-v2/",
|
||
"tests/aios_v2_candidate/",
|
||
"tasks/",
|
||
)
|
||
|
||
CONFLICT_MARKERS = ("<<<<<<<", "=======", ">>>>>>>")
|
||
|
||
|
||
def run_git(repo: Path, *args: str) -> tuple[int, str]:
|
||
process = subprocess.run(
|
||
["git", "-C", str(repo), *args],
|
||
text=True,
|
||
capture_output=True,
|
||
timeout=20,
|
||
)
|
||
return process.returncode, process.stdout.strip()
|
||
|
||
|
||
def chinese_ratio(text: str) -> float:
|
||
chinese = len(re.findall(r"[\u3400-\u9fff]", text))
|
||
latin = len(re.findall(r"[A-Za-z]", text))
|
||
denominator = chinese + latin
|
||
return chinese / denominator if denominator else 0.0
|
||
|
||
|
||
def all_text(paths: Iterable[Path]) -> str:
|
||
return "\n".join(path.read_text(encoding="utf-8") for path in paths)
|
||
|
||
|
||
def check_required_files(repo: Path, results: list[CheckResult]) -> list[Path]:
|
||
document_paths: list[Path] = []
|
||
for relative in REQUIRED_DOCUMENTS:
|
||
path = repo / relative
|
||
if not path.is_file():
|
||
results.append(CheckResult("错误", f"缺少必需文档：{relative}"))
|
||
continue
|
||
size = path.stat().st_size
|
||
if size < 500:
|
||
results.append(CheckResult("错误", f"文档正文过短：{relative}（{size} 字节）"))
|
||
else:
|
||
document_paths.append(path)
|
||
for relative in REQUIRED_CODE_PATHS:
|
||
if not (repo / relative).is_file():
|
||
results.append(CheckResult("错误", f"文档引用的实现路径不存在：{relative}"))
|
||
return document_paths
|
||
|
||
|
||
def check_document_quality(paths: list[Path], results: list[CheckResult]) -> None:
|
||
for path in paths:
|
||
text = path.read_text(encoding="utf-8")
|
||
relative = path.as_posix()
|
||
if any(marker in text for marker in CONFLICT_MARKERS):
|
||
results.append(CheckResult("错误", f"文档存在未解决 Git 冲突标记：{relative}"))
|
||
ratio = chinese_ratio(text)
|
||
if ratio < 0.18:
|
||
results.append(
|
||
CheckResult(
|
||
"错误",
|
||
f"文档中文正文比例过低：{relative}（{ratio:.1%}）",
|
||
)
|
||
)
|
||
headings = re.findall(r"^#{1,6}\s+.+$", text, flags=re.MULTILINE)
|
||
if len(headings) < 3:
|
||
results.append(CheckResult("错误", f"文档章节不完整：{relative}"))
|
||
if re.search(r"(?mi)^\s*(TODO|TBD)\s*$", text):
|
||
results.append(CheckResult("错误", f"文档包含未完成占位符：{relative}"))
|
||
|
||
|
||
def check_requirements(repo: Path, results: list[CheckResult]) -> None:
|
||
requirements = (repo / "docs/aios-v2/01-USER-REQUIREMENTS.md").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
matrix = (repo / "docs/aios-v2/05-TRACEABILITY-MATRIX.md").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
for number in range(1, 16):
|
||
requirement = f"UR-{number:03d}"
|
||
if requirement not in requirements:
|
||
results.append(CheckResult("错误", f"需求文档缺少：{requirement}"))
|
||
if requirement not in matrix:
|
||
results.append(CheckResult("错误", f"追踪矩阵缺少：{requirement}"))
|
||
|
||
|
||
def check_api_and_models(repo: Path, docs_text: str, results: list[CheckResult]) -> None:
|
||
code_paths = [repo / path for path in REQUIRED_CODE_PATHS if path.endswith(".py")]
|
||
code_text = all_text(path for path in code_paths if path.is_file())
|
||
for api in REQUIRED_APIS:
|
||
if api not in docs_text:
|
||
results.append(CheckResult("错误", f"正式文档未覆盖核心 API：{api}"))
|
||
if api not in code_text:
|
||
results.append(CheckResult("错误", f"代码中未找到核心 API：{api}"))
|
||
for model in REQUIRED_MODELS:
|
||
if model not in docs_text:
|
||
results.append(CheckResult("错误", f"正式文档未覆盖核心模型：{model}"))
|
||
if model not in code_text:
|
||
results.append(CheckResult("错误", f"代码中未找到核心模型：{model}"))
|
||
|
||
|
||
def check_adr_links(repo: Path, results: list[CheckResult]) -> None:
|
||
index = (repo / "docs/aios-v2/adr/README.md").read_text(encoding="utf-8")
|
||
linked = re.findall(r"\((ADR-[^)]+\.md)\)", index)
|
||
if len(set(linked)) < 6:
|
||
results.append(CheckResult("错误", "ADR 索引未覆盖六项必需决策"))
|
||
for filename in linked:
|
||
if not (repo / "docs/aios-v2/adr" / filename).is_file():
|
||
results.append(CheckResult("错误", f"ADR 索引引用不存在：{filename}"))
|
||
|
||
|
||
def check_changed_docs(
|
||
repo: Path,
|
||
base_ref: str | None,
|
||
results: list[CheckResult],
|
||
) -> None:
|
||
if not base_ref:
|
||
results.append(CheckResult("警告", "未提供基准分支，跳过变更级文档同步检查"))
|
||
return
|
||
code, output = run_git(repo, "diff", "--name-only", f"{base_ref}...HEAD")
|
||
if code != 0:
|
||
results.append(CheckResult("警告", f"无法读取与 {base_ref} 的变更，跳过变更级检查"))
|
||
return
|
||
changed = [line.strip() for line in output.splitlines() if line.strip()]
|
||
code_changed = any(path.startswith(CODE_CHANGE_PREFIXES) for path in changed)
|
||
docs_changed = any(path.startswith("docs/aios-v2/") for path in changed)
|
||
if code_changed and not docs_changed:
|
||
results.append(
|
||
CheckResult(
|
||
"错误",
|
||
"检测到架构、功能、测试、任务或部署变化，但本交付范围没有 docs/aios-v2/ 变化",
|
||
)
|
||
)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="AIOS V2 文档—代码匹配检查")
|
||
parser.add_argument("--repo", default=".", help="仓库根目录")
|
||
parser.add_argument(
|
||
"--base-ref",
|
||
default=None,
|
||
help="用于检查代码变化是否同步文档的基准引用",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
repo = Path(args.repo).resolve()
|
||
if not (repo / ".git").exists() and run_git(repo, "rev-parse", "--is-inside-work-tree")[0] != 0:
|
||
print(f"错误：不是 Git 仓库：{repo}", file=sys.stderr)
|
||
return 2
|
||
|
||
results: list[CheckResult] = []
|
||
document_paths = check_required_files(repo, results)
|
||
check_document_quality(document_paths, results)
|
||
check_requirements(repo, results)
|
||
docs_text = all_text(document_paths)
|
||
check_api_and_models(repo, docs_text, results)
|
||
check_adr_links(repo, results)
|
||
check_changed_docs(repo, args.base_ref, results)
|
||
|
||
errors = [item for item in results if item.level == "错误"]
|
||
warnings = [item for item in results if item.level == "警告"]
|
||
for item in results:
|
||
stream = sys.stderr if item.level == "错误" else sys.stdout
|
||
print(f"{item.level}：{item.message}", file=stream)
|
||
|
||
print(
|
||
f"文档—代码匹配检查完成：错误 {len(errors)}，警告 {len(warnings)}，正式文档 {len(document_paths)}。"
|
||
)
|
||
return 1 if errors else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|