Agent/examples/process_pipeline/script/search_eval/procedure-dsl/spec-back-1/tools/verify-io.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""verify-io.py — Phase 1 收口的专用「IO 校验」步骤 (填完 value 后、进 Phase 2 前必跑)。

专门校验三件事, 报问题就回去补 (修完重跑直到通过):
  1. **文本 IO 逐字**: 每个**文本类** input/output 的 value 是不是在原文(正文+配图 OCR)里有
     连续对应 —— 媒体类(图/视频/音频)IO **不要求**对应, 直接跳过。
  2. **提示词提取**: 每个**工具生成步**(via 是具体生成工具)是不是把原文给这步的整段 prompt
     提取进了 directive 或某个文本输入 —— 弱模型常把生成步建成「媒体输入→媒体输出」, 把原文
     那段 prompt(最该提取的内容, 如「半剖面视图…」「视觉层：…技术层：…」)整段漏掉。
  3. **declarations 收口**: 每个工序的 declarations.returns 补了没(必), inputs/resources 提醒。

退出码: 0 通过 / 1 有阻塞项要修 / 2 CLI/文件错。

用法:
    python spec/tools/verify-io.py --workflow outputs/case-{N}/workflow.json \
        --source input/case-{N}.json --ocr outputs/case-{N}/_scratch/ocr.txt
