Agent/examples/process_pipeline/script/procedure-dsl/spec/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 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/focus 仅非 block 步要 (block 是控制容器)
    step_required = ['id', 'kind', 'via', 'feature', 'inputs', 'outputs', 'intent']
    exec_only = ['effect', 'action', 'focus']      # kind != block 才要
    io_required = ['type', 'name', '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


# =============================================================================
# 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='跳过校验直接渲染 (不推荐)')
    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)
        if 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()