框架: 终态体检脚本固化+切窗收尾缝隙自检(批7实测50章断点悬空段根治)

2026-07-16 00:57:36 +08:00 · 2026-07-16 00:57:36 +08:00 · bdc9f15ee1
commit bdc9f15ee1
parent 42db159632
2 changed files with 167 additions and 0 deletions
--- a/.claude/skills/parse-book/scripts/parse_health.py
+++ b/.claude/skills/parse-book/scripts/parse_health.py
@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""parse-book skill：全链终态体检（零 LLM 机械扫描）。
+
+fable5 抽检优化项落地（创始人 2026-07-16）：抽检报告中近半发现可机械化——
+固化为体检脚本，每批跑完自动体检，AI 抽检只做机械查不了的判断题。
+覆盖四层：章级拆书 / 窗级大纲 / 范式出卡 / 作品面升格 + 全局健康红线。
+
+用法：health --work-id N（单书）或 health（全部书）
+"""
+import re
+import sys
+
+import click
+import psycopg
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parents[2] / "llm" / "scripts"))
+from parse_upgrade import DSN, TENANT, _is_garbage  # noqa: E402
+from parse_ingest import IP_LEAK_WORDS  # noqa: E402
+
+
+def sec_chapter(conn, wid):
+    """章级：总章 / 细纲状态分布。"""
+    total = conn.execute(
+        "SELECT count(*) FROM muse_content_chapter WHERE work_id=%s AND deleted=FALSE",
+        (wid,)).fetchone()[0]
+    st = dict(conn.execute(
+        """SELECT scaffold_status, count(*) FROM example_parse_task t
+           JOIN muse_content_chapter c ON c.id=t.chapter_id AND c.deleted=FALSE
+           WHERE t.work_id=%s GROUP BY 1""", (wid,)).fetchall())
+    done, failed = st.get("done", 0), st.get("failed", 0)
+    flag = "✓" if done >= total and not failed else ("⚠" if failed else "…")
+    return f"  章级 {flag}：{total} 章｜细纲 done {done} / failed {failed} / 其他 {total - done - failed}"
+
+
+def sec_outline(conn, wid):
+    """窗大纲：窗数 / 章覆盖缝隙 / 终检状态。"""
+    wins = conn.execute(
+        """SELECT from_order, to_order, check_status FROM example_parse_outline
+           WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE ORDER BY from_order""",
+        (TENANT, wid)).fetchall()
+    if not wins:
+        return "  窗大纲 …：0 窗"
+    gaps = [f"{a2}-{b2}章前有缝隙" for (_, b1, _), (a2, b2, _) in zip(wins, wins[1:]) if a2 != b1 + 1]
+    ck = {}
+    for _, _, s in wins:
+        ck[s or "pending"] = ck.get(s or "pending", 0) + 1
+    last = wins[-1][1]
+    total = conn.execute(
+        "SELECT max(order_no) FROM muse_content_chapter WHERE work_id=%s AND deleted=FALSE",
+        (wid,)).fetchone()[0]
+    tail = "" if last >= total else f"｜尾部 {last + 1}-{total} 章未切窗"
+    flag = "⚠" if gaps or tail else "✓"
+    return (f"  窗大纲 {flag}：{len(wins)} 窗（覆盖至 {last}/{total} 章）｜终检 {ck}"
+            + (f"｜缝隙 {gaps[:3]}" if gaps else "") + tail)
+
+
+def sec_cards(conn, wid, title):
+    """范式卡：按来源书统计五型分布（出处.书名匹配）。"""
+    rows = conn.execute(
+        """SELECT draft_payload->>'型', count(*) FROM muse_knowledge_draft
+           WHERE source_type='parse_book' AND deleted=FALSE
+             AND draft_payload->'出处'->>'书名'=%s GROUP BY 1 ORDER BY 2 DESC""",
+        (title,)).fetchall()
+    n = sum(r[1] for r in rows)
+    dist = " ".join(f"{t}{c}" for t, c in rows)
+    return f"  范式卡：{n} 张（{dist or '无'}）"
+
+
+def sec_upgrade(conn, wid):
+    """升格：窗态 / 卡数 / 摘要重刷 / 审计规模。"""
+    st = dict(conn.execute(
+        """SELECT status, count(*) FROM example_upgrade_window
+           WHERE work_id=%s AND deleted=FALSE GROUP BY 1""", (wid,)).fetchall())
+    if not st:
+        return "  升格 …：未启动"
+    n_card = conn.execute(
+        """SELECT count(*) FROM example_upgrade_card_state cs
+           JOIN muse_knowledge_draft d ON d.id=cs.draft_id
+           WHERE d.work_id=%s AND d.deleted=FALSE""", (wid,)).fetchone()[0]
+    n_sum = conn.execute(
+        """SELECT count(DISTINCT a.draft_id) FROM example_upgrade_audit a
+           JOIN muse_knowledge_draft d ON d.id=a.draft_id
+           WHERE d.work_id=%s AND a.field_name='清洗-摘要重刷'""", (wid,)).fetchone()[0]
+    total = sum(st.values())
+    flag = "✓" if st.get("done", 0) == total else ("⚠" if st.get("failed") else "…")
+    return (f"  升格 {flag}：窗 {st}｜实体卡 {n_card} 张｜摘要重刷 {n_sum} 卡")
+
+
+def sec_health(conn):
+    """全局健康红线（抽检机械化项）：结构垃圾/IP词/巨型粘连/单字别名/同名卡对。"""
+    lines = []
+    # 升格卡结构垃圾（字段值含变更包特征）
+    n = conn.execute(
+        """SELECT count(*) FROM muse_knowledge_draft d
+           JOIN example_upgrade_card_state cs ON cs.draft_id=d.id
+           WHERE d.deleted=FALSE AND (d.draft_payload->'字段')::text LIKE '%draft\\_id%'""").fetchone()[0]
+    lines.append(f"  升格卡结构垃圾残留：{n}{'  ⚠' if n else ' ✓'}")
+    # 升格卡巨型分号粘连
+    n = conn.execute(
+        r"""SELECT count(*) FROM muse_knowledge_draft d
+            JOIN example_upgrade_card_state cs ON cs.draft_id=d.id,
+            jsonb_each(d.draft_payload->'字段') f,
+            jsonb_array_elements_text(CASE WHEN jsonb_typeof(f.value)='array'
+                THEN f.value ELSE '[]'::jsonb END) e
+            WHERE d.deleted=FALSE AND e.value ~ '；\[窗'""").fetchone()[0]
+    lines.append(f"  升格卡巨型粘连残留：{n}{'  ⚠' if n else ' ✓'}")
+    # 范式卡 IP 词残留（名称/摘要/字段；实例与出处豁免）
+    n = conn.execute(
+        """SELECT count(*) FROM muse_knowledge_draft d
+           WHERE d.source_type='parse_book' AND d.deleted=FALSE
+             AND EXISTS (SELECT 1 FROM unnest(%s::text[]) w
+                 WHERE (d.draft_payload->>'名称') LIKE '%%'||w||'%%'
+                    OR (d.draft_payload->>'一句话摘要') LIKE '%%'||w||'%%'
+                    OR (d.draft_payload->'字段')::text LIKE '%%'||w||'%%')""",
+        (list(IP_LEAK_WORDS),)).fetchone()[0]
+    lines.append(f"  范式卡 IP 专名残留：{n}{'  ⚠' if n else ' ✓'}")
+    # 单字别名（预扫噪声源）
+    n = conn.execute(
+        "SELECT count(*) FROM example_upgrade_alias WHERE length(alias) < 2").fetchone()[0]
+    lines.append(f"  单字别名：{n}{'  ⚠' if n else ' ✓'}")
+    # 升格同名卡对（漏并候选）
+    rows = conn.execute(
+        """SELECT d.work_id, d.draft_payload->>'名称', count(*) FROM muse_knowledge_draft d
+           JOIN example_upgrade_card_state cs ON cs.draft_id=d.id
+           WHERE d.deleted=FALSE GROUP BY 1, 2 HAVING count(*) > 1""").fetchall()
+    lines.append(f"  升格同名卡对：{len(rows)}{'  ⚠ ' + str([r[1] for r in rows[:5]]) if rows else ' ✓'}")
+    return "\n".join(lines)
+
+
+@click.command()
+@click.option("--work-id", type=int, default=0, help="只体检指定书（0=全部）")
+def health(work_id):
+    with psycopg.connect(DSN) as conn:
+        works = conn.execute(
+            "SELECT id, title FROM muse_content_work WHERE deleted=FALSE"
+            + (" AND id=%s" % work_id if work_id else "") + " ORDER BY id").fetchall()
+        for wid, title in works:
+            n_ch = conn.execute(
+                "SELECT count(*) FROM muse_content_chapter WHERE work_id=%s AND deleted=FALSE",
+                (wid,)).fetchone()[0]
+            if not n_ch:
+                continue
+            click.echo(f"《{title}》(work={wid})")
+            click.echo(sec_chapter(conn, wid))
+            click.echo(sec_outline(conn, wid))
+            click.echo(sec_cards(conn, wid, title))
+            click.echo(sec_upgrade(conn, wid))
+        click.echo("―― 全局健康红线 ――")
+        click.echo(sec_health(conn))
+
+
+if __name__ == "__main__":
+    try:
+        health()
+    except psycopg.Error as e:
+        click.echo(f"[错误] {type(e).__name__}: {e}", err=True)
+        sys.exit(1)
--- a/.claude/skills/parse-book/scripts/parse_outline.py
+++ b/.claude/skills/parse-book/scripts/parse_outline.py
@ -126,6 +126,16 @@ def window(work_id, window, model, from_, to_):
            _do_window(conn, work_id, title, win_no, cur, model, done)
        elif cur:
            click.echo(f"尾段 {len(cur)} 章（{cur_len:,} 字）不足半窗且未到书末，留待与后续章合窗")
+        # 收尾缝隙自检（批7 实测：三本书都在 50 章分批断点处留了永久悬空段——
+        # "留待合窗"的尾段被下次续跑的 from 跳过。切完就报，别等体检才发现）
+        allw = conn.execute(
+            """SELECT from_order, to_order FROM example_parse_outline
+               WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE ORDER BY from_order""",
+            (TENANT, work_id)).fetchall()
+        gaps = [f"{b1 + 1}-{a2 - 1}" for (_, b1), (a2, _) in zip(allw, allw[1:]) if a2 > b1 + 1]
+        if gaps:
+            click.echo(f"⚠ 本书窗覆盖存在缝隙章段 {gaps}：用 --from/--to 锁定缝隙补切"
+                       f"（段太小不足半窗时把 --window 调小到段字数以下）")


 def _do_window(conn, work_id, title, win_no, chs, model, done):