agent-radar/scripts/validate.py

#!/usr/bin/env python3
"""档案 schema 校验：在发布/镜像之前把结构性缺口挡住。

为什么需要这个
--------------
2026-09-02 的教训：我手写了一个临时探针去查某个档案的 claim_vs_reality，
把 `risk.claim_vs_reality` 写成了顶层 `claim_vs_reality`，
于是得到一个"该字段为 0 条"的错误结论，差点据此去补一个根本不存在的缺口。

问题不在探针写错——**每次都靠临时手写探针，出错是迟早的事**。
写错的那一层恰好是层级，肉眼极难发现，因为它不会报错，只会安静地返回 0。

所以这个脚本的职责是：把"档案应该长什么样"固化成一处可执行的规则，
而不是散落在每次的临时检查里。

用法
----
    python scripts/validate.py            # 校验全部档案，有问题则退出码 1
    python scripts/validate.py --id orca  # 只校验一个

退出码：0 = 全部通过；1 = 有 ERROR；2 = 只有 WARN（--strict 时也返回 1）
"""

from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _common import PROJECTS, load_yaml  # noqa: E402

# ---- 必填结构 ----
REQUIRED_TOP = [
    "id", "name", "repo", "identity", "business", "scenario", "tech",
    "ecosystem", "risk", "chart", "verdict", "mirror",
]
REQUIRED_LICENSE = ["spdx", "openness", "redistributable"]
REQUIRED_CHART = ["form_depth", "engine_neutral", "local_first_score", "deliverable_depth"]
REQUIRED_MIRROR = ["upstream", "gitea_repo", "visibility", "mode"]

# claim_vs_reality 是整套体系的立身之本：
# "宣称 vs 实测落差"是这个追踪体系区别于 README 复读机的唯一东西。
# 少于 MIN_CVR 条基本等于没做深度分析。
MIN_CVR = 3

# 会随样本增加而立即过时的表述。
# 2026-09-02：maka 的 one_liner 写着「四个里唯一」，那时样本已经 7 个了——
# 这种话每加一批样本就错一次，而且错得很隐蔽（数字还在，只是不对了）。
# 「已收录样本中」是隐含的，不必写数字。
COUNT_WORDS = re.compile(r"(七个里|七个样本里|四个里|四个样本里|\d+\s*个里|\d+\s*个样本里)")


def validate_one(path: Path) -> list[tuple[str, str, str]]:
    """返回 [(级别, 位置, 说明)]，级别为 ERROR / WARN。"""
    issues: list[tuple[str, str, str]] = []
    d = load_yaml(path) or {}
    pid = d.get("id") or path.stem

    def err(where: str, msg: str) -> None:
        issues.append(("ERROR", f"{pid}.{where}", msg))

    def warn(where: str, msg: str) -> None:
        issues.append(("WARN", f"{pid}.{where}", msg))

    # ---- 顶层字段 ----
    for k in REQUIRED_TOP:
        if not d.get(k):
            err(k, "必填字段缺失或为空")

    # ---- 许可证 ----
    lic = (d.get("business") or {}).get("license") or {}
    for k in REQUIRED_LICENSE:
        if k not in lic:
            err(f"business.license.{k}", "许可证块缺该字段")

    # ---- 目录切分许可的排除后判定 ----
    # 与 mirror_sync.sh 的 fail-closed 规则保持一致：
    # 声明了 excludes 就必须给出排除后的可分发性，否则镜像脚本会直接拒绝。
    # 在这里提前报，比跑到镜像那一步才炸要好定位得多。
    mirror = d.get("mirror") or {}
    if mirror.get("excludes") and "redistributable_after_excludes" not in lic:
        err("business.license.redistributable_after_excludes",
            "档案声明了 mirror.excludes，但许可证块没有排除后的可分发性判定，"
            "mirror_sync.sh 会拒绝执行")

    # ---- claim_vs_reality ----
    cvr = (d.get("risk") or {}).get("claim_vs_reality") or []
    if len(cvr) < MIN_CVR:
        err("risk.claim_vs_reality",
            f"仅 {len(cvr)} 条，少于下限 {MIN_CVR}。这是本体系的核心字段，"
            f"不足意味着该样本没做深度分析")
    for i, item in enumerate(cvr):
        if not isinstance(item, dict):
            err(f"risk.claim_vs_reality[{i}]", "不是映射结构")
            continue
        for k in ("claim", "actual", "source"):
            if not item.get(k):
                err(f"risk.claim_vs_reality[{i}].{k}",
                    "缺该字段——没有 source 的落差分析只是观点，不是核实过的结论")

    # ---- 坐标 ----
    chart = d.get("chart") or {}
    for k in REQUIRED_CHART:
        if chart.get(k) is None:
            err(f"chart.{k}", "象限图坐标缺失，该项目不会出现在对比图里")

    # ---- mirror ----
    for k in REQUIRED_MIRROR:
        if not mirror.get(k):
            err(f"mirror.{k}", "镜像配置缺该字段")

    # ---- 会过时的计数表述 ----
    one_liner = str((d.get("verdict") or {}).get("one_liner") or "")
    m = COUNT_WORDS.search(one_liner)
    if m:
        warn("verdict.one_liner",
             f"含样本计数「{m.group(1)}」，样本增加后即过时。"
             f"去掉数字即可，「已收录样本中」本就是隐含的")

    # ---- sync_note 与 mode 自相矛盾 ----
    # 2026-09-02：qwenpaw 的 mode 已是 source-only（镜像完成），
    # sync_note 却仍写着"尚未执行"。档案里写着与事实相反的话，比没有更糟。
    mode = str(mirror.get("mode") or "")
    note = str(mirror.get("sync_note") or "")
    done_modes = {"full", "source-only"}
    stale_words = ("尚未执行", "暂未镜像", "未执行", "待带宽窗口")
    if mode in done_modes and any(w in note for w in stale_words):
        warn("mirror.sync_note",
             f"mirror.mode={mode}（已镜像）但备注仍写着未执行类措辞，档案与事实矛盾")

    return issues


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--id", help="只校验指定项目")
    ap.add_argument("--strict", action="store_true", help="WARN 也视为失败")
    args = ap.parse_args()

    if args.id:
        files = [PROJECTS / f"{args.id}.yaml"]
        if not files[0].exists():
            print(f"档案不存在: {files[0]}", file=sys.stderr)
            return 1
    else:
        files = sorted(PROJECTS.glob("*.yaml"))

    all_issues: list[tuple[str, str, str]] = []
    for f in files:
        all_issues.extend(validate_one(f))

    n_err = sum(1 for lv, _, _ in all_issues if lv == "ERROR")
    n_warn = sum(1 for lv, _, _ in all_issues if lv == "WARN")

    print(f"==> 校验 {len(files)} 个项目档案")
    for lv, where, msg in all_issues:
        mark = "✗" if lv == "ERROR" else "⚠"
        print(f"  {mark} [{lv}] {where}: {msg}")

    if not all_issues:
        print("  全部通过")
    print(f"  合计：{n_err} 错误，{n_warn} 警告")

    if n_err:
        return 1
    if n_warn and args.strict:
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())