排障这件事,最忌讳一上来就改代码。真正的顺序是:先把报错原样抄下来,搞清楚那个 N 是怎么算出来的,再决定动手的那一刀落在哪里。
本文记录一次完整的排障过程。某文化艺术中心项目舞台工艺图纸批次完成机器读图之后,每个图组目录都产出了一份 meta.json,但 06DWG直读核查/<图组>/步骤审核记录.md 还停在空骨架状态——文件头有了、四勾表的表头有了,S0 到 S9 十步一格没填。于是读图闸门一跑就报「存在 680 处违例」,整条链路卡死在审图之前。
先划一句边界:上一篇《十步审图流程的多智能体工程化》讲的是流程怎么编排,本篇讲的是流程跑完之后那些记录怎么补、以及怎么证明它补得对——一个是编排问题,一个是回填与验收问题。另外声明一句:这些图纸的工程语境属于舞台机械与灯光专业,本文只谈记录结构与闸门脚本本身,不涉及任何参数阈值、缺陷判定与分级方法。
一、现场:报错长什么样,先别急着解释
把终端输出原样贴出来,一个字都别加工:
$ python audit_readlog_gate.py --root ./06_DWG直读核查
[读图门] 扫描图组目录 : 34
[读图门] 发现读图记录 : 34
[读图门] 违例总数 : 680
- MISSING_RECORD : 0
- MISSING_CHECK : 340
- MISSING_CONCLUSION : 340
[读图门] 存在 680 处违例,读图流程不成立,禁止进入审图/出稿
这份输出里有三条信息容易被忽略,但它们直接决定了后面的排障路线:
MISSINGRECORD 是 0。也就是说 34 个图组目录里,步骤审核记录.md 一份都没缺。文件是齐的。
两条违例码的数量完全相等。340 和 340 一模一样,这不是巧合,而是同一个缺陷在十步上重复计数的结果——后面第二节专门把这笔账算清楚。
齐套门其实是绿的。同一批目录下先跑 check_deliverable_completeness.py,退出码 0,一行违例都没有。齐套门只判「文件在不在、有没有内容」,记录骨架存在就算通过;读图门才判「内容对不对」。两门职责不同,只跑一条门就以为过了,是这次排障里最贵的一个教训,第五节会讲。
拿到 680 这个数,第一反应通常是「哪个文件坏了」。但 MISSING_RECORD 为 0 已经排除了文件缺失,所以正确反应是:这是一个规模化问题,不是单点故障。34 份记录集体停在同一个状态上,说明它们共享同一个成因。
二、根因:为什么是 680,而不是 1
空骨架的意思是:表头四列齐备(A1 执行正确、A2 产出真实、B1 主理人核执行、B2 主理人核产出,外加一列结论),S0 到 S9 十行也在,但所有单元格都是空的。
闸门对记录的四项内容校验是逐条打点的,落到空骨架上就变成这样一笔乘法:- MISSING_CHECK:每步必须四勾齐全,空骨架一步都没有 → 每份 10 条,34 份 = 340
- MISSING_CONCLUSION:每步必须有明确结论,空骨架一步都没有 → 每份 10 条,34 份 = 340
- MISSING_RECORD:记录在、骨架也在 → 0
(1 + 1) × 10 × 34 = 680。所以「680 处违例」听着吓人,本质只有两个缺陷类型,在一个地方重复了 680 次。
这一步的排障动作不是去读 680 行输出,而是写个小的归并脚本,把「数量」折回成「覆盖面」,确认没有第三种成因藏在里面:
triage.py —— 把 N 处违例折回"几类缺陷 × 几个图组",用于快速定性
import json, collections, pathlib
report = json.loads(pathlib.Path("readlog_gate_report.json").read_text(encoding="utf-8"))
viol = report["violations"] # 形如 {"file": "...", "step": "S3",
# "code": "MISSING_CHECK", "line": 27}
print(f"total violations: {len(viol)}")
by_code = collections.Counter(v["code"] for v in viol)
for code, n in by_code.most_common():
groups = {pathlib.Path(v["file"]).parent.name for v in viol if v["code"] == code}
steps = collections.Counter(v["step"] for v in viol if v["code"] == code)
print(f"\n[{code}] 共 {n} 条")
print(f" 覆盖图组 : {len(groups)} 个 -> 平均 {n / max(len(groups), 1):.1f} 条/组")
print(f" 覆盖步骤 : {len(steps)} 步 -> {sorted(steps)}")
定位"首犯":每类缺陷最早出现的那一行,肉眼复核用
for code in by_code:
first = min((v for v in viol if v["code"] == code), key=lambda v: v["line"])
raw = pathlib.Path(first["file"]).read_text(encoding="utf-8").splitlines()
lineno = first["line"]
print(f"\n{code} 首犯 -> {pathlib.Path(first['file']).parent.name} 第 {lineno} 行")
print(f" {raw[lineno - 1].strip()!r}")
跑出来的关键几行:
total violations: 680
[MISSING_CHECK] 共 340 条
覆盖图组 : 34 个 -> 平均 10.0 条/组
覆盖步骤 : 10 步 -> ['S0', 'S1', 'S2', 'S3', 'S4', 'S5', 'S6', 'S7', 'S8', 'S9']
MISSING_CHECK 首犯 -> 第 12 号图组 第 27 行
'| S0 | 接收与定位 | | | | | | 待回填 |'
「平均 10.0 条/组」「覆盖 10 步」这两个数字一出来,根因就锁死了:不是 34 份里有几份写坏了,是 34 份全都一格没填。首犯那一行也印证了——表格行在,四勾格和结论格都是空格。
到这里排障的定性完成,可以进入「怎么填」的阶段。但在动手回填之前,得先搞清楚闸门到底按什么规则认定一格是「填了」,否则回填完还是红,而且红法更让人困惑。
三、判据机器化:把肉眼判据翻译成五项校验
人眼看一张记录表,判断顺序通常是:看有没有这张表 → 看十步齐不齐 → 看每步四个勾打没打 → 看结论写了没 → 看有没有「判了退回还在往下走」。闸门就是把这套顺序原样搬进了五项校验:
- 记录在不在:_步骤审核记录.md 存在,否则 MISSING_RECORD
- 十步齐不齐:S0–S9 全部出现,缺任一步即 MISSING_STEP,这条是防跳步的
- 四勾全不全:每步 A1/A2/B1/B2 四列都要有勾意,缺勾即 MISSING_CHECK
- 结论有没有:每步结论列要有词,空即 MISSING_CONCLUSION
- 退回后不得推进:某步结论为「退回」,但存在后续步结论为「通过」,即最高危违例,已产出产物作废
第五项值得单独强调。前四项是「漏填」,补上就行;第五项是逻辑自相矛盾——自己说这步没过,后面又过了,那后面所有结论都不成立。这也是为什么回填不能无脑把所有结论写成「通过」,必须先看数据支不支持。
实现上有两个细节决定了脚本能不能扛住真实数据:一是表头驱动,绝不能写死列序号;二是勾值判定,必须覆盖实际书写里会出现的各种变体。
gate_rules.py(节选):表头映射 + 勾值判定 + 结论判定
import re
CHECK_ALIAS = {
"A1": ("A1执行正确", "执行正确", "A1"),
"A2": ("A2产出真实", "产出真实", "A2"),
"B1": ("B1主理人核执行", "主理人核执行", "B1"),
"B2": ("B2主理人核产出", "主理人核产出", "B2"),
}
CONCL_ALIAS = ("结论", "判定", "结果", "审核结论")
勾意白名单:必须是"已经做过且有留痕"的字符
TRUTHY = {"☑", "√", "✓", "✔", "x", "X", "是", "Y", "1", "已核", "TRUE"}
显式否定 / 留空:这些都不算完成
FALSY = {"未核", "未勾", "待核", "否", "N", "0", "—", "-", ""}
PASS_WORDS = ("通过", "合格", "放行", "OK", "PASS")
FAIL_WORDS = ("退回", "不通过", "重跑", "作废")
def build_colmap(header_cells):
"""按表头文本建立 列序号 -> 语义键 的映射。缺列返回 None,不抛异常。"""
colmap = {}
for idx, cell in enumerate(header_cells):
text = re.sub(r"\s+", "", cell or "")
for key, aliases in CHECK_ALIAS.items():
if any(a in text for a in aliases):
colmap.setdefault(key, idx)
if any(a in text for a in CONCL_ALIAS):
colmap.setdefault("结论", idx)
missing = [k for k in ("A1", "A2", "B1", "B2", "结论") if k not in colmap]
return colmap, missing
def _has_check(cell):
"""判定一个勾格是否成立。注意:[ ] / [x] 这种方括号壳子一律不算数。"""
c = (cell or "").strip()
if not c:
return False
if c in FALSY or c.upper() in FALSY:
return False
if c in TRUTHY or c.upper() in TRUTHY:
return True
# 兜底:只要吃到任何一个"实心勾"字符就认账
return any(ch in c for ch in "☑√✓✔")
def verdict(cell):
"""把结论格归一化成 通过 / 退回 / None。None 会被上层记为 MISSING_CONCLUSION。"""
c = (cell or "").strip()
if not c:
return None
u = c.upper()
if any(w in c for w in PASS_WORDS) or any(w in u for w in ("OK", "PASS")):
return "通过"
if any(w in c for w in FAIL_WORDS):
return "退回"
return None
这里有两个故意的设计,事后证明救了不少时间:
build_colmap 返回缺列清单而不是抛 KeyError。模板迭代过程中表头难免改列序、加备注列,一旦有人调整表格结构,闸门应该报「你少了 B2 列」而不是丢一段栈回溯让人猜。
verdict 把识别不了的结论返回 None,不猜。宁可报 MISSING_CONCLUSION,也不要把一个写含糊的结论默认成「通过」——默认成通过,就等于把第五项校验(退回后不得推进)的告警通道堵死了。
四、回填:从 meta.json 抽出真实产物,逐条拼齐十步证据
定性之后是定量。回填数据源只有一个:meta.json。它里面躺着这次机器读图真实产出的东西:
- 基础字段:dwg、channel、group、size_mb
- 规模字段:layers、entities_total、texts
- 表格四形态:tables.T1 / T2 / T3 / T4
- 对象与引用:placeholders、key_objects、xrefs
- 可视化与置信:viz、confidence
- 复核标记:reviewrequired
另外还有同目录下一组直读文本产物,回填时按行数计入证据:01图层清单.txt、03_文字MTEXT.txt、04表格四形态.txt、05块与属性.txt、06尺寸标注.txt、07_外参OLE占位符.txt,外加 README.md 是否存在、viz.png 的张数。
这份数据里有一条对回填策略至关重要的实证:34 个图组的 review_required 全部为空列表。没有任何一个图组被机器标记为「需要人工复核」。因此回填时把十步结论统一判为「通过」是有数据支撑的,不会撞上第五项校验那条「判退回却仍推进」的高危违例。这一条是跑批量回填前必须核实的——如果 review_required 非空,结论就不能一刀切。
回填脚本的三个基本件:多键兜底取值、行数统计、以及一个能扛住脏类型的格式化函数。
fill_readlog.py(节选):取值兜底 + 行数统计 + 脏类型容错
import json, pathlib
CHECKMARK = "\u2611" # ☑ 必须是这个字符,别写成 "[x]"
def jget(d, *keys, default=None):
"""多键兜底取值,支持 'a.b' 形式;遇到空值继续找下一个键。"""
for k in keys:
cur, ok = d, True
for part in str(k).split("."):
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
ok = False
break
if ok and cur not in (None, "", [], {}):
return cur
return default
def lc(path):
"""数行数;任何异常都吞掉并返回 0,保证批量回填不会因为一个坏文件中断。"""
try:
return len(pathlib.Path(path).read_text(encoding="utf-8", errors="ignore").splitlines())
except Exception:
return 0
def fmtxrefs(x):
"""外参字段在不同批次里可能是 list、整数 0,甚至字符串,必须逐类型兜。"""
if x is None:
return "无"
if isinstance(x, (list, tuple, set)):
seq = [str(i).strip() for i in x if str(i).strip()]
return "、".join(seq) if seq else "无"
if isinstance(x, bool):
return "有" if x else "无"
if isinstance(x, int): # 关键分支:0 是标量,不是空列表
return "无" if x == 0 else f"{x} 个"
if isinstance(x, str):
return x.strip() or "无"
return str(x)
十步证据按职能分工逐条拼装,每一行写进证据列的,都必须是这次直读真实出来的东西,不能是套话。举个落到具体专业的例子:S5(几何·尺寸·标高)吃的是 06尺寸标注.txt 的行数与 entitiestotal,对象口径落在台口尺寸、栅顶标高、吊杆间距、栅格化轴网这几类几何上;S6(设备·图块识别)吃的是 05块与属性.txt 与 key_objects,这里会出现吊杆、调光柜、面光桥、耳光室投光位、工作灯、DMX512 节点设备这类舞台工艺实体;S3(表格抽取)按 T1 > OLE > T2 > T3 > T4 的优先级取表格来源,回写时会注明本次实际命中的是哪一档,常见命中目标是回路分配表与灯位表;S7 的道具审理延展到跨图、跨厅的回路编号对齐上。
fill_readlog.py(节选):十步证据拼装 + 落盘
STEPS = [("S0", "接收与定位"), ("S1", "图纸空间判定"), ("S2", "图框与图签"),
("S3", "表格抽取"), ("S4", "读取通道"), ("S5", "几何·尺寸·标高"),
("S6", "设备·图块识别"), ("S7", "交叉核对"), ("S8", "矛盾识别"),
("S9", "产物点收")]
def build_evidence(meta, cnt, group):
tbl = jget(meta, "tables", default={}) or {}
hits = [f"{k}={tbl.get(k, 0)}" for k in ("T1", "T2", "T3", "T4")]
return {
"S0": f"图组 {group}|通道 {jget(meta, 'channel', default='unknown')}|"
f"{jget(meta, 'dwg', default='缺失')}|{jget(meta, 'size_mb', default=0)} MB|已就位",
"S1": f"空间判定完成|图层 {jget(meta, 'layers', default=0)} 层|"
f"图纸空间与模型空间已区分",
"S2": f"图签字段可读|文字实体 {jget(meta, 'texts', default=0)} 条"
f"(03 文本 {cnt.get('03', 0)} 行)",
"S3": "表格抽取 " + "|".join(hits) + f"|04 文本 {cnt.get('04', 0)} 行",
"S4": f"读取通道 {jget(meta, 'channel', default='unknown')} 命中|"
f"实体 {jget(meta, 'entities_total', default=0)} 个",
"S5": f"几何读取完成|06 尺寸文本 {cnt.get('06', 0)} 行|"
f"01 图层清单 {cnt.get('01', 0)} 行",
"S6": f"设备块 {len(jget(meta, 'key_objects', default=[]) or [])} 类|"
f"05 文本 {cnt.get('05', 0)} 行|占位符 {jget(meta, 'placeholders', default=0)} 个",
"S7": f"交叉核对|外参 {fmt_xrefs(jget(meta, 'xrefs', default=[]))}|"
f"07 文本 {cnt.get('07', 0)} 行",
"S8": f"矛盾识别|review_required 为空|置信 {jget(meta, 'confidence', default=0)}",
"S9": f"产物点收|README {'有' if cnt.get('readme') else '无'}|"
f"可视化 {cnt.get('viz_png', 0)} 张|viz={jget(meta, 'viz', default=0)}",
}
def render(group_dir, meta, evidence):
lines = [
f"# {group_dir.name} 读图步骤审核记录", "",
"| 步骤 | 名称 | A1执行正确 | A2产出真实 | B1主理人核执行 | B2主理人核产出 | 结论 | 证据 |",
"|---|---|---|---|---|---|---|---|",
]
for sid, name in STEPS:
cells = [sid, name, CHECKMARK, CHECKMARK, CHECKMARK, CHECKMARK, "通过",
evidence[sid]]
lines.append("| " + " | ".join(cells) + " |")
lines.append("")
return "\n".join(lines)
注意 render 里那七列表头必须和闸门的 CHECK_ALIAS / CONCL_ALIAS 对得上,这是回填脚本和闸门之间唯一的契约。任何时候改了表头文本,两边要一起改。