Agent/examples/process_pipeline/script/search_eval/procedure-dsl/procedure/tools/validate.py

#!/usr/bin/env python3
"""procedure skill 唯一校验器.

agent 直接 Write 出 workflow.json 后跑本脚本, 一次性报出全部问题:
  ✗ 错误 (必须修, exit 1): 结构缺字段 / 词表不命中 / 数据流断链 / 编号重复 / value 是引用占位
  ⚠ 警告 (建议核对, 不拦截): 类型不一致 / 生成步缺提示词输入 / value 未在原文逐字命中

用法:
    python procedure/tools/validate.py --workflow outputs/case-N/workflow.json \
        [--source input/case-N.json] [--ocr outputs/case-N/_scratch/ocr.txt]

退出码: 0 = 无错误 / 1 = 有错误 / 2 = CLI 或文件问题
"""

import argparse
import json
import re
import sys
from difflib import SequenceMatcher
from pathlib import Path

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")
    sys.stderr.reconfigure(encoding="utf-8")

# 词表常量与 wf-patch.py 同源 (同目录 vocab.py)
sys.path.insert(0, str(Path(__file__).resolve().parent))
from vocab import EFFECT_LEAVES, ACTION_LEAF2PATH, TYPE_LEAVES  # noqa: E402

CATEGORIES = {"产物创造", "资产建设", "自动化", "分析", "学习"}
KINDS = {"step", "block", "nested"}
STEP_ID_RE = re.compile(r"^s\d+(\.\d+)*$")
SOURCE_REQUIRED = ["platform", "author", "date", "title", "excerpt"]
PROC_REQUIRED = ["id", "name", "purpose", "category", "declarations", "steps"]
STEP_REQUIRED = ["id", "kind", "via", "inputs", "outputs", "intent"]
IO_REQUIRED = ["type", "value", "anchor"]
INTENT_TOKEN_KINDS = {"effect", "via", "act", "in-type", "out-type"}


def norm_ws(s):
    return re.sub(r"\s+", "", s or "")


def _norm_map(raw):
    """去空白的归一文本 + 每个归一字符在原文里的下标 (用于把命中片段映射回原文逐字)."""
    chars, idx = [], []
    for i, ch in enumerate(raw):
        if not ch.isspace():
            chars.append(ch)
            idx.append(i)
    return "".join(chars), idx


