#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ lint-case.py — workflow.json 轻量 lint + 自动 record 新 type 到 type_suggestions.md. 设计哲学: **不严格**. - 不分 error/warning 等级, 不卡 exit code (都返 0) - 主要副作用是 record 新 type 到 spec/taxonomy/type_suggestions.md - 检测项打 stdout 给 Agent / 用户看, 决定要不要回去修 用法: python spec/tools/lint-case.py --workflow outputs/case-{N}/workflow.json --case-id {N} python spec/tools/lint-case.py --workflow outputs/case-{N}/workflow.json --case-id {N} --no-record # 只校验不写 退出码: 0 始终 (不阻塞流程) 2 CLI 参数错误 / 文件不存在 """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path # spec/tools/lint-case.py → procedure-dsl/ DSL_ROOT = Path(__file__).resolve().parent.parent.parent TYPE_JSON = DSL_ROOT / 'spec' / 'taxonomy' / 'type.json' SUGGESTIONS = DSL_ROOT / 'spec' / 'taxonomy' / 'type_suggestions.md' # Windows 控制台 UTF-8 for _s in (sys.stdout, sys.stderr): if hasattr(_s, 'reconfigure'): try: _s.reconfigure(encoding='utf-8', errors='replace') except Exception: pass def load_type_leaves() -> set[str]: """读 spec/taxonomy/type.json 的 $leaves 集合.""" if not TYPE_JSON.exists(): return set() return set(json.loads(TYPE_JSON.read_text(encoding='utf-8')).get('$leaves', [])) # =========================================================================== # Check 1: type 完整性提示 # =========================================================================== def _iter_procedures(case_data: dict): """遍历 workflow.json 的 procedures. Yields: (procedure_label, procedure_dict) — 含 steps + type_registry. """ for p in case_data.get('procedures') or []: label = p.get('id') or p.get('name') or '?' yield (label, p) def check_type_completeness(case_data: dict) -> list[str]: """IO 用了 case-specific type 但 type_registry 漏写 entry → 提示. 只 hint 不 fail. Agent 看输出回去修. 多工序时, hint 前缀加 [proc_id] 让用户知道是哪个工序的问题. """ leaves = load_type_leaves() hints: list[str] = [] for proc_label, proc in _iter_procedures(case_data): type_reg = proc.get('type_registry') or {} for i, step in enumerate(proc.get('steps') or []): if not isinstance(step, dict): continue for kind in ('inputs', 'outputs'): for j, item in enumerate(step.get(kind) or []): if not isinstance(item, dict): continue t = item.get('type', '') or '' if not t: continue if t in leaves: continue # 字典叶子, OK if t not in type_reg: hints.append( f"[{proc_label}] step[{i}].{kind}[{j}].type={t!r} 是 case-specific " f"但 type_registry 没注册" ) else: entry = type_reg[t] if isinstance(entry, dict): if not entry.get('extends'): hints.append(f"[{proc_label}] type_registry[{t!r}] 缺 extends 字段") if not entry.get('desc'): hints.append(f"[{proc_label}] type_registry[{t!r}] 缺 desc 字段 (renderer drawer 显示需要)") return hints # =========================================================================== # Check 2: value / directive 自包含性 (禁止引用占位) # =========================================================================== # value/directive 应填数据本身, 不是 anchor 的引用. 命中即「没真正回填」. META_REF = re.compile(r'[（(]?\s*同\s*s[\d]|见\s*s[\d]|←\s*s[\d]|同上') def check_value_selfcontained(case_data: dict) -> list[str]: """扫每个 IO 的 value + 每个 directive, 找「引用占位」文案 (同 sX / 见 sX / ← sX ...). spec: value 逐字回填数据本身, 引用归 anchor (workflow-format.md §2 数据流). 这种占位 schema/type 检查抓不到, 专门一条. 只 hint. 可用 `wf-patch.py --resolve-passthrough` 自动从源回填. """ hints: list[str] = [] for proc_label, proc in _iter_procedures(case_data): for i, step in enumerate(proc.get('steps') or []): if not isinstance(step, dict): continue for kind in ('inputs', 'outputs'): for j, item in enumerate(step.get(kind) or []): if not isinstance(item, dict): continue v = item.get('value') if isinstance(v, str) and META_REF.search(v): hints.append(f"[{proc_label}] step[{i}].{kind}[{j}].value 是引用占位 {v[:24]!r} — 应逐字回填数据本身") t = step.get('directive') if isinstance(t, str) and META_REF.search(t): hints.append(f"[{proc_label}] step[{i}].directive 是引用占位 {t[:24]!r} — 应填实际 prompt 原文") return hints # =========================================================================== # Check 3: value 占位 / directive 缺失 (提示用 quote-source 回填真内容) # =========================================================================== # 纯 <...> 占位 (value 该填真实内容, <...> 仅限无文字的图/视频) PLACEHOLDER_RE = re.compile(r'^\s*<[^>]*>\s*$') # 模态分类关键词 (TEXT 优先于 MEDIA, 因 "配音文案" 这类既含媒体词又是文本) _TEXT_KW = ('提示词', '描述', '参数', '评', '大纲', '脚本', '文案', '歌词', '字幕', '标题', '正文', '词', '知识', '工作流', '对标', '规格', '批处理', '模板', '版式', '数据', '分析', '报告', '记录', '方案', '思路', '设定', '依据', '标准', '清单', '列表', '文本', '文字') _MEDIA_KW = ('图', '视频', '音频', '帧', '片段', '截图', '蒙版', '音效', '配音', 'BGM', '数字人', '滤镜', '海报', '封面') def _type_modality(type_name: str, type_reg: dict) -> str: """按类型名(case-specific 类型先经 type_registry.extends 解析到 stdlib 叶子)判模态. 返回 'media' (图/视频/音频 — 可 <描述>) / 'text' (提示词/数据/报告 — 必须真实文本) / 'unknown'. media 用关键词可靠识别; 非 media 一律按"需真实文本"对待 (data/text 占多数, 宁严勿漏). """ 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_content(case_data: dict) -> list[str]: """逐 IO 按模态审计 value + 工具步骤 directive → 提示用 quote-source 回填真内容. 规则 (phase1 §value): 文本类 IO(提示词/数据/报告)的 value 必须是从原文匹配到的真实内容, 不能写 <…> 占位; 原文确实没有 → 标 inferred:true + inferred_reason 显式说明 (本检查放行). 媒体类 IO(图/视频/音频)允许 <具体描述>. 工具步骤(via 是具体工具)必须带原文那段 prompt 当 directive. 弱模型常把所有 value 写成 <…> 占位、整个漏 directive (实测 test-7 全踩), 这条逐 IO 抓, 只 hint. """ hints: list[str] = [] for proc_label, proc in _iter_procedures(case_data): type_reg = proc.get('type_registry') or {} for i, step in enumerate(proc.get('steps') or []): if not isinstance(step, dict): continue for kind in ('inputs', 'outputs'): for j, item in enumerate(step.get(kind) or []): if not isinstance(item, dict): continue if item.get('inferred'): # 已显式标 inferred 说明 → 放行 continue v = item.get('value') if not isinstance(v, str): continue if re.search(r'原文(未提供|未给出|没有|无)', v): continue # 显式标「原文未提供」→ 放行 (LLM 确认原文确无) if not PLACEHOLDER_RE.match(v): continue # value 不是 <…> 占位 (已填真内容) t = item.get('type', '') or '' mod = _type_modality(t, type_reg) if mod == 'media': continue # 图/视频/音频 用 <描述> 合理 label = '文本类' if mod == 'text' else '非媒体(疑似数据/文本)' # 输出占位 = 步骤产出物没回填; 原文/OCR 里通常紧跟在 prompt 后展示了它 extra = ('；这是步骤**产出物**, 原文/配图 OCR 里常紧跟 prompt 展示了它, ' '用 quote-source --from/--to 把那段产出也捞进 value') if kind == 'outputs' else '' hints.append( f"[{proc_label}] step[{i}].{kind}[{j}] type={t!r}({label}) value={v.strip()!r} 仍是占位 " f"—— 你即便已 quote 到原文也**必须把真实内容替换进 value**(别只填 directive){extra}; " f"原文确无则标 inferred:true + inferred_reason; 若其实是无文字图/视频, 让类型/描述体现" ) via = (step.get('via') or '').strip() directive = (step.get('directive') or '').strip() if step.get('kind', 'step') == 'step' and via and via not in ('human', '-') and not directive: hints.append( f"[{proc_label}] step[{i}](via={via!r}) directive 空 — 若原文有给工具的提示词/指令, " f"用 quote-source 捞原文那段填进 directive" ) # substance/form 缺失 (Phase 2 该提炼实质/形式; 纯技术步可显式设 null, 但别整个漏掉 key) if step.get('kind', 'step') in ('step', 'nested'): miss = [f for f in ('substance', 'form') if f not in step] if miss: hints.append( f"[{proc_label}] step[{i}] 缺 {'/'.join(miss)} — Phase 2 漏做了实质/形式提炼; " f"读懂这步内容提炼元素点填上(纯技术步可显式设 null, 但别漏掉字段)" ) # intent 缺失 (Phase 2 每步都要填目的列, 一句话概括) if step.get('kind', 'step') in ('step', 'block', 'nested') and not (step.get('intent') or '').strip(): hints.append( f"[{proc_label}] step[{i}] 缺 intent — Phase 2 每步都要填目的列(一句话概括这步在做什么, ≤25 字)" ) return hints # =========================================================================== # Check 4: 章节覆盖 (结构强制 — 需 --source) + value 逐字 (值强制 — 需 --source) # =========================================================================== # # 弱模型在 Phase 1 骨架阶段走两条最省力的路, 都靠"看原文"才抓得到: # (结构) 只挑两个最显眼的工序就收工, 整段章节(框架/附加案例/总结)漏抽 # (值) 挑中的 value 也打字缩写成标题纲要, 不是逐字原文 (能过 render 门禁因为不是 <占位>) # 这两条 check 都需要原文 (--source input/case-N.json [--ocr ocr.txt]) 才能比对. # 比对噪声: 空白 + 各式引号 (原文 “”、骨架常写成 「」/""，内容一致只是引号风格不同, 不该算缩写) _QUOTE_NOISE = dict.fromkeys(map(ord, '「」『』“”‘’"\'＂＇'), None) def _norm(s: str) -> str: """归一化用于子串比对: 去所有空白 (原文常把一个词拆到两行) + 抹掉引号风格差异.""" return re.sub(r'\s+', '', s or '').translate(_QUOTE_NOISE) def _load_source_corpus(source_path: Path | None, ocr_path: Path | None) -> tuple[str, str]: """读原文语料: 返回 (raw_text, normed). raw 用来切章节, normed 用来子串比对. source = input/case-N.json 的 title + body_text; ocr = 配图 OCR 文本 (可选). """ parts: list[str] = [] if source_path and source_path.exists(): try: sd = json.loads(source_path.read_text(encoding='utf-8')) parts.append(sd.get('title', '') or '') parts.append(sd.get('body_text', '') or sd.get('content', '') or '') except Exception: parts.append(source_path.read_text(encoding='utf-8')) if ocr_path and ocr_path.exists(): parts.append(ocr_path.read_text(encoding='utf-8')) raw = '\n'.join(parts) return raw, _norm(raw) def _sections(body: str) -> list[tuple[str, str, str]]: """切原文章节: 按行首 `0N |` 标号 (排除 `图 0N |` 配图说明). 返回 [(号, 标题, 正文段)].""" marks = [(m.start(), m.group(1)) for m in re.finditer(r'(?m)^\s*(0\d)\s*[|｜]', body)] out: list[tuple[str, str, str]] = [] for idx, (pos, num) in enumerate(marks): end = marks[idx + 1][0] if idx + 1 < len(marks) else len(body) seg = body[pos:end] after = re.split(r'[|｜]', seg, 1) tail = after[-1] if len(after) > 1 else seg title = '' for line in tail.splitlines(): line = line.strip() if line: title = line[:24] break out.append((num, title, seg)) return out # 章节正文里的"要点标记": 思路X / 第X层 / 第X步 / 案例X / 冒号短标签 (人物特征：…) _POINT_MARKER = re.compile( r'(?m)^\s*(思路[一二三四五]|第[一二三四五六七八九十]+[层步]|案例[一二三四五六七八九十]+)') _POINT_COLON = re.compile(r'(?m)^\s*([^\n：:（(]{2,12})\s*[：:]') def _section_points(seg: str) -> list[str]: """抽一节正文的要点短语 (用来量化它被骨架覆盖了多少).""" pts: list[str] = [] for m in _POINT_MARKER.finditer(seg): line = seg[m.start():].splitlines()[0].strip() pts.append(line[:16]) for m in _POINT_COLON.finditer(seg): lab = m.group(1).strip() # 纯序号标记 (第X步/第X层/思路X/案例X) 是结构序号不是内容要点, 骨架改写成动作后必然对不上 → 跳过 if re.fullmatch(r'(思路[一二三四五]|第[一二三四五六七八九十]+[层步]|案例[一二三四五六七八九十]+)', lab): continue if re.search(r'[一-龥]', lab): pts.append(lab) seen: set[str] = set() out: list[str] = [] for p in pts: if p and p not in seen: seen.add(p) out.append(p) return out def _point_covered(point: str, wf_norm: str) -> bool: """要点是否被骨架覆盖: 去掉结构前缀后, 任一 4-gram 命中 workflow 文本即算覆盖 (从宽).""" core = re.sub(r'^(思路[一二三四五]|第[一二三四五六七八九十]+[层步]|案例[一二三四五六七八九十]+)', '', point) core = _norm(core) or _norm(point) if len(core) < 4: return core in wf_norm return any(core[k:k + 4] in wf_norm for k in range(len(core) - 3)) def check_section_coverage(case_data: dict, source_raw: str, wf_norm: str) -> list[str]: """结构强制: 逐章节算骨架覆盖率, 整段漏抽的章节 (<40%) 报出来 + 给缺失要点样例.""" hints: list[str] = [] secs = _sections(source_raw) if not secs: return hints for num, title, seg in secs: pts = _section_points(seg) if len(pts) < 2: continue # 没足够要点 (纯过渡/口号段), 不评判 missed = [p for p in pts if not _point_covered(p, wf_norm)] ratio = 1 - len(missed) / len(pts) if ratio < 0.40: sample = '、'.join(missed[:5]) hints.append( f"章节『{num} {title}』覆盖率 {ratio:.0%} ({len(pts) - len(missed)}/{len(pts)} 要点) " f"—— 疑似整段漏抽; 缺: {sample}{' …' if len(missed) > 5 else ''}; " f"回去为它补 procedure/step (每个 0N 章节至少对应一个工序或子步骤)" ) return hints def _longest_run(v_norm: str, source_norm: str) -> int: """value 在原文里能连续命中的最长子串长度. 逐字原文应是原文一整段连续文本; 拼接/缩写出来的(把分散的小标题用、串起来)最长连续命中会很短.""" n = len(v_norm) best = 0 for i in range(n): if n - i <= best: break # 剩余长度已不可能超过 best lo, hi = 0, n - i while lo < hi: # 二分该起点能命中的最长长度 mid = (lo + hi + 1) // 2 if v_norm[i:i + mid] in source_norm: lo = mid else: hi = mid - 1 if lo > best: best = lo return best def check_value_verbatim(case_data: dict, source_norm: str) -> list[str]: """值强制: 文本类 value 必须是原文里的「一整段连续文本」. 最长连续命中 <80% 判缩写/改写/截断. 跳过: 占位<…>(归 check3)、inferred、原文未提供、未 resolve 的 @quote、媒体类、短值(<12字). 用最长连续命中而非逐子句覆盖: 后者会被「人物、产品、环境」这种"原文小标题拼盘"骗过 (每个词单独在原文里, 但整体不是任何一段原文 — 真正的逐字细节全被丢了). """ hints: list[str] = [] if not source_norm: return hints for proc_label, proc in _iter_procedures(case_data): type_reg = proc.get('type_registry') or {} for i, step in enumerate(proc.get('steps') or []): if not isinstance(step, dict): continue for kind in ('inputs', 'outputs'): for j, item in enumerate(step.get(kind) or []): if not isinstance(item, dict): continue if item.get('inferred'): continue v = item.get('value') if not isinstance(v, str) or not v.strip(): continue if PLACEHOLDER_RE.match(v): continue # 占位 → check3 管 if v.startswith('@quote'): continue # 未回填的 quote, resolve 后才比 if re.search(r'原文(未提供|未给出|没有|无)', v): continue if _type_modality(item.get('type', '') or '', type_reg) == 'media': continue # 媒体描述不要求逐字 vn = _norm(v) if len(vn) < 12: continue # 短标签不查 (无所谓缩写) run = _longest_run(vn, source_norm) ratio = run / len(vn) # 绝对护栏: 连续命中 ≥80 字 = 铁证级真引用 (没人会"凑巧"逐字打 80 字), # 哪怕整体比例因中途一处微小偏差掉到 80% 也放行, 只抓"短值缩写"。 if run >= 80: continue if ratio < 0.80: hints.append( f"[{proc_label}] step[{i}].{kind}[{j}] value 最长连续命中原文仅 {run}/{len(vn)} 字" f"({ratio:.0%}) —— 整体不是一整段原文(疑似开头逐字后就缩写/改写); value={v[:40]!r}…; " f"用 @quote|起锚|止锚 + wf-patch --resolve-quotes 把整段原文逐字拽进来(原文那段提示词约 350 字)" ) return hints # =========================================================================== # Side effect: record 新 type 到 type_suggestions.md # =========================================================================== def record_new_types(case_data: dict, suggestions_path: Path = SUGGESTIONS) -> list[str]: """把 case_data.type_registry 里的 case-specific type append 到 suggestions. 幂等: 同一 (type_name, case_id) 二元组只 append 一次. Dedup 靠 grep 现有文件, 抽 `(来自 case-{N})` + 类型名 二元组. Returns: 本次新写入的条目 list (空 list = 没新东西要 record). """ # 合并所有 procedures.type_registry type_reg: dict = {} for p in case_data.get('procedures', []): type_reg.update(p.get('type_registry') or {}) if not type_reg: return [] leaves = load_type_leaves() case_id = case_data.get('case_id') or '?' text = suggestions_path.read_text(encoding='utf-8') if suggestions_path.exists() else '' # 已 record 过的 (type_name, case_id) — 用 regex 抓 markdown list entry existing = set(re.findall( r'^- `([^`]+)`:.*?\(来自 case-([^,)\s]+)', text, re.M )) new_lines: list[str] = [] for tname, entry in type_reg.items(): if not isinstance(entry, dict): continue if tname in leaves: continue # 已是字典叶子, 不是新 type (Agent 误把 stdlib type 加进 case_data.type_registry) if (tname, str(case_id)) in existing: continue ext = entry.get('extends', '?') desc = entry.get('desc') or '(无 desc)' new_lines.append(f'- `{tname}`: {desc} (来自 case-{case_id}, extends `{ext}`)') if new_lines: # 确保 suggestions 文件存在 (没有就建个空骨架) if not suggestions_path.exists(): suggestions_path.write_text( '# Type 字典扩展建议\n\n## 累积条目\n\n', encoding='utf-8' ) # append 末尾 with suggestions_path.open('a', encoding='utf-8') as f: f.write('\n' + '\n'.join(new_lines) + '\n') return new_lines # =========================================================================== # main # =========================================================================== def main() -> None: ap = argparse.ArgumentParser( prog='lint-case.py', description='workflow 轻量 lint + 自动 record 新 type 到 type_suggestions.md', ) ap.add_argument('--workflow', type=Path, required=True, help='workflow.json (含 procedures 数组). lint 内部读 procedures + type_registry') ap.add_argument('--case-id', type=str, default=None, help='record suggestions 用的 case_id. 不传就 fallback workflow.case_id 或 ?') ap.add_argument('--source', type=Path, default=None, help='原文 input/case-N.json. 传了才启用「章节覆盖」+「value 逐字」两条结构/值强制校验') ap.add_argument('--ocr', type=Path, default=None, help='配图 OCR 文本 (可选). 并入原文语料, 让逐字校验也认配图里的文字') ap.add_argument('--no-record', action='store_true', help='只校验, 不写 suggestions') args = ap.parse_args() target_path = args.workflow if not target_path.exists(): print(f'lint-case: 文件不存在 {target_path}', file=sys.stderr) sys.exit(2) try: case_data = json.loads(target_path.read_text(encoding='utf-8')) except json.JSONDecodeError as e: print(f'lint-case: {target_path} 不是合法 JSON: {e}', file=sys.stderr) sys.exit(2) # workflow 模式: 注入 case_id (suggestions record 需要) if args.case_id is not None and 'case_id' not in case_data: try: case_data['case_id'] = int(args.case_id) except ValueError: case_data['case_id'] = args.case_id case_id = case_data.get('case_id', '?') print(f'[lint] case-{case_id} ({target_path.name})') # check: type 完整性提示 hints = check_type_completeness(case_data) if hints: print(f' · type 完整性: {len(hints)} 个提示') for h in hints: print(f' - {h}') else: print(' · type 完整性: OK') # check: value / directive 自包含性 (引用占位) vhints = check_value_selfcontained(case_data) if vhints: print(f' · value 自包含: {len(vhints)} 个引用占位 (跑 wf-patch.py --resolve-passthrough 自动回填)') for h in vhints: print(f' - {h}') else: print(' · value 自包含: OK') # check: value 占位 / directive 缺失 (用 quote-source 回填真内容) chints = check_placeholder_content(case_data) if chints: print(f' · value/directive 真实性: {len(chints)} 处占位/缺失 (用 quote-source.py 从原文/配图 OCR 捞真内容回填)') for h in chints: print(f' - {h}') else: print(' · value/directive 真实性: OK') # check: 章节覆盖 + value 逐字 (结构/值强制 — 需 --source) if args.source is not None: source_raw, source_norm = _load_source_corpus(args.source, args.ocr) wf_norm = _norm(json.dumps(case_data, ensure_ascii=False)) shints = check_section_coverage(case_data, source_raw, wf_norm) if shints: print(f' · 章节覆盖(结构强制): {len(shints)} 个章节疑似漏抽 —— 骨架要覆盖原文每个章节') for h in shints: print(f' - {h}') else: print(' · 章节覆盖(结构强制): OK (原文各章节都有对应工序)') bhints = check_value_verbatim(case_data, source_norm) if bhints: print(f' · value 逐字(值强制): {len(bhints)} 处疑似缩写/改写 —— 文本类 value 要逐字搬原文(用 @quote)') for h in bhints: print(f' - {h}') else: print(' · value 逐字(值强制): OK (文本类 value 都逐字命中原文)') else: print(' · 章节覆盖 + value 逐字: skipped (传 --source input/case-N.json [--ocr ocr.txt] 启用结构/值强制)') # side effect: record 新 type if not args.no_record: new_lines = record_new_types(case_data) if new_lines: print(f' · 已 record {len(new_lines)} 条新 type 到 {SUGGESTIONS.name}:') for ln in new_lines: print(f' {ln}') else: # 合并所有 procedure type_registry merged_reg: dict = {} for p in case_data.get('procedures', []): merged_reg.update(p.get('type_registry') or {}) if not merged_reg: print(' · 无新 type 可 record (type_registry 为空 — 全部 type 命中字典叶子)') else: print(' · 无新 type 可 record (type_registry 里的项已全部 record 过)') # 不卡 exit code sys.exit(0) if __name__ == '__main__': main()