Agent/examples/process_pipeline/script/search_eval/procedure-dsl/spec-back-1/tools/render-case.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
render-case.py — Procedure DSL · 阶段三 3.3 落盘&渲染脚本.

把 case_data (符合 spec/format/case-data.schema.json) 通过 spec/tools/renderer.build_html
渲染成单文件 HTML.

用法:

  渲染:
       render-case.py --workflow outputs/case-N/workflow.json \
                      --source-input input/case-N-raw.json \
                      --page-title "Case N · 主题" --case-id N \
                      --out outputs/case-N/case-N-<slug>.html

  只校验, 不渲染:
       render-case.py --workflow outputs/case-N/workflow.json --source-input input/case-N-raw.json --validate

退出码:
  0  成功 (渲染或校验通过)
  1  IO / schema / 渲染异常
  2  CLI 参数错误

依赖:
  必需: spec/tools/renderer.py (跟本脚本同目录, 自动加入 sys.path)
  可选: jsonschema (装了就用 Draft 2020-12 校验; 没装则只做最小结构检查)
"""
from __future__ import annotations

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

# Windows cp1252 console 不能直接打中文/勾选号; 把 stdout/stderr 强转 UTF-8.
for _stream in (sys.stdout, sys.stderr):
    if hasattr(_stream, 'reconfigure'):
        try:
            _stream.reconfigure(encoding='utf-8', errors='replace')
        except Exception:
            pass

# DSL_ROOT = procedure-dsl/  (本脚本位于 procedure-dsl/spec/tools/)
DSL_ROOT = Path(__file__).resolve().parent.parent.parent
SCHEMA_PATH = DSL_ROOT / 'spec' / 'format' / 'case-data.schema.json'
# renderer.py 跟本脚本同目录 (spec/tools/), 直接用 parent 最 robust.
TEMPLATE_DIR = Path(__file__).resolve().parent


def _import_build_html():
    """延迟 import build_html, 把 examples/_build 接到 sys.path. 失败时给出明确错误."""
    sys.path.insert(0, str(TEMPLATE_DIR))
    try:
        from renderer import build_html  # type: ignore
    except ImportError as e:
        die(f'无法导入 renderer.build_html (期望路径: {TEMPLATE_DIR}/renderer.py): {e}')
    return build_html


def die(msg: str, code: int = 1):
    print(f'render-case: {msg}', file=sys.stderr)
    sys.exit(code)


def load_json(path: Path) -> dict:
    if not path.exists():
        die(f'文件不存在: {path}')
    try:
        return json.loads(path.read_text(encoding='utf-8'))
    except json.JSONDecodeError as e:
        die(f'{path} 不是合法 JSON: {e}')


def die_short(msg: str):
    die(msg)


def merge_raw_source(case_data: dict, raw_path: Path) -> None:
    """从 input/case-{N}-raw.json 抽 body_text + 封面 + 图集兜底, in-place 填到 case_data.source.

    设计: raw 是事实源 (原帖正文 / 封面 / 标题 / URL), case_data.source.body_text 和 cover_image
    被 raw 覆盖. 其他字段 (excerpt / author / date / platform) **保留 case_data 原值** ——
    那些是 Agent 推断/总结的友好版本, 不应被 raw 覆盖.

    图集兜底逻辑: 微信公众号正文里有 `[image:URL]` inline markup → renderer 自动渲染就好.
    小红书等"短文 + 独立图集" 平台 body_text 不含 inline markup → 把 image_url_list 里
    既没 inline 也不是封面的图, 以 `[image: URL]` 形式 append 到 body_text 末尾 (附 "附图" 分隔符).
    这样 renderer.render_source_body 一视同仁地把它们也渲染成 <img>, 不重复封面.
    """
    import re

    raw = load_json(raw_path)
    src = case_data.setdefault('source', {})

    body = raw.get('body_text', '') or ''

    # cover_image: image_url_list 里 image_type=2 (封面) 的第一张; 没有就 fallback 整列第一张
    images = raw.get('image_url_list') or []
    covers = [it for it in images if isinstance(it, dict) and it.get('image_type') == 2]
    pick = covers[0] if covers else (images[0] if images and isinstance(images[0], dict) else None)
    cover_url = ''
    if pick and pick.get('image_url'):
        cover_url = pick['image_url']
        src['cover_image'] = cover_url

    # 图集兜底: 检查哪些图既没在 body inline, 也不是封面 → append 到 body 末尾
    inlined_urls = set(re.findall(r'\[image:\s*(\S+?)\]', body))
    missing = []
    for it in images:
        if not isinstance(it, dict):
            continue
        u = it.get('image_url', '') or ''
        if not u:
            continue
        if u in inlined_urls:
            continue   # 已 inline 不重复
        if u == cover_url:
            continue   # 封面已 src-cover 单独渲染
        missing.append(u)

    if missing:
        suffix = '\n\n--- 附图 ---\n' + '\n'.join(f'[image: {u}]' for u in missing)
        body = body + suffix

    if body:
        src['body_text'] = body

    # title / url 兜底: case_data 漏填时从 raw 补 (Agent 一般会填)
    if not src.get('title') and raw.get('title'):
        src['title'] = raw['title']
    if not src.get('url') and raw.get('content_link'):
        src['url'] = raw['content_link']


# =============================================================================
# 校验
# =============================================================================
def validate(case_data: dict, strict: bool = True) -> list[str]:
    """返回错误信息列表 (空 list = 通过).

    优先使用 jsonschema (Draft 2020-12); 不可用时退化到最小结构检查 (必填 key 存在 + 类型对).
    strict=True 时, 任何错误抛 SystemExit(1); strict=False 仅返回错误列表.
    """
    errors: list[str] = []

    try:
        from jsonschema import Draft202012Validator  # type: ignore
        schema = load_json(SCHEMA_PATH)
        validator = Draft202012Validator(schema)
        for err in sorted(validator.iter_errors(case_data), key=lambda e: list(e.absolute_path)):
            path = '$' + ''.join(f'.{p}' if isinstance(p, str) else f'[{p}]' for p in err.absolute_path)
            errors.append(f'{path}: {err.message}')
    except ImportError:
        # 退化: 最小检查
        errors.extend(_minimal_check(case_data))

    if errors and strict:
        for e in errors:
            print(f'  ✗ {e}', file=sys.stderr)
        print(f'  ↳ schema: {SCHEMA_PATH}  (要看字段约束直接 Read 这个文件, 不要猜路径)', file=sys.stderr)
        die(f'schema 校验失败, 共 {len(errors)} 项 (装 `pip install jsonschema` 可看完整 Draft 2020-12 报告)')
    return errors


def _minimal_check(d: dict) -> list[str]:
    """无 jsonschema 时的兜底: 仅检 顶层必填 + steps 元素必填 + IO 元素必填."""
    errs: list[str] = []
    top_required = ['page_title', 'procedure', 'declarations', 'source', 'steps']
    for k in top_required:
        if k not in d:
            errs.append(f'$.{k}: 缺失必填字段')
    if not isinstance(d.get('steps'), list):
        errs.append('$.steps: 必须是 array')
        return errs
    # 所有 step 都要的基础字段; effect/action 仅非 block 步要 (block 是控制容器)
    step_required = ['id', 'kind', 'via', 'inputs', 'outputs', 'intent']
    exec_only = ['effect', 'action']               # kind != block 才要
    io_required = ['type', 'value', 'anchor']
    for i, step in enumerate(d['steps']):
        if not isinstance(step, dict):
            errs.append(f'$.steps[{i}]: 必须是 object'); continue
        need = step_required + ([] if step.get('kind') == 'block' else exec_only)
        for k in need:
            if k not in step:
                errs.append(f'$.steps[{i}].{k}: 缺失必填字段')
        for io_kind in ('inputs', 'outputs'):
            for j, item in enumerate(step.get(io_kind, []) or []):
                if not isinstance(item, dict):
                    errs.append(f'$.steps[{i}].{io_kind}[{j}]: 必须是 object'); continue
                for k in io_required:
                    if k not in item:
                        errs.append(f'$.steps[{i}].{io_kind}[{j}].{k}: 缺失必填字段')
    return errs


# =============================================================================
# 占位门禁: 文本类 value 是 <占位> → 拒绝渲染 (强制回填真实内容)
# 模态判定与 lint-case.py Check 3 一致: 文本类必须真内容, 媒体类可 <描述>, inferred 放行.
# =============================================================================
_PH_RE = re.compile(r'^\s*<[^>]*>\s*$')
# 显式「原文未提供」标记 → 放行 (LLM 判断原文确无该信息时的轻量逃生, 等同 inferred)。
# 写法如: <占位>(原文未提供) / <某分析>（原文未给出）
_NOSRC_RE = re.compile(r'原文(未提供|未给出|没有|无)')
_TEXT_KW = ('提示词', '描述', '参数', '评', '大纲', '脚本', '文案', '歌词', '字幕',
            '标题', '正文', '词', '知识', '工作流', '对标', '规格', '批处理', '模板', '版式',
            '数据', '分析', '报告', '记录', '方案', '思路', '设定', '依据', '标准', '清单', '列表', '文本', '文字')
_MEDIA_KW = ('图', '视频', '音频', '帧', '片段', '截图', '蒙版', '音效', '配音', 'BGM',
             '数字人', '滤镜', '海报', '封面')


def _modality(type_name: str, type_reg: dict) -> str:
    base, seen = type_name, set()
    while base in (type_reg or {}) and base not in seen:
        seen.add(base)
        ent = type_reg[base]
        ext = ent.get('extends') if isinstance(ent, dict) else None
        if not ext:
            break
        base = ext
    nm = base or type_name or ''
    if any(k in nm for k in _TEXT_KW):
        return 'text'
    if any(k in nm for k in _MEDIA_KW):
        return 'media'
    return 'unknown'


def check_placeholder_values(case_data: dict) -> list[str]:
    """揪出文本类(非媒体、非 inferred)IO value 仍是 <占位> 的——这些必须回填真内容才放行渲染."""
    bad: list[str] = []
    for p in case_data.get('procedures') or []:
        pid = p.get('id') or '?'
        type_reg = p.get('type_registry') or {}
        for i, step in enumerate(p.get('steps') or []):
            if not isinstance(step, dict):
                continue
            for kind in ('inputs', 'outputs'):
                for j, io in enumerate(step.get(kind) or []):
                    if not isinstance(io, dict) or io.get('inferred'):
                        continue
                    v = io.get('value')
                    if not isinstance(v, str):
                        continue
                    if _NOSRC_RE.search(v):
                        continue   # 显式标了「原文未提供」→ 放行 (LLM 确认原文确无)
                    if not _PH_RE.match(v):
                        continue
                    if _modality(io.get('type', '') or '', type_reg) == 'media':
                        continue   # 图/视频/音频 用 <描述> 合理
                    bad.append(f"[{pid}] {step.get('id')}.{kind}[{j}] type={io.get('type','')!r} value={v.strip()!r}")
    return bad


# =============================================================================
# main
# =============================================================================
def main() -> None:
    ap = argparse.ArgumentParser(
        prog='render-case.py',
        description='Procedure DSL · case_data → HTML 渲染',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__.split('用法:')[1] if __doc__ else '',
    )
    ap.add_argument('--workflow', type=Path, required=True,
                    help='workflow.json (含 procedures 数组). 内部组装 case_data = '
                         'workflow + source-input merge + --page-title + --case-id, 不落盘 case_data.json. '
                         '配合 --source-input / --page-title / --case-id 一起用.')
    ap.add_argument('--page-title', type=str, default=None,
                    help='页面标题, 仅与 --workflow 配对. e.g. "Case 5 · 产品宣传图 AI 工作流可视化"')
    ap.add_argument('--case-id', type=str, default=None,
                    help='case 编号, 仅与 --workflow 配对')
    ap.add_argument('--source-input', type=Path, dest='source_input',
                    help='可选: 原帖 raw json 路径 (e.g. input/case-N-raw.json). 给了的话, 自动把 '
                         'raw.body_text + 封面图 / 标题 / URL 填到 case_data.source 对应字段 '
                         '(body_text/cover_image 直接覆盖 case_data 同字段, title/url 仅在 case_data 缺时填), '
                         '让 HTML 折叠原文区显示完整原帖正文 + 内嵌图. excerpt/author/date 等 Agent 推断字段不动.')
    ap.add_argument('--out', type=Path,
                    help='输出 HTML 路径. --validate 时可省略')
    ap.add_argument('--validate', action='store_true',
                    help='只做 schema 校验, 不渲染. 退出码 0=通过, 1=失败')
    ap.add_argument('--no-validate', action='store_true',
                    help='跳过校验直接渲染 (不推荐)')
    ap.add_argument('--allow-placeholders', action='store_true',
                    help='跳过「文本类 value 占位」门禁 (调试用; 默认: 有文本类 <占位> value 就拒绝出 HTML)')
    args = ap.parse_args()

    # workflow.json 装 procedures 主体 + cli 套 case-level meta
    case_data = load_json(args.workflow)
    if args.page_title:
        case_data['page_title'] = args.page_title
    if args.case_id is not None:
        try:
            case_data['case_id'] = int(args.case_id)
        except ValueError:
            case_data['case_id'] = args.case_id
    # source 由 --source-input 提供 (workflow.json 不含 source); 没给就让 schema validate 报错

    # raw 原帖 → case_data.source merge (在 validate 前, 让 schema 校验看到的就是完整版)
    if args.source_input:
        merge_raw_source(case_data, args.source_input)

    def _count_steps(cd):
        return sum(len(p.get('steps') or []) for p in cd.get('procedures', []))

    if not args.no_validate:
        validate(case_data, strict=not args.validate)

    # 占位硬门禁: 文本类 value 仍是 <占位> → 拒绝出 HTML, 逼回填真内容 (媒体类/inferred/原文未提供 放行)。
    # 放在这里 (validate 块外), 所以 --no-validate 也绕不过; --validate 模式也会查。
    if not args.allow_placeholders:
        ph = check_placeholder_values(case_data)
        if ph:
            print(f'render-case: ✗ {len(ph)} 处文本类 value 仍是 <占位>, 拒绝出 HTML —— 必须回填真实内容:', file=sys.stderr)
            for h in ph:
                print(f'    - {h}', file=sys.stderr)
            print('  → 用 wf-patch 的 `@quote|<起锚>|<止锚>` + `--resolve-quotes --source <原文> --ocr <ocr.txt>` '
                  '从原文/OCR 回填真内容;\n'
                  '    原文确无该信息 → value 写成 `<占位>(原文未提供)`(或给该 IO 标 `inferred:true`+reason)即可放行;\n'
                  '    媒体类(图/视频/音频)用 `<描述>` 本就不受此限。', file=sys.stderr)
            print('  (确需带占位渲染才加 --allow-placeholders)', file=sys.stderr)
            die(f'{len(ph)} 处文本类 value 占位未回填')

    if not args.no_validate and args.validate:
        n_procs = len(case_data.get('procedures', []))
        print(f'render-case: ✓ {SCHEMA_PATH.name} 校验通过 '
              f'({n_procs} procedure{"s" if n_procs > 1 else ""}, {_count_steps(case_data)} steps)')
        if not args.out:
            return

    if not args.out:
        ap.error('--out 必填 (除非用 --validate 只校验)')

    build_html = _import_build_html()
    html = build_html(case_data)
    args.out.parent.mkdir(parents=True, exist_ok=True)
    # newline='' 保留 build_html 返回串里的换行原貌, 防 Windows 把 \n 翻译成 \r\n
    args.out.write_text(html, encoding='utf-8', newline='')
    n_procs = len(case_data.get('procedures', []))
    print(f'render-case: ✓ wrote {args.out} ({len(html):,} chars, '
          f'{n_procs} procedure{"s" if n_procs > 1 else ""}, {_count_steps(case_data)} steps)')


if __name__ == '__main__':
    main()