def _best_span(nv, doc_norm, doc_idx, doc_raw):
    """在一个文档里找与 nv 最相似的连续片段. 返回 (相似度, 原文逐字片段) 或 None."""
    sm = SequenceMatcher(None, nv, doc_norm, autojunk=False)
    m = sm.find_longest_match(0, len(nv), 0, len(doc_norm))
    if m.size < min(15, max(8, len(nv) // 4)):     # 连最长公共串都很短 → 内容不在这个文档里
        return None
    # 以最长公共串为锚, 框出 nv 在文档里应当对应的窗口 (前后各留 1/5 余量)
    slack = max(10, len(nv) // 5)
    w_start = max(0, m.b - m.a - slack)
    w_end = min(len(doc_norm), m.b + (len(nv) - m.a) + slack)
    sm2 = SequenceMatcher(None, nv, doc_norm[w_start:w_end], autojunk=False)
    blocks = [b for b in sm2.get_matching_blocks() if b.size > 0]
    if not blocks:
        return None
    t_start = w_start + blocks[0].b                 # 收紧到首尾命中块之间
    t_end = w_start + blocks[-1].b + blocks[-1].size
    ratio = SequenceMatcher(None, nv, doc_norm[t_start:t_end], autojunk=False).ratio()
    return ratio, doc_raw[doc_idx[t_start]: doc_idx[t_end - 1] + 1]


def fix_verbatim(wf, docs):
    """把未逐字命中的文本 value 替换成原文里最相似的连续片段 (相似度够高才动).

    docs = [(norm, idx_map, raw), ...] — 原文各字段 + OCR 各算一个文档, 片段不跨文档拼接.
    返回 (fixed, skipped): fixed=[(path, ratio, 旧字数, 新字数)], skipped=[(path, 最高相似度)].
    """
    fixed, skipped = [], []
    for proc in wf.get("procedures") or []:
        pid = proc.get("id") or "p?"
        for s in proc.get("steps") or []:
            if not isinstance(s, dict):
                continue
            for arr in ("inputs", "outputs"):
                for i, io in enumerate(s.get(arr) or []):
                    if not isinstance(io, dict):
                        continue
                    v = io.get("value")
                    if not isinstance(v, str) or v.lstrip().startswith("<"):
                        continue
                    nv = norm_ws(v)
                    if len(nv) < 40 or any(nv in d[0] for d in docs):
                        continue                    # 短文本 / 已逐字命中
                    best = None
                    for dn, di, dr in docs:
                        r = _best_span(nv, dn, di, dr)
                        if r and (best is None or r[0] > best[0]):
                            best = r
                    path = f"{pid}.{s.get('id', 's?')}.{arr}[{i}]"
                    new_n = norm_ws(best[1]) if best else ""
                    if best and best[0] >= 0.60 and 0.4 * len(nv) <= len(new_n) <= 2.5 * len(nv):
                        io["value"] = best[1]
                        fixed.append((path, best[0], len(nv), len(new_n)))
                    else:
                        skipped.append((path, best[0] if best else 0.0))
    return fixed, skipped


def collect_strings(obj, out):
    if isinstance(obj, str):
        out.append(obj)
    elif isinstance(obj, list):
        for x in obj:
            collect_strings(x, out)
    elif isinstance(obj, dict):
        for x in obj.values():
            collect_strings(x, out)


class Report:
    def __init__(self):
        self.errors = []
        self.warns = []

    def err(self, path, msg):
        self.errors.append((path, msg))

    def warn(self, path, msg):
        self.warns.append((path, msg))


def check_io(rep, io, path, is_output, proc_output_ids):
    if not isinstance(io, dict):
        rep.err(path, "IO 项必须是对象")
        return
    for k in IO_REQUIRED:
        if k not in io:
            rep.err(path, f"缺字段 {k}")
    if is_output and not io.get("id"):
        rep.err(path, "输出缺 id (如 s2o1)")

    value = io.get("value", "")
    if isinstance(value, str):
        if not value.strip():
            rep.err(path, "value 为空 — 文字填原文逐字内容, 媒体填 <整段描述>")
        elif re.match(r"^\s*(←|\(同|（同|见\s*s\d)", value):
            rep.err(path, f"value 是引用占位 ({value[:20]!r}) — 引用写 anchor, value 抄真实内容")

    anchor = io.get("anchor")
    if isinstance(anchor, str) and anchor.strip():
        a = anchor.strip()
        if is_output:
            if not a.startswith("→"):
                rep.err(path, f"输出 anchor 应以 → 开头, 现为 {a!r}")
        else:
            if not a.startswith("←"):
                rep.err(path, f"输入 anchor 应以 ← 开头, 现为 {a!r}")
            else:
                ref = a.lstrip("←").strip()
                ref_id = re.sub(r"\[[^\]]*\]$", "", ref).strip()
                if re.match(r"^s\d", ref_id) and ref_id not in proc_output_ids:
                    rep.err(path, f"anchor 引用了不存在的输出编号 {ref_id!r}")

    if io.get("inferred") is True and not io.get("inferred_reason"):
        rep.err(path, "inferred=true 必须带 inferred_reason")


def check_intent(rep, intent, path):
    if not isinstance(intent, str) or not intent.strip():
        rep.err(path, "缺 intent")
        return
    if "→" in intent:
        rep.err(path, "intent 写成了公式 (含 →) — 要一句通顺人话")
    for kind in re.findall(r"\{([^:{}]+):", intent):
        if kind not in INTENT_TOKEN_KINDS:
            rep.err(path, f"intent 标记类别 {{{kind}:}} 非法 — 只能用 {sorted(INTENT_TOKEN_KINDS)}")
    plain = re.sub(r"\{[^:{}]+:([^{}]*)\}", r"\1", intent)
    if len(plain) > 40:
        rep.warn(path, f"intent 偏长 ({len(plain)} 字, 建议 ≤25)")


def validate(wf, vocab, source_text):
    rep = Report()

    src = wf.get("source")
    if not isinstance(src, dict):
        rep.err("source", "缺 source 块")
    else:
        for k in SOURCE_REQUIRED:
            if not src.get(k):
                rep.err("source", f"缺字段 {k}")

    procs = wf.get("procedures")
    if not isinstance(procs, list) or not procs:
        rep.err("procedures", "procedures 必须是非空数组")
        return rep

    for proc in procs:
        pid = proc.get("id") or "p?"
        for k in PROC_REQUIRED:
            if k not in proc:
                rep.err(pid, f"缺字段 {k}")
        if proc.get("category") and proc["category"] not in CATEGORIES:
            rep.err(pid, f"category={proc['category']!r} 不合法 — 只能是 {sorted(CATEGORIES)}")

        decl = proc.get("declarations")
        if isinstance(decl, dict):
            for k in ("inputs", "resources", "returns"):
                if k not in decl:
                    rep.err(f"{pid}.declarations", f"缺字段 {k}")
        # type_registry: 每条要 extends 到标准类型叶子 + desc
        registry = proc.get("type_registry") or {}
        for tname, entry in registry.items():
            tpath = f"{pid}.type_registry.{tname}"
            if not isinstance(entry, dict) or not entry.get("extends"):
                rep.err(tpath, "缺 extends")
            elif entry["extends"] not in vocab["type_leaves"]:
                rep.err(tpath, f"extends={entry['extends']!r} 不是 type 词表叶子")
            if not isinstance(entry, dict) or not entry.get("desc"):
                rep.err(tpath, "缺 desc")
        known_types = vocab["type_leaves"] | set(registry)

        steps = proc.get("steps") or []
        block_ids = {s.get("id") for s in steps if isinstance(s, dict) and s.get("kind") == "block"}
        output_ids = set()
        out_type_by_id = {}
        for s in steps:
            if not isinstance(s, dict):
                continue
            for o in s.get("outputs") or []:
                if isinstance(o, dict) and o.get("id"):
                    spath = f"{pid}.{s.get('id', 's?')}"
                    if o["id"] in output_ids:
                        rep.err(spath, f"输出编号 {o['id']!r} 重复")
                    output_ids.add(o["id"])
                    out_type_by_id[o["id"]] = o.get("type")

        for s in steps:
            if not isinstance(s, dict):
                rep.err(pid, "steps 含非对象项")
                continue
            sid = s.get("id") or "s?"
            spath = f"{pid}.{sid}"
            for k in STEP_REQUIRED:
                if k not in s:
                    rep.err(spath, f"缺字段 {k}")
            if s.get("id") and not STEP_ID_RE.match(s["id"]):
                rep.err(spath, f"step id {s['id']!r} 不合规 (s1 / s5.1)")
            kind = s.get("kind")
            if kind not in KINDS:
                rep.err(spath, f"kind={kind!r} 不合法 — 只能是 {sorted(KINDS)}")
            if kind == "nested":
                if not s.get("group"):
                    rep.err(spath, "nested 步缺 group")
                elif s["group"] not in block_ids:
                    rep.err(spath, f"group={s['group']!r} 不是已存在的 block id")

            # render schema: directive 若存在必须是字符串 (人工/控制步没有 prompt 就省略字段或写空串)
            if "directive" in s and not isinstance(s["directive"], str):
                rep.err(spath, f"directive={s['directive']!r} 必须是字符串 — 没有就删掉该字段, 不要写 null")
            if kind in ("step", "nested"):
                # render 硬门禁对齐: via 非空、inputs/outputs 非空数组 (缺 IO 按工艺推断补, 标 inferred)
                if not str(s.get("via") or "").strip():
                    rep.err(spath, "via 为空 — 步骤要写用的工具 (human / 工具名 / 占位 (AI 生图工具))")
                for arr, label in (("inputs", "输入"), ("outputs", "输出")):
                    if not s.get(arr):
                        rep.err(spath, f"{arr} 为空数组 — 步骤必有{label}; 原文没明写就按工艺推断补, "
                                       f"标 inferred:true + inferred_reason, 不要为过校验而删 IO")
                # effect / action 必须命中词表
                eff = s.get("effect")
                if not eff:
                    rep.err(spath, "缺 effect")
                elif eff not in vocab["effect_leaves"]:
                    rep.err(spath, f"effect={eff!r} 不是合法叶子 — 9 选 1: {sorted(vocab['effect_leaves'])}")
                act = s.get("action")
                if not act:
                    rep.err(spath, "缺 action")
                elif act not in vocab["action_leaves"] and act not in vocab["action_paths"]:
                    rep.err(spath, f"action={act!r} 不在动作词表 (叶子名或 根/…/叶 全路径)")
                # substance / form 必须显式处理 (填值或 null)
                for k in ("substance", "form"):
                    if k not in s:
                        rep.err(spath, f"缺 {k} (没有就显式设 null)")

            check_intent(rep, s.get("intent"), spath)

            for arr, is_out in (("inputs", False), ("outputs", True)):
                for i, io in enumerate(s.get(arr) or []):
                    iopath = f"{spath}.{arr}[{i}]"
                    check_io(rep, io, iopath, is_out, output_ids)
                    if not isinstance(io, dict):
                        continue
                    t = io.get("type")
                    if t and t not in known_types:
                        rep.err(iopath, f"type={t!r} 不在词表也没在 type_registry 挂靠")
                    # 输入类型与来源输出类型一致性
                    if not is_out and isinstance(io.get("anchor"), str):
                        ref_id = re.sub(r"\[[^\]]*\]$", "", io["anchor"].lstrip("←").strip()).strip()
                        src_t = out_type_by_id.get(ref_id)
                        if src_t and t and src_t != t:
                            rep.warn(iopath, f"输入 type={t!r} 与来源 {ref_id} 的 type={src_t!r} 不一致")

            # 生成步建议有提示词输入 + directive
            if kind in ("step", "nested") and isinstance(s.get("action"), str) and s["action"].split("/")[0] == "生成":
                in_types = {io.get("type") for io in s.get("inputs") or [] if isinstance(io, dict)}
                if not in_types & {"提示词", "负向提示词", "描述"}:
                    rep.warn(spath, "生成步没有 提示词/描述 类输入 — 确认是否漏建")

        # value 逐字核对 (有 --source 才跑)
        if source_text:
            for s in steps:
                if not isinstance(s, dict):
                    continue
                for arr in ("inputs", "outputs"):
                    for i, io in enumerate(s.get(arr) or []):
                        if not isinstance(io, dict):
                            continue
                        v = io.get("value")
                        if not isinstance(v, str) or v.lstrip().startswith("<"):
                            continue  # 媒体类描述不要求逐字
                        nv = norm_ws(v)
                        if len(nv) >= 40 and nv not in source_text:
                            rep.warn(f"{pid}.{s.get('id','s?')}.{arr}[{i}]",
                                     f"value ({len(nv)} 字) 未在原文/OCR 逐字命中 — 确认没缩写/改写"
                                     f" (可加 --fix-verbatim 自动替换为原文片段)")
    return rep


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--workflow", required=True)
    ap.add_argument("--source", help="原文 case json — 给了才跑 value 逐字核对")
    ap.add_argument("--ocr", help="配图 OCR 文本, 并入逐字核对语料")
    ap.add_argument("--fix-verbatim", action="store_true",
                    help="把未逐字命中的文本 value 自动替换为原文里最相似的连续片段 "
                         "(相似度 ≥60%% 才动, 改不动的留警告; 需要 --source)")
    args = ap.parse_args()

    try:
        wf = json.loads(Path(args.workflow).read_text(encoding="utf-8"))
    except FileNotFoundError:
        print(f"✗ 文件不存在: {args.workflow}")
        return 2
    except json.JSONDecodeError as e:
        print(f"✗ JSON 解析失败: {e}")
        return 1

    vocab = {
        "effect_leaves": EFFECT_LEAVES,
        "action_leaves": set(ACTION_LEAF2PATH),
        "action_paths": set(ACTION_LEAF2PATH.values()),
        "type_leaves": TYPE_LEAVES,
    }

    source_text = ""
    docs = []          # 逐字段文档 [(norm, idx_map, raw)], --fix-verbatim 用 (片段不跨字段拼)
    if args.source:
        chunks = []
        collect_strings(json.loads(Path(args.source).read_text(encoding="utf-8")), chunks)
        source_text = norm_ws("".join(chunks))
        docs = [(*_norm_map(c), c) for c in chunks if len(norm_ws(c)) >= 40]
    if args.ocr and Path(args.ocr).exists():
        ocr_raw = Path(args.ocr).read_text(encoding="utf-8")
        source_text += norm_ws(ocr_raw)
        docs.append((*_norm_map(ocr_raw), ocr_raw))

    if args.fix_verbatim:
        if not docs:
            print("✗ --fix-verbatim 需要 --source (原文语料)")
            return 2
        fixed, skipped = fix_verbatim(wf, docs)
        if fixed:
            Path(args.workflow).write_text(
                json.dumps(wf, ensure_ascii=False, indent=2), encoding="utf-8")
        for p, r, a, b in fixed:
            print(f"  ✦ [{p}] value 已替换为原文逐字片段 (相似度 {r:.0%}, {a}→{b} 字)")
        for p, r in skipped:
            print(f"  ⚠ [{p}] 找不到足够相似的原文连续片段 (最高 {r:.0%}) — 可能是多段拼接/自创内容, 人工处理")

    rep = validate(wf, vocab, source_text)

    for path, msg in rep.errors:
        print(f"  ✗ [{path}] {msg}")
    for path, msg in rep.warns:
        print(f"  ⚠ [{path}] {msg}")
    print(f"[validate] {args.workflow}: {len(rep.errors)} 错误, {len(rep.warns)} 警告")
    return 1 if rep.errors else 0


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