176 lines
8.5 KiB
Python
176 lines
8.5 KiB
Python
#!/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 ' ✓'}")
|
||
# 尾部拼接残渣 + 前缀重复（批9 样张走查病灶：条目尾挂 ",/'] 等符号；
|
||
# 一条是另一条的严格前缀=同内容重复追加）——SQL 难表达，Python 循环扫
|
||
tail_re = re.compile(r"""(?:",|'\]|"\]|"}|'}|',)\s*$""")
|
||
pref_re = re.compile(r"^(?:\[[^\]]{1,12}\]\s*)+")
|
||
n_tail = n_pref = 0
|
||
for (fields,) in conn.execute(
|
||
"""SELECT d.draft_payload->'字段' FROM muse_knowledge_draft d
|
||
JOIN example_upgrade_card_state cs ON cs.draft_id=d.id
|
||
WHERE d.deleted=FALSE""").fetchall():
|
||
for v in (fields or {}).values():
|
||
if not isinstance(v, list):
|
||
continue
|
||
bodies = [pref_re.sub("", str(x)) for x in v]
|
||
n_tail += sum(1 for x in v if tail_re.search(str(x).rstrip()))
|
||
n_pref += sum(1 for i, a in enumerate(bodies) if len(a) > 80 and any(
|
||
j != i and b.startswith(a) and len(b) > len(a) for j, b in enumerate(bodies)))
|
||
lines.append(f" 升格卡尾部残渣：{n_tail}{' ⚠' if n_tail else ' ✓'}")
|
||
lines.append(f" 升格卡前缀重复：{n_pref}{' ⚠' if n_pref 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)
|