"""
import argparse
import importlib.util
import re
import sys
from pathlib import Path

import json

# 媒体类 value 应整体用 <…> 包成 <具体描述> (强模型参考一致): <极简写实背景图：深钢蓝-松绿-冷白渐层…>
_MEDIA_BRACKET_OK = re.compile(r'^\s*<.+>\s*$', re.S)

DSL_ROOT = Path(__file__).resolve().parent.parent.parent

for _s in (sys.stdout, sys.stderr):
    if hasattr(_s, 'reconfigure'):
        try:
            _s.reconfigure(encoding='utf-8', errors='replace')
        except Exception:
            pass


def _load_lint():
    """复用 lint-case.py 的 _norm / check_value_verbatim / _type_modality (文件名带连字符, 按路径载)."""
    p = Path(__file__).resolve().parent / 'lint-case.py'
    spec = importlib.util.spec_from_file_location('_lint_case_mod', p)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


# 装配/排版/人工类工具不要求 prompt (它们不是"按提示词生成"). 用较长的判别词, 避免 'na' 误伤 nanobanana
_NON_PROMPT_VIA = ('human', 'ppt', 'adobe', 'illustrat', 'inkscape', 'photoshop',
                   '剪辑', '剪映', '拼接', '排版', '拼版', '合成', '后期', '裁剪', '修图',
                   '人工', 'excel', 'word')


def _is_generation_via(via: str) -> bool:
    v = (via or '').strip().lower()
    if not v or v in ('-', 'n/a', 'na', '人工', '人'):
        return False
    return not any(k in v for k in _NON_PROMPT_VIA)


# 原文 prompt 的**独特**分层/要素标记: 命中后原文紧跟这些 = 只提了一部分 (截断)。
# 只放区分度高的 (视觉层/技术层/核心要素…); 不放 产品/人物/环境 这种常见词 (满篇都是, 会误判)。
_LAYER_MARKERS = ('视觉层', '技术层', '氛围层', '核心要素', '视觉风格', '输出要求', '技术规格')


def _has_prompt_type(t: str) -> bool:
    return '提示词' in (t or '') or 'prompt' in (t or '').lower()


def check_prompt_extraction(data: dict, lint, source_norm: str) -> list:
    """提示词建模 + 完整性 (对齐强模型参考: 提示词是**数据**, 建成 type=提示词 的 IO value, directive 不装它):
      1. 生成步必须有 type=提示词 的输入 (原文整段 prompt 当数据流入), 没有=漏建;
      2. directive 里疑似装了提示词原文 = 放错地方, 该挪到 IO;
      3. type=提示词 的 value 逐字但截断 (原文后面还有分层/要素) = 只提了一部分。
    (value 是否逐字由「文本 IO 逐字」那条统一管, 这里只管"建对地方"和"提全".)
    """
    issues = []
    for p in data.get('procedures', []):
        for s in p.get('steps', []) or []:
            if not isinstance(s, dict) or s.get('kind') == 'block':
                continue
            sid = f"{p.get('id')}.{s.get('id')}"
            via = (s.get('via') or '').strip()
            directive = (s.get('directive') or '').strip()

            # 2. directive 不该装提示词 (长且像在引原文) —— 不限生成步
            if directive:
                dn = lint._norm(directive)
                if len(dn) >= 12 and lint._longest_run(dn, source_norm) >= 12:
                    issues.append(
                        f"[{sid}] directive 里疑似装了提示词原文 —— directive 只放「给工具的指示」"
                        f"(如严格反推/比例 2:3); **提示词原文是数据, 挪到 type=提示词 的输入/输出 value**。")

            # 1. 生成步必须有 type=提示词 的输入
            if _is_generation_via(via):
                has_prompt_in = any(_has_prompt_type(io.get('type', '')) for io in (s.get('inputs') or []))
                if not has_prompt_in:
                    issues.append(
                        f"[{sid}] via={via!r} 是生成步但**没有 type=提示词 的输入** —— 原文的整段 prompt 是数据, "
                        f"建成 type=提示词 的输入(value 逐字完整; 通常由一个 human「写提示词」步 OUT=提示词、本步 anchor 引入), "
                        f"别塞 directive、也别只写「XX需求」标签。原文确实没给详细 prompt 才可省。")

            # 3. type=提示词 的 value 逐字但截断 (原文紧跟还有分层/要素)
            for kind in ('inputs', 'outputs'):
                for io in (s.get(kind) or []):
                    if not _has_prompt_type(io.get('type', '')):
                        continue
                    v = (io.get('value') or '').strip()
                    if not v or v.startswith('<'):
                        continue
                    vn = lint._norm(v)
                    if len(vn) < 8:
                        continue
                    pos = source_norm.find(vn)
                    if pos >= 0:
                        tail = source_norm[pos + len(vn): pos + len(vn) + 40]
                        cont = [m for m in _LAYER_MARKERS if m in tail and m not in vn]
                        if cont:
                            issues.append(
                                f"[{sid}].{kind[:3]} type=提示词 的 value 可能**只提了一部分** —— 原文后面紧跟 "
                                f"{cont[:3]} 等, @quote 止锚要延到这段 prompt 真正结束(把所有分层/要素提全)。")
    return issues


def check_media_brackets(data: dict, lint) -> list:
    """媒体类(图/视频/音频)IO 的 value 必须整体用 <…> 包成 <具体描述>, 别写裸标签."""
    issues = []
    for p in data.get('procedures', []):
        reg = p.get('type_registry') or {}
        for s in p.get('steps', []) or []:
            if not isinstance(s, dict) or s.get('kind') == 'block':
                continue
            for kind in ('inputs', 'outputs'):
                for io in (s.get(kind) or []):
                    if io.get('inferred'):
                        continue
                    if lint._type_modality(io.get('type', ''), reg) != 'media':
                        continue
                    v = (io.get('value') or '').strip()
                    if not v:
                        continue
                    if not _MEDIA_BRACKET_OK.match(v):
                        issues.append(
                            f"[{p.get('id')}.{s.get('id')}.{kind[:3]}] type={io.get('type', '')!r}(媒体) "
                            f"value={v[:24]!r} 没用 <…> 包 —— 媒体类 value 写成 <具体描述>"
                            f"(如 <一张冲锋衣登山者暴雨场景图，水珠滚落，冷色调>)。")
    return issues


def check_declarations(data: dict) -> tuple:
    """returns 空 = 必补; inputs 空 = 提醒. 返回 (block_list, warn_list)."""
    block, warn = [], []
    for p in data.get('procedures', []):
        decl = p.get('declarations') or {}
        rets = decl.get('returns')
        if not rets or (isinstance(rets, dict) and not any(rets.values())):
            block.append(f"[{p.get('id')}] declarations.returns 空 — 补这个工序最终返回什么(type)")
        if not decl.get('inputs'):
            warn.append(f"[{p.get('id')}] declarations.inputs 空 — 确认外部输入(没有可不填)")
    return block, warn


def main() -> None:
    ap = argparse.ArgumentParser(prog='verify-io.py', description='Phase 1 收口 IO 校验')
    ap.add_argument('--workflow', type=Path, required=True)
    ap.add_argument('--source', type=Path, required=True, help='原文 input/case-N.json (逐字比对用)')
    ap.add_argument('--ocr', type=Path, default=None, help='配图 OCR 文本 (可选, 并入比对语料)')
    args = ap.parse_args()

    if not args.workflow.exists():
        print(f'verify-io: workflow 不存在 {args.workflow}', file=sys.stderr)
        sys.exit(2)
    try:
        data = json.loads(args.workflow.read_text(encoding='utf-8'))
    except json.JSONDecodeError as e:
        print(f'verify-io: workflow 不是合法 JSON: {e}', file=sys.stderr)
        sys.exit(2)

    lint = _load_lint()
    _raw, source_norm = lint._load_source_corpus(args.source, args.ocr)
    if not source_norm:
        print(f'verify-io: 读不到原文 {args.source} (逐字校验需要它)', file=sys.stderr)
        sys.exit(2)

    case_id = data.get('case_id', '?')
    print(f'[verify-io] case-{case_id} ({args.workflow.name})')

    # 1. 文本 IO 逐字 (媒体跳过)
    io_hints = lint.check_value_verbatim(data, source_norm)
    if io_hints:
        print(f'  · 文本 IO 逐字: {len(io_hints)} 处未对上原文 (媒体类已跳过)')
        for h in io_hints:
            print(f'      ✗ {h}')
        print('     → 此时可重读原文; 用 @quote|起锚|止锚 + wf-patch --resolve-quotes 拽完整原文; '
              '原文确无则标 inferred:true+reason 或 value 写「原文未提供」。')
    else:
        print('  · 文本 IO 逐字: OK (文本类 value 都对上原文; 媒体类不要求)')

    # 2. 提示词提取
    prompt_issues = check_prompt_extraction(data, lint, source_norm)
    if prompt_issues:
        print(f'  · 提示词提取: {len(prompt_issues)} 个生成步没提取到提示词')
        for h in prompt_issues:
            print(f'      ✗ {h}')
    else:
        print('  · 提示词提取: OK (生成步都有提示词)')

    # 3. 媒体 value 的 <…> 包裹
    media_issues = check_media_brackets(data, lint)
    if media_issues:
        print(f'  · 媒体 value <…>: {len(media_issues)} 处没包 (媒体类 value 要写 <具体描述>)')
        for h in media_issues:
            print(f'      ✗ {h}')
    else:
        print('  · 媒体 value <…>: OK')

    # 4. declarations 收口
    decl_block, decl_warn = check_declarations(data)
    if decl_block or decl_warn:
        print(f'  · declarations 收口: {len(decl_block)} 必补 / {len(decl_warn)} 待确认')
        for d in decl_block:
            print(f'      ✗ {d}')
        for d in decl_warn:
            print(f'      ⚠ {d}')
        print('     → 用 wf-patch 补 declarations.inputs/resources/returns, '
              '例: `--set p1.declarations.returns.type=产品概念图`。')
    else:
        print('  · declarations 收口: OK')

    n_block = len(io_hints) + len(prompt_issues) + len(media_issues) + len(decl_block)
    if n_block:
        print(f'\n有 {n_block} 个阻塞项要修 (文本IO {len(io_hints)} + 提示词 {len(prompt_issues)} '
              f'+ 媒体<…> {len(media_issues)} + returns {len(decl_block)})。修完**重跑本脚本**确认通过, 才进 Phase 2。',
              file=sys.stderr)
        sys.exit(1)
    print('\n✓ IO 校验通过, 可进 Phase 2 归类标注。')
    sys.exit(0)


if __name__ == '__main__':
    main()