46704ee5da
Bug1(parse_ingest _absorb):归并时用输家书名无条件覆写每条实例的来源书名→'二级归并母卡'里跨书实例的真实来源被抹成输家书名;改为保留实例原有书名(ins.get('书') or book)。已跑的718归并中75条实例/31母卡受影响,可从回溯归并审计的被并卡全文还原(待remediate)。
Bug2(parse_upgrade INLINE_CHAP_RE):裸'N章'分支把'隔3章/花了3章篇幅'数量词误当章号、还标最高置信'内嵌',违背人工精确化(拍板#5);改为只认'第X章'/'Ch.X'(区间取首章'第489-490章'→489),裸写法一律落待LLM重抽。迁移仅dry-run未落库、无数据影响。11条边界用例亲测通过。
独立opus子代理对抗复核确认无阻断性问题:无物理删知识卡/无dry-run漏写/无额度绕过/软删可逆/连接三段/调用契约完好。
738 lines
45 KiB
Python
738 lines
45 KiB
Python
#!/usr/bin/env python3
|
||
"""parse-book 配套确定性脚本：拆书产物校验入库 + 任务状态机（B2/B4-S3）。
|
||
|
||
职责边界：extractor(LLM) 只产结构化 JSON 文件，不碰库；本脚本做机械校验后写库——
|
||
字段 key 合法性（对库内字段合同）、五型归型、出处必填、**脱敏红线 15 连字检测**（硬阻断）。
|
||
状态全在库（example_parse_task），断点续跑与幂等按章/按窗。
|
||
|
||
B4-S3 起管线两级：scaffold（章级：细纲+实体+范式候选线索）→ cards（窗级：聚类母卡）。
|
||
本脚本仅在两个受控点调用 LLM/嵌入服务（不产内容）：
|
||
① 新卡嵌入（判重与检索基座共用）；② 相似度 ≥0.85 时的归并判定（输入两卡 JSON，输出 merge/keep）。
|
||
patterns 命令是 S3 前的章级出卡入口，保留作回滚保险，新管线不再使用。
|
||
"""
|
||
import hashlib
|
||
import json
|
||
import pathlib
|
||
import re
|
||
import sys
|
||
|
||
import click
|
||
import psycopg
|
||
from psycopg.types.json import Jsonb
|
||
|
||
# 受控点依赖：嵌入走 embed skill 同一实现（同模型同维），归并判定走 llm skill 统一入口
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "embed" / "scripts"))
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "llm" / "scripts"))
|
||
from embed_drafts import _session, build_embed_text, embed_texts # noqa: E402
|
||
from llm import chat_governed, extract_json # noqa: E402 # 归并判定走全局额度治理入口
|
||
|
||
DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
|
||
"?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
|
||
TENANT, ACTOR = 1, "1"
|
||
PATTERN_TYPES = {"craft", "combat", "emotion", "scene_pattern", "trope"} # 拍板①：首轮只拆五型
|
||
NGRAM = 15 # 脱敏红线：≥15 连续字与原文重合=违规（parse-book skill）
|
||
# 版权 IP 与系统专名词表（复抽实证：机战无限为高达系同人，IP 背景词不在实体名池）。
|
||
# 词形边界（opus 终检教训）：
|
||
# - "浮游炮"不入表——已泛化为通用武器品类词（同"光剑"），入表实测误伤 3 卡；
|
||
# - "战功"/"负能"不入表——通用词（立下战功/负能量）子串误伤面大，其系统义
|
||
# （机战货币"战功"、星环设定"负能"）由批次清洗按语境甄别；
|
||
# - 原创书自造宇宙观术语（伪造物主/幽畸/锐眼）同属泄漏，一并列管。
|
||
IP_LEAK_WORDS = ("GN粒子", "太阳炉", "扎古", "高达", "夏亚", "阿姆罗", "米诺夫斯基",
|
||
"脑量子波", "GN-Bit", "影印人", "殖装", "战功点", "宇宙世纪",
|
||
"爆种", "次元兽", "伪造物主", "幽畸", "锐眼")
|
||
EMBED_MODEL = "Qwen/Qwen3-Embedding-8B"
|
||
SIM_MERGE = 0.80 # 强候选线：≥该值＝高置信近义（前向修复后 0.75+ 均送 merge_judge，本值只作档位标注）
|
||
SIM_MARK = 0.75 # 候选线：≥该值即送 merge_judge 终判（cards 前向判重 + recluster 回溯聚类 同一判据）
|
||
|
||
|
||
def norm_scaffold(data: dict) -> dict:
|
||
"""键名归一：StructuredOutput 工具 schema 只允许 ASCII 键（API 硬约束 2026-07-13 实测），
|
||
库内 payload 统一中文键（供下游 prompt 消费）。两种键名都接受。"""
|
||
ents = []
|
||
for e in data.get("实体") or data.get("entities") or []:
|
||
ents.append({"型": e.get("型") or e.get("type"), "名称": e.get("名称") or e.get("name"),
|
||
"一句话摘要": e.get("一句话摘要") or e.get("brief")})
|
||
hints = []
|
||
for h in data.get("线索") or data.get("hints") or []:
|
||
hints.append({"型": h.get("型") or h.get("type"), "短名": h.get("短名") or h.get("name"),
|
||
"线索": h.get("线索") or h.get("clue"), "证据": h.get("证据") or h.get("evidence")})
|
||
return {"细纲": data.get("细纲") or data.get("outline") or "", "实体": ents, "线索": hints}
|
||
|
||
|
||
def norm_card(c: dict) -> dict:
|
||
"""范式卡键名归一（ASCII→中文）；窗级卡带实例数组 [{章,定位}]。"""
|
||
src = c.get("出处") or c.get("source") or {}
|
||
inss = []
|
||
for ins in c.get("实例") or c.get("instances") or []:
|
||
try:
|
||
ch_no = int(ins.get("章") or ins.get("ch"))
|
||
except (TypeError, ValueError):
|
||
continue # 章号非数字=编造，丢弃该实例
|
||
inss.append({"章": ch_no, "定位": ins.get("定位") or ins.get("anchor") or ""})
|
||
return {"型": c.get("型") or c.get("type"), "名称": c.get("名称") or c.get("name"),
|
||
"一句话摘要": c.get("一句话摘要") or c.get("brief"),
|
||
"字段": c.get("字段") or c.get("fields") or {},
|
||
"实例": inss,
|
||
"出处": {"书名": src.get("书名") or src.get("book"),
|
||
"回目": src.get("回目") or src.get("chapter"),
|
||
"定位": src.get("定位") or src.get("anchor")}}
|
||
|
||
|
||
def chapter_of(conn, work_id, order_no):
|
||
"""取章 id 与正文。"""
|
||
row = conn.execute(
|
||
"""SELECT c.id, b.content_text FROM muse_content_chapter c
|
||
JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=FALSE
|
||
WHERE c.tenant_id=%s AND c.work_id=%s AND c.order_no=%s AND c.deleted=FALSE""",
|
||
(TENANT, work_id, order_no)).fetchone()
|
||
if not row:
|
||
raise click.ClickException(f"章不存在: work={work_id} order={order_no}")
|
||
return row
|
||
|
||
|
||
def field_contract(conn, ttype):
|
||
"""库内字段合同：型 → 合法字段 key 集合。"""
|
||
rows = conn.execute(
|
||
"""SELECT f.field_key FROM muse_meta_field f
|
||
JOIN muse_meta_schema_version sv ON sv.id=f.schema_version_id
|
||
JOIN muse_meta_schema s ON s.active_version_id=sv.id
|
||
WHERE s.tenant_id=%s AND s.schema_key=%s AND f.deleted=FALSE""",
|
||
(TENANT, ttype)).fetchall()
|
||
return {r[0] for r in rows}
|
||
|
||
|
||
def load_entity_names(conn, work_id):
|
||
"""本书专名词典：scaffold 实体名就是现成专名表（短专名泄漏 15 连字抓不到，
|
||
金标准抓到 4 张「海拉/千亿夸克」级泄漏）；出处/实例定位字段豁免（设计允许专名只进出处）。"""
|
||
names = {e.get("名称") or e.get("name") for (ents,) in conn.execute(
|
||
"""SELECT entities FROM example_parse_scaffold
|
||
WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE""",
|
||
(TENANT, work_id)).fetchall() for e in ents}
|
||
return {n for n in names if n and len(n) >= 2}
|
||
|
||
|
||
def leak_check(card_texts, source_text):
|
||
"""脱敏机械检查：卡内任一文本值含与原文 ≥NGRAM 连续字重合 → 返回违规片段。"""
|
||
src = re.sub(r'\s', '', source_text)
|
||
grams = {src[i:i + NGRAM] for i in range(0, max(0, len(src) - NGRAM + 1))}
|
||
for t in card_texts:
|
||
tt = re.sub(r'\s', '', str(t))
|
||
for i in range(0, max(0, len(tt) - NGRAM + 1)):
|
||
if tt[i:i + NGRAM] in grams:
|
||
return tt[i:i + NGRAM]
|
||
return None
|
||
|
||
|
||
def validate_card(c, contracts, entity_names, src_text):
|
||
"""卡片通用机械守卫（章级 patterns 与窗级 cards 共用——守卫同源，禁两处各写一套）。
|
||
返回 (拒卡原因列表, 改型审计|None)；c 可能被就地改型。"""
|
||
t = c.get("型")
|
||
reasons, retyped = [], None
|
||
if t not in PATTERN_TYPES:
|
||
reasons.append(f"型不合法:{t}（首轮只拆五型）")
|
||
if not c.get("名称"):
|
||
reasons.append("缺名称")
|
||
fields = c.get("字段") or {}
|
||
if t in contracts and fields:
|
||
illegal = set(fields) - contracts[t]
|
||
if illegal:
|
||
# 字段指纹改型：LLM 的 type 标注不可靠（实测 M3 系统性全标 craft），
|
||
# 五型字段 key 互不重叠——字段集合全命中唯一他型合同时，以字段为准改型
|
||
fits = [t2 for t2 in PATTERN_TYPES
|
||
if t2 != t and fields and set(fields) <= contracts[t2]]
|
||
if len(fits) == 1:
|
||
retyped = f"{t}→{fits[0]}(字段指纹改型)"
|
||
c["型"] = t = fits[0]
|
||
else:
|
||
# 裁剪降级（B4-S3 实测：M3 往 trope 塞「读者收益」类他型增益字段屡教不改）：
|
||
# 越界字段剥离（内容在本型无处安放，不该陪葬整卡），审计留痕；
|
||
# 裁掉必填字段的情况由下方 craft 双模板必填校验兜底拒卡
|
||
for k in illegal:
|
||
fields.pop(k)
|
||
c["裁剪字段"] = sorted(illegal)
|
||
# 守卫（B4-S2）：craft 双模板条件必填/禁填 + 装置类型闭合枚举 + 禁复合标签
|
||
if t == "craft" and not reasons:
|
||
form = (fields.get("装置形态") or "").strip()
|
||
dtype = (fields.get("装置类型") or "").strip()
|
||
if form not in ("结构装置", "场景手法"):
|
||
reasons.append(f"装置形态非法:{form!r}（须为 结构装置/场景手法）")
|
||
elif form == "结构装置":
|
||
miss = [k for k in ("埋设手法", "回收点", "记忆维持", "间隔纪律") if not fields.get(k)]
|
||
if miss:
|
||
reasons.append(f"结构装置必填缺失:{miss}")
|
||
else: # 场景手法：禁填间隔纪律（同场景闭环没有间隔，硬填必出伪纪律）
|
||
miss = [k for k in ("复用节奏", "异质化要求", "单章上限") if not fields.get(k)]
|
||
if miss:
|
||
reasons.append(f"场景手法必填缺失:{miss}")
|
||
if fields.get("间隔纪律"):
|
||
reasons.append("场景手法禁填间隔纪律（伪纪律来源）")
|
||
if dtype:
|
||
if any(sep in dtype for sep in ("+", "、", "/", "，")):
|
||
reasons.append(f"装置类型禁复合标签:{dtype!r}（只填最主要的一个）")
|
||
elif dtype not in ("伏笔", "契诃夫之枪", "信息差", "重复意象", "倒计时", "身份错认", "非装置"):
|
||
reasons.append(f"装置类型不在闭合枚举:{dtype!r}")
|
||
# 原理句式占位符检测（实测病：M3 把模板「当X时做Y因为读者会Z」的占位符字面抄进卡）
|
||
if re.search(r"[做当会][XYZ]", str(fields.get("原理") or "")):
|
||
reasons.append("原理含未展开的模板占位符（X/Y/Z 须替换为具体内容）")
|
||
# 脱敏红线 + 专名词典（实例定位/出处豁免——设计允许专名只进定位）
|
||
texts = [x for x in (c.get("名称"), c.get("一句话摘要"), *fields.values()) if x]
|
||
leak = leak_check(texts, src_text)
|
||
if leak:
|
||
reasons.append(f"脱敏违规(≥{NGRAM}连字重合):「{leak}」")
|
||
hit_names = [n for n in entity_names if any(n in str(x) for x in texts)]
|
||
if hit_names:
|
||
reasons.append(f"专名泄漏(本书实体名):{hit_names[:3]}")
|
||
# 版权 IP 与系统专名词表（fable5 复抽实证盲区：同人书的 IP 背景词不在 scaffold
|
||
# 实体名里，批7 全部 12 张硬泄漏均由此逃逸，其中 4 张专名直入卡名——texts 含
|
||
# 卡名，此处一并覆盖。词表按实证泄漏词维护，勿加过泛词防误伤）
|
||
hit_ip = [w for w in IP_LEAK_WORDS if any(w in str(x) for x in texts)]
|
||
if hit_ip:
|
||
reasons.append(f"专名泄漏(版权IP/系统词):{hit_ip[:3]}")
|
||
return reasons, retyped
|
||
|
||
|
||
def set_task(conn, work_id, chapter_id, **cols):
|
||
"""推进任务状态机（attempt 自增）。"""
|
||
sets = ", ".join(f"{k}=%s" for k in cols)
|
||
conn.execute(
|
||
f"""UPDATE example_parse_task SET {sets}, attempt_count=attempt_count+1, updater=%s
|
||
WHERE tenant_id=%s AND work_id=%s AND chapter_id=%s""",
|
||
(*cols.values(), ACTOR, TENANT, work_id, chapter_id))
|
||
|
||
|
||
@click.group()
|
||
def cli():
|
||
"""拆书入库与任务状态机"""
|
||
|
||
|
||
@cli.command("init-tasks")
|
||
@click.option("--work-id", type=int, required=True)
|
||
@click.option("--from", "from_", type=int, default=1, show_default=True)
|
||
@click.option("--to", type=int, required=True)
|
||
def init_tasks(work_id, from_, to):
|
||
"""按章建任务行（幂等），并把参考书档案 parse_scope/parse_status 置为拆书中。"""
|
||
with psycopg.connect(DSN) as conn:
|
||
chs = conn.execute(
|
||
"""SELECT id, order_no FROM muse_content_chapter
|
||
WHERE tenant_id=%s AND work_id=%s AND order_no BETWEEN %s AND %s AND deleted=FALSE
|
||
ORDER BY order_no""", (TENANT, work_id, from_, to)).fetchall()
|
||
n = 0
|
||
for ch_id, _no in chs:
|
||
conn.execute(
|
||
"""INSERT INTO example_parse_task (work_id, chapter_id, creator, updater, tenant_id)
|
||
VALUES (%s,%s,%s,%s,%s)
|
||
ON CONFLICT (tenant_id, work_id, chapter_id) DO NOTHING""",
|
||
(work_id, ch_id, ACTOR, ACTOR, TENANT))
|
||
n += 1
|
||
conn.execute(
|
||
"""UPDATE example_reference_work SET parse_scope=%s, parse_status='parsing', updater=%s
|
||
WHERE tenant_id=%s AND work_id=%s""",
|
||
(Jsonb({"from": from_, "to": to}), ACTOR, TENANT, work_id))
|
||
conn.commit()
|
||
click.echo(f"任务行就绪: work={work_id} 章 {from_}–{to}（{n} 行）")
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--work-id", type=int, required=True)
|
||
@click.option("--chapter-order", type=int, required=True)
|
||
@click.option("--file", "file_", type=click.Path(exists=True), required=True)
|
||
def scaffold(work_id, chapter_order, file_):
|
||
"""章级入库：{细纲, 实体:[{型,名称,一句话摘要}], 线索:[{型,短名,线索,证据}]}；比例约束校验。"""
|
||
data = norm_scaffold(json.loads(pathlib.Path(file_).read_text()))
|
||
with psycopg.connect(DSN) as conn:
|
||
ch_id, src = chapter_of(conn, work_id, chapter_order)
|
||
outline = (data.get("细纲") or "").strip()
|
||
if not outline:
|
||
raise click.ClickException("细纲为空")
|
||
ratio = len(re.sub(r'\s', '', outline)) / max(1, len(re.sub(r'\s', '', src)))
|
||
# 8% 防的是「摘要化伪装结构化」（长章复述）；≤60 字的细纲对任何章都是高度压缩、
|
||
# 不可能是复述，绝对豁免——否则极短章（感言/公告，实测 116 字）永远无法过闸
|
||
if ratio > 0.08 and len(re.sub(r'\s', '', outline)) > 60: # 拍板值 3–5%，8% 为机械硬顶
|
||
set_task(conn, work_id, ch_id, scaffold_status="failed",
|
||
error_message=f"细纲比例超标 {ratio:.1%}>8%")
|
||
conn.commit()
|
||
raise click.ClickException(f"细纲比例 {ratio:.1%} 超标（>8%），已记失败退回重解析")
|
||
# 实体坏行宽容（放量实测：M3 偶发报 名称=null 的实体,一条坏行不该陪葬整章——
|
||
# 细纲/线索/其余实体全丢且任务无失败记录,窗切分会报缺章）
|
||
ents, n_bad_ent = [], 0
|
||
for e in data.get("实体") or []:
|
||
if e.get("名称") and e.get("型"):
|
||
ents.append(e)
|
||
else:
|
||
n_bad_ent += 1
|
||
# 范式候选线索（B4-S3）：候选层宽容处置——坏行丢弃计数，不整批拒
|
||
hints, n_bad = [], 0
|
||
for h in data.get("线索") or []:
|
||
if h.get("短名") and h.get("线索") and (h.get("型") in PATTERN_TYPES or h.get("型") == "?"):
|
||
hints.append(h)
|
||
else:
|
||
n_bad += 1
|
||
conn.execute(
|
||
"""INSERT INTO example_parse_scaffold (work_id, chapter_id, outline_text, entities,
|
||
pattern_hints, creator, updater, tenant_id)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
|
||
ON CONFLICT (tenant_id, chapter_id)
|
||
DO UPDATE SET outline_text=EXCLUDED.outline_text, entities=EXCLUDED.entities,
|
||
pattern_hints=EXCLUDED.pattern_hints, deleted=FALSE, updater=EXCLUDED.updater""",
|
||
(work_id, ch_id, outline, Jsonb(ents), Jsonb(hints), ACTOR, ACTOR, TENANT))
|
||
set_task(conn, work_id, ch_id, scaffold_status="done", error_message=None)
|
||
conn.commit()
|
||
click.echo(f"scaffold✓ work={work_id} ch#{chapter_order}: 细纲{len(outline)}字({ratio:.1%}) "
|
||
f"实体{len(ents)}" + (f"(弃坏行{n_bad_ent})" if n_bad_ent else "")
|
||
+ f" 线索{len(hints)}" + (f"(弃坏行{n_bad})" if n_bad else ""))
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--work-id", type=int, required=True)
|
||
@click.option("--chapter-order", type=int, required=True)
|
||
@click.option("--file", "file_", type=click.Path(exists=True), required=True)
|
||
def patterns(work_id, chapter_order, file_):
|
||
"""【旧管线·回滚保险】章级范式卡入库；B4-S3 后由窗级 cards 替代，新管线勿用。"""
|
||
raw = json.loads(pathlib.Path(file_).read_text())
|
||
if isinstance(raw, dict): # 兼容 {cards:[…]}/{卡:[…]} 包装
|
||
raw = raw.get("cards") or raw.get("卡") or []
|
||
if not isinstance(raw, list):
|
||
raise click.ClickException("patterns 文件须为卡片数组")
|
||
cards = [norm_card(c) for c in raw]
|
||
with psycopg.connect(DSN) as conn:
|
||
ch_id, src = chapter_of(conn, work_id, chapter_order)
|
||
contracts = {t: field_contract(conn, t) for t in PATTERN_TYPES}
|
||
book = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
|
||
entity_names = load_entity_names(conn, work_id)
|
||
# 幂等：重跑本章 = 软删本章旧 draft。
|
||
# 但新轮 0 卡时保留旧卡——LLM 判卡有随机性，重跑「0 卡」不应清掉上轮已验证的产出
|
||
if cards:
|
||
conn.execute(
|
||
"""UPDATE muse_knowledge_draft SET deleted=TRUE, updater=%s
|
||
WHERE tenant_id=%s AND source_type='parse_book' AND source_id=%s
|
||
AND draft_payload->>'章序'=%s AND deleted=FALSE""",
|
||
(ACTOR, TENANT, work_id, str(chapter_order)))
|
||
ok, rejected = 0, []
|
||
for i, c in enumerate(cards):
|
||
reasons, retyped = validate_card(c, contracts, entity_names, src)
|
||
src_ref = c.get("出处") or {}
|
||
if not (src_ref.get("书名") and src_ref.get("回目")):
|
||
reasons.append("出处不完整（需书名+回目）")
|
||
if reasons:
|
||
rejected.append({"卡": c.get("名称") or f"#{i}", "原因": reasons})
|
||
continue
|
||
payload = {"型": c["型"], "名称": c["名称"], "一句话摘要": c.get("一句话摘要", ""),
|
||
"字段": c.get("字段") or {}, "出处": src_ref, "目标库": "公共范式库",
|
||
"章序": chapter_order, "来源": f"拆书@{book}", "状态": "草稿"}
|
||
if retyped:
|
||
payload["改型"] = retyped # 审计：机械改型可追溯
|
||
if c.get("裁剪字段"):
|
||
payload["裁剪字段"] = c["裁剪字段"] # 审计：越合同字段被剥离入库
|
||
cid = f"parse-{work_id}-{chapter_order}-{i}-" + hashlib.sha256(
|
||
json.dumps(payload, ensure_ascii=False, sort_keys=True).encode()).hexdigest()[:8]
|
||
conn.execute(
|
||
"""INSERT INTO muse_knowledge_draft (work_id, draft_type, draft_payload, status,
|
||
source_type, source_id, command_id, creator, updater, tenant_id)
|
||
VALUES (0,'entity',%s,'pending','parse_book',%s,%s,%s,%s,%s)
|
||
ON CONFLICT (tenant_id, command_id) WHERE command_id IS NOT NULL DO NOTHING""",
|
||
(Jsonb(payload), work_id, cid, ACTOR, ACTOR, TENANT))
|
||
ok += 1
|
||
set_task(conn, work_id, ch_id, pattern_status="done",
|
||
error_message=None if not rejected else f"拒卡{len(rejected)}: " + json.dumps(rejected, ensure_ascii=False)[:900])
|
||
conn.commit()
|
||
click.echo(f"patterns✓ work={work_id} ch#{chapter_order}: 入库{ok} 拒{len(rejected)}")
|
||
for r in rejected:
|
||
click.echo(f" [拒] {r['卡']}: {'; '.join(r['原因'])}")
|
||
|
||
|
||
def merge_judge(new_payload, old_payload, model="MiniMax-M3"):
|
||
"""受控 LLM 点②：嵌入初筛 ≥SIM_MERGE 后的归并终判（输入两卡 JSON，输出 merge/keep）。
|
||
双保险设计：embedding 只做初筛，是否同一手法由 LLM 判——单靠阈值必产生错误合并。"""
|
||
prompt = ("两张「公共范式卡」由不同窗/不同书各自独立归纳（JSON 附后）。判断它们是否为同一写作手法：\n"
|
||
"- 运作机制相同（哪怕载体/题材/叫法不同）→ merge；\n"
|
||
"- 机制不同、或适用场景本质不同 → keep。拿不准时选 keep（错误合并比重复卡更伤）。\n"
|
||
'输出规则（只输出一个 JSON 对象）：{"verdict": "merge"或"keep", "reason": "一句话依据"}\n\n'
|
||
f"【卡A（已入库）】\n{json.dumps(old_payload, ensure_ascii=False)}\n\n"
|
||
f"【卡B（新卡）】\n{json.dumps(new_payload, ensure_ascii=False)}")
|
||
try:
|
||
content, _, used = chat_governed(prompt, model=model)
|
||
# 全局额度链全部拦截（content 为 None）：保守判「keep」＝不合并——判重失败绝不能误并，
|
||
# 也不该因此卡住或崩书（错误合并比重复卡更伤，宁重复不误并）
|
||
if content is None:
|
||
return "keep", "归并判定全链耗尽（内容安全/不可用），保守保留不合并"
|
||
data = extract_json(content)
|
||
return data.get("verdict"), data.get("reason", "")
|
||
except Exception as e: # 判定失败=keep（不阻断入库，宁重复不误并）
|
||
return "keep", f"归并判定调用失败({type(e).__name__})，默认保留"
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--work-id", type=int, required=True)
|
||
@click.option("--from-order", "from_order", type=int, required=True, help="窗起始章（对应大纲窗行）")
|
||
@click.option("--file", "file_", type=click.Path(exists=True), required=True)
|
||
@click.option("--model", default="MiniMax-M3", show_default=True, help="归并判定用模型")
|
||
def cards(work_id, from_order, file_, model):
|
||
"""窗级母卡入库（B4-S3）：{cards:[{型,名称,摘要,字段,实例[{章,定位}]}]}。
|
||
守卫=章级全部 + 窗级新增：实例域校验 / 间隔章数机械计算（伪精确灭绝）/ 嵌入判重→归并判定。"""
|
||
raw = json.loads(pathlib.Path(file_).read_text())
|
||
if isinstance(raw, dict):
|
||
raw = raw.get("cards") or raw.get("卡") or []
|
||
cards_in = [norm_card(c) for c in raw]
|
||
with psycopg.connect(DSN) as conn:
|
||
win = conn.execute(
|
||
"""SELECT to_order FROM example_parse_outline
|
||
WHERE tenant_id=%s AND work_id=%s AND from_order=%s AND deleted=FALSE""",
|
||
(TENANT, work_id, from_order)).fetchone()
|
||
if not win:
|
||
raise click.ClickException(f"窗不存在: work={work_id} from={from_order}——先跑 parse_outline window")
|
||
to_order = win[0]
|
||
book = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
|
||
# 窗内正文拼接（15 连字红线对照源：卡可能引到窗内任何一章）
|
||
src = "".join(t for (t,) in conn.execute(
|
||
"""SELECT b.content_text FROM muse_content_chapter c
|
||
JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=FALSE
|
||
WHERE c.tenant_id=%s AND c.work_id=%s AND c.order_no BETWEEN %s AND %s
|
||
AND c.deleted=FALSE ORDER BY c.order_no""",
|
||
(TENANT, work_id, from_order, to_order)).fetchall())
|
||
contracts = {t: field_contract(conn, t) for t in PATTERN_TYPES}
|
||
entity_names = load_entity_names(conn, work_id)
|
||
# 先全量校验，收集通过集——软删旧卡延后到「确有新卡通过」才执行。
|
||
# 教训（放量实测）：软删在校验前时，新卡全拒（如 M3 整批丢实例键）=旧卡已删+新卡全拒=净损失
|
||
passed, rejected = [], []
|
||
for i, c in enumerate(cards_in):
|
||
reasons, retyped = validate_card(c, contracts, entity_names, src)
|
||
# 实例域校验：章号必须落在本窗内——越窗章号=编造（跨窗同手法靠嵌入判重归并）
|
||
inss = [ins for ins in c.get("实例") or [] if from_order <= ins["章"] <= to_order]
|
||
n_drop = len(c.get("实例") or []) - len(inss)
|
||
if not inss:
|
||
reasons.append("实例为空或章号全部越窗（窗级卡必须带窗内实例）")
|
||
fields = c.get("字段") or {}
|
||
# 间隔章数机械计算（伪精确灭绝）：结构装置实例≥2 → 章号差为权威值追加；
|
||
# 单实例结构装置=本窗证据不全（可能收在后续窗），标「跨窗待证」入库不拒
|
||
cross_pend = False
|
||
if not reasons and c.get("型") == "craft" and fields.get("装置形态") == "结构装置":
|
||
ch_nos = sorted({ins["章"] for ins in inss})
|
||
if len(ch_nos) >= 2:
|
||
fields["间隔纪律"] = ((fields.get("间隔纪律") or "").strip()
|
||
+ f"（实测：#{ch_nos[0]}埋→#{ch_nos[-1]}收，隔{ch_nos[-1] - ch_nos[0]}章）").strip()
|
||
else:
|
||
cross_pend = True
|
||
if reasons:
|
||
rejected.append({"卡": c.get("名称") or f"#{i}", "原因": reasons})
|
||
continue
|
||
payload = {"型": c["型"], "名称": c["名称"], "一句话摘要": c.get("一句话摘要", ""),
|
||
"字段": fields, "实例": inss,
|
||
"出处": {"书名": book, "回目": f"第{inss[0]['章']}章", "定位": inss[0]["定位"]},
|
||
"目标库": "公共范式库", "窗起": from_order, "来源": f"拆书@{book}", "状态": "草稿"}
|
||
if retyped:
|
||
payload["改型"] = retyped
|
||
if c.get("裁剪字段"):
|
||
payload["裁剪字段"] = c["裁剪字段"] # 审计：越合同字段被剥离入库
|
||
if cross_pend:
|
||
payload["跨窗待证"] = True # 审核环可见：结构装置只有单端实例
|
||
if n_drop:
|
||
payload["越窗实例弃"] = n_drop # 审计：M3 报了窗外章号
|
||
passed.append((i, payload, inss))
|
||
# 确有通过卡才软删本窗旧卡（幂等重出；全拒/空批不动旧卡——0 卡保护的窗级版）
|
||
if passed:
|
||
conn.execute(
|
||
"""UPDATE muse_knowledge_draft SET deleted=TRUE, updater=%s
|
||
WHERE tenant_id=%s AND source_type='parse_book' AND source_id=%s
|
||
AND draft_payload->>'窗起'=%s AND deleted=FALSE""",
|
||
(ACTOR, TENANT, work_id, str(from_order)))
|
||
sess = _session() # 嵌入通道（判重+新卡即时嵌入共用）
|
||
ok, merged = 0, []
|
||
for i, payload, inss in passed:
|
||
# ── 受控点①②：嵌入判重（失败不阻断——判重是增强不是红线，卡不能因通道故障丢）──
|
||
embed_text = build_embed_text(payload)
|
||
vec = None
|
||
try:
|
||
vecs, bad = embed_texts(sess, [embed_text])
|
||
vec = None if bad else vecs[0]
|
||
except Exception:
|
||
pass
|
||
if vec is not None:
|
||
top = conn.execute(
|
||
"""SELECT d.id, d.draft_payload, 1-(e.embedding<=>%s::vector) AS score
|
||
FROM example_knowledge_embedding e
|
||
JOIN muse_knowledge_draft d ON d.id=e.draft_id
|
||
WHERE e.tenant_id=%s AND e.deleted=FALSE AND d.deleted=FALSE
|
||
AND d.source_type='parse_book'
|
||
ORDER BY score DESC LIMIT 1""", (json.dumps(vec), TENANT)).fetchone()
|
||
# 前向修复(2026-07-17)：不再"只有 ≥SIM_MERGE 才判、0.75-0.85 仅打标记"——
|
||
# 最近邻 ≥SIM_MARK(0.75) 即送 merge_judge，与 recluster 回溯聚类同一判据。
|
||
# WHY：跨书 0.75-0.85 的近义卡此前从不真正判定、只挂"未达判定线"，是重复卡沉积的病根。
|
||
if top and top[2] >= SIM_MARK:
|
||
old_id, old_payload, score = top
|
||
verdict, why = merge_judge(payload, old_payload, model)
|
||
if verdict == "merge":
|
||
# 归并=旧卡追加实例（跨书时带书名）+完整审计（含被归并卡全文，可逆）
|
||
add = [dict(ins, 书=book) for ins in inss]
|
||
old_payload["实例"] = (old_payload.get("实例") or []) + add
|
||
old_payload.setdefault("归并审计", []).append(
|
||
{"相似度": round(float(score), 4), "判定依据": why, "被归并卡": payload})
|
||
conn.execute(
|
||
"UPDATE muse_knowledge_draft SET draft_payload=%s, updater=%s WHERE id=%s",
|
||
(Jsonb(old_payload), ACTOR, old_id))
|
||
merged.append(f"《{payload['名称']}》并入已有卡#{old_id}《{old_payload.get('名称')}》({score:.2f}) {why}")
|
||
continue
|
||
# keep 留痕；档位区分强候选(≥SIM_MERGE)/扩展候选(≥SIM_MARK)，供审核判读近似度
|
||
payload["判重"] = {"相似卡": old_payload.get("名称"), "相似度": round(float(score), 4),
|
||
"判定": "keep", "档": "强候选" if score >= SIM_MERGE else "扩展候选",
|
||
"依据": why}
|
||
cid = f"parse-{work_id}-w{from_order}-{i}-" + hashlib.sha256(
|
||
json.dumps(payload, ensure_ascii=False, sort_keys=True).encode()).hexdigest()[:8]
|
||
row = conn.execute(
|
||
"""INSERT INTO muse_knowledge_draft (work_id, draft_type, draft_payload, status,
|
||
source_type, source_id, command_id, creator, updater, tenant_id)
|
||
VALUES (0,'entity',%s,'pending','parse_book',%s,%s,%s,%s,%s)
|
||
ON CONFLICT (tenant_id, command_id) WHERE command_id IS NOT NULL DO NOTHING
|
||
RETURNING id""",
|
||
(Jsonb(payload), work_id, cid, ACTOR, ACTOR, TENANT)).fetchone()
|
||
# 新卡即时嵌入（下一窗/下一书判重立即可见；失败留给 embed skill 批量补）
|
||
if row and vec is not None:
|
||
h = hashlib.sha256(f"{embed_text}|{EMBED_MODEL}".encode()).hexdigest()
|
||
conn.execute(
|
||
"""INSERT INTO example_knowledge_embedding
|
||
(draft_id, content_hash, embed_text, model, dimensions, embedding,
|
||
creator, updater, tenant_id)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
ON CONFLICT (tenant_id, content_hash, model) DO NOTHING""",
|
||
(row[0], h, embed_text, EMBED_MODEL, 1024, json.dumps(vec), ACTOR, ACTOR, TENANT))
|
||
ok += 1
|
||
# 窗内章 pattern_status 批量置 done（窗级出卡完成=这些章的范式阶段完成）
|
||
conn.execute(
|
||
"""UPDATE example_parse_task SET pattern_status='done', updater=%s
|
||
WHERE tenant_id=%s AND work_id=%s AND chapter_id IN (
|
||
SELECT id FROM muse_content_chapter
|
||
WHERE tenant_id=%s AND work_id=%s AND order_no BETWEEN %s AND %s AND deleted=FALSE)""",
|
||
(ACTOR, TENANT, work_id, TENANT, work_id, from_order, to_order))
|
||
conn.commit()
|
||
click.echo(f"cards✓ work={work_id} win#{from_order}–{to_order}: 入库{ok} 归并{len(merged)} 拒{len(rejected)}")
|
||
for m in merged:
|
||
click.echo(f" [归并] {m}")
|
||
for r in rejected:
|
||
click.echo(f" [拒] {r['卡']}: {'; '.join(r['原因'])}")
|
||
|
||
|
||
def _judge_view(payload):
|
||
"""送 merge_judge 的"手法身份"视图：只留 型/名称/摘要/字段（判"是否同一手法"的全部依据）。
|
||
WHY 裁剪：审核(三角色判词)/实例/出处/判重 是噪声——尤其母卡累积实例与「回溯归并审计」后
|
||
payload 会塞进多张别的卡全文，直接喂 merge_judge 既爆 token 又把判定带偏；裁到身份四件套后
|
||
判定输入稳定（母卡吸并再多，身份不变）、便宜、聚焦手法本身。"""
|
||
return {"型": payload.get("型"), "名称": payload.get("名称"),
|
||
"一句话摘要": payload.get("一句话摘要"), "字段": payload.get("字段") or {}}
|
||
|
||
|
||
def _mother_key(payload, cid):
|
||
"""母卡优先序（best-first 排序键）：审核判定优 > 均分高 > 实例多 > id 小。
|
||
WHY：让存活下来的母卡是该手法的最佳代表——下游审核/导出以母卡为规范样本；实例多者是更厚的
|
||
证据锚点；id 兜底保证确定性可复现。审核缺失(他型可能未审)排在最后，不抢占母卡位。"""
|
||
r = payload.get("审核") or {}
|
||
rank = {"pass": 0, "revise": 1, "reject": 2}.get(r.get("判定"), 3)
|
||
return (rank, -(r.get("均分") or 0), -len(payload.get("实例") or []), cid)
|
||
|
||
|
||
def _absorb(mother, loser, sim, why):
|
||
"""把输家并入母卡（内存态，写库延后到写段统一落）：输家实例(带书名归属)追加进母卡实例数组
|
||
+ 记「回溯归并审计」（相似度/判定依据/被归并卡全文——全文留痕保证归并可逆）。"""
|
||
book = (loser.get("出处") or {}).get("书名") or ""
|
||
# 跨书归属：优先保留实例原有的书名——输家可能是"此前已跨书吸并过的母卡"，其实例里混有别书来源，
|
||
# 绝不能用输家自己的书名无条件覆写（那会把二级归并的真实来源抹成输家书名）；无原书名的才补输家书名。
|
||
add = [dict(ins, 书=ins.get("书") or book) for ins in (loser.get("实例") or [])]
|
||
mother["实例"] = (mother.get("实例") or []) + add
|
||
mother.setdefault("回溯归并审计", []).append(
|
||
{"相似度": round(float(sim), 4), "判定依据": why, "被归并卡": loser})
|
||
|
||
|
||
def _sim_band(s):
|
||
"""附着边相似度分档（供 dry-run 报告：不同档 LLM 剪枝率差异大，同型词汇共享令 0.75-0.80 多为假近）。"""
|
||
return "≥0.85" if s >= 0.85 else ("0.80–0.85" if s >= 0.80 else "0.75–0.80")
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--type", "ptype", required=True, type=click.Choice(sorted(PATTERN_TYPES)),
|
||
help="回溯聚类的范式型（一次只跑一型）")
|
||
@click.option("--dry-run", is_flag=True, help="只算不写：报簇结构/计划归并/N→M + 抽样 merge_judge 判定")
|
||
@click.option("--limit", type=int, default=0, help="只取前 N 张卡（调试用，0=全型）")
|
||
@click.option("--sample", type=int, default=10, show_default=True,
|
||
help="dry-run 下抽样真跑 merge_judge 的候选对数（按相似度档分层取，露出各档合并率）")
|
||
@click.option("--model", default="MiniMax-M3", show_default=True, help="归并判定用模型")
|
||
def recluster(ptype, dry_run, limit, sample, model):
|
||
"""回溯聚类去重：把同型近义卡（嵌入初筛 + merge_judge 终判）归并到母卡。
|
||
|
||
病根：现行 cards() 入库只比最近 1 张、阈值偏高、按书顺序入库→跨书 0.75-0.85 的近义卡从不送判定。
|
||
本命令一次回溯：同型全量按余弦相似度成簇，逐对经 LLM 确认后把输家并入母卡（软删、可逆）。
|
||
|
||
连接三段式（血泪教训：DB 事务里夹 LLM，长空转会被 Tailscale 掐断整批崩）：
|
||
读段 短连接取全型卡 + 候选对（pgvector self-join，纯 SQL 无 LLM），读完即释放；
|
||
算段 本地贪心成簇 + merge_judge 逐对终判（此阶段 recluster 不持任何 DB 连接，
|
||
merge_judge 走 chat_governed 自持额度账本短连接，互不干扰）；
|
||
写段 全新短连接批量落库（母卡累积 payload / 输家卡 deleted / 输家嵌入 deleted），一次提交。
|
||
双保险：embedding 只初筛出候选，是否同一手法一律由 merge_judge 定夺，拿不准 keep（宁重复不误并）。"""
|
||
# ── 读段：短连接读完即释放 ──
|
||
with psycopg.connect(DSN) as rconn:
|
||
rows = rconn.execute(
|
||
"""SELECT d.id, d.draft_payload FROM muse_knowledge_draft d
|
||
WHERE d.tenant_id=%s AND d.source_type='parse_book' AND d.deleted=FALSE
|
||
AND d.draft_payload->>'型'=%s ORDER BY d.id""",
|
||
(TENANT, ptype)).fetchall()
|
||
if limit:
|
||
rows = rows[:limit]
|
||
cards = {cid: p for cid, p in rows}
|
||
ids = set(cards)
|
||
# 候选对：同型嵌入 self-join，余弦相似度 ≥SIM_MARK。纯 SQL（无 LLM）；pgvector 一次算完 O(N²)，
|
||
# combat 886 张实测 ~3s。取回后本地成簇，判定/写库阶段不再回这条连接。
|
||
pair_rows = rconn.execute(
|
||
"""WITH emb AS (
|
||
SELECT e.draft_id, e.embedding FROM example_knowledge_embedding e
|
||
JOIN muse_knowledge_draft d ON d.id=e.draft_id
|
||
WHERE e.tenant_id=%s AND e.deleted=FALSE AND d.deleted=FALSE
|
||
AND d.source_type='parse_book' AND d.draft_payload->>'型'=%s)
|
||
SELECT a.draft_id, b.draft_id, 1-(a.embedding<=>b.embedding) AS sim
|
||
FROM emb a JOIN emb b ON a.draft_id<b.draft_id
|
||
WHERE 1-(a.embedding<=>b.embedding) >= %s""",
|
||
(TENANT, ptype, SIM_MARK)).fetchall()
|
||
# 邻接表（id→[(邻卡,相似度)]）；--limit 调试时过滤掉越界 id
|
||
adj = {}
|
||
for a, b, s in pair_rows:
|
||
if a in ids and b in ids:
|
||
s = float(s)
|
||
adj.setdefault(a, []).append((b, s))
|
||
adj.setdefault(b, []).append((a, s))
|
||
n = len(ids)
|
||
if not n:
|
||
click.echo(f"[recluster·{ptype}] 无活卡")
|
||
return
|
||
|
||
# ── 算段：贪心「最近母卡」聚类（best-first）──
|
||
# 为何不用 union-find：0.75 阈下同型卡词汇高度共享，单链传递会把几乎整型并成一个巨簇
|
||
#（combat 实测 38665 对 / 817 张连通）。贪心「只附着到最近的一个母卡」天然避免链式膨胀，
|
||
# 且判定次数最省：每张非母卡至多 1 次 merge_judge（判它 vs 最近母卡）。
|
||
order = sorted(ids, key=lambda cid: _mother_key(cards[cid], cid))
|
||
leaders, leader_set = [], set()
|
||
plan = [] # 计划/已确认归并：(母卡id, 输家id, 相似度)
|
||
|
||
def nearest_leader(cid):
|
||
"""该卡邻居中"已是母卡"的最近一个（sim 最大）；无则 None。"""
|
||
best = None
|
||
for nbr, s in adj.get(cid, []):
|
||
if nbr in leader_set and (best is None or s > best[1]):
|
||
best = (nbr, s)
|
||
return best
|
||
|
||
if dry_run:
|
||
# 纯向量投影（几乎不烧 LLM）：每卡附着到最近母卡＝一条计划归并，无最近母卡＝自成母卡。
|
||
# 这是归并的「上限」——实跑每对还要过 merge_judge，keep 者不并→真实存活更多。
|
||
for cid in order:
|
||
nb = nearest_leader(cid)
|
||
if nb:
|
||
plan.append((nb[0], cid, nb[1]))
|
||
else:
|
||
leaders.append(cid)
|
||
leader_set.add(cid)
|
||
else:
|
||
# 实跑：每次附着都由 merge_judge 终判（受控点②）。此阶段 recluster 不持 DB 连接。
|
||
for cid in order:
|
||
nb = nearest_leader(cid)
|
||
if nb:
|
||
lid, sim = nb
|
||
verdict, why = merge_judge(_judge_view(cards[cid]), _judge_view(cards[lid]), model)
|
||
if verdict == "merge":
|
||
_absorb(cards[lid], cards[cid], sim, why) # 内存改母卡，写段统一落库
|
||
plan.append((lid, cid, sim))
|
||
continue
|
||
leaders.append(cid) # 无最近母卡 或 判 keep → 自成母卡
|
||
leader_set.add(cid)
|
||
|
||
# ── 写段：全新短连接批量落库（红线：软删，绝不物理删）──
|
||
if not dry_run and plan:
|
||
movers = {lid for lid, _lo, _s in plan} # 被吸并过的母卡（去重，每张只写一次最终态）
|
||
with psycopg.connect(DSN) as wconn:
|
||
for lid in movers:
|
||
wconn.execute("UPDATE muse_knowledge_draft SET draft_payload=%s, updater=%s WHERE id=%s",
|
||
(Jsonb(cards[lid]), ACTOR, lid))
|
||
for _lid, loser_id, _s in plan:
|
||
wconn.execute("UPDATE muse_knowledge_draft SET deleted=TRUE, updater=%s WHERE id=%s",
|
||
(ACTOR, loser_id)) # 输家卡软删
|
||
wconn.execute(
|
||
"""UPDATE example_knowledge_embedding SET deleted=TRUE, updater=%s
|
||
WHERE tenant_id=%s AND draft_id=%s AND deleted=FALSE""",
|
||
(ACTOR, TENANT, loser_id)) # 输家嵌入软删：未来判重/检索不再命中已并走的卡
|
||
wconn.commit()
|
||
|
||
# ── 报告 ──
|
||
from collections import Counter
|
||
per_mother = Counter(lid for lid, _lo, _s in plan) # 母卡 → 吸并数
|
||
band = Counter(_sim_band(s) for _l, _lo, s in plan) # 附着边相似度分档
|
||
surv = n - len(plan) # 存活母卡数
|
||
tag = "dry-run·向量投影上限" if dry_run else "实跑·LLM确认"
|
||
click.echo(f"[recluster·{ptype}·{tag}] 活卡 {n} → 存活 {surv}"
|
||
f"（归并 {len(plan)}，缩减 {len(plan) / n:.0%}）")
|
||
if dry_run:
|
||
click.echo(" 注：向量投影是归并上限；实跑每对经 merge_judge 终判，keep 者不并→实际存活更多。")
|
||
click.echo(" 附着边相似度：" + " ".join(f"{k}={band.get(k, 0)}" for k in ("≥0.85", "0.80–0.85", "0.75–0.80")))
|
||
click.echo(f" 含归并的簇 {len(per_mother)}（另有 {surv - len(per_mother)} 张孤卡母卡）；最大簇 top5：")
|
||
for mid, cnt in per_mother.most_common(5):
|
||
click.echo(f" #{mid}《{cards[mid].get('名称')}》聚 {cnt + 1} 张（母1+并{cnt}）")
|
||
|
||
# dry-run 抽样：按相似度档「分层」各取若干真跑 merge_judge——展示判定质量 + 分档外推真实归并数。
|
||
# WHY 分层而非等距：附着边多数落在 0.75-0.80（同型词汇高度共享的"假近"区，LLM 几乎全 keep），
|
||
# 等距抽样会几乎全落该档、把高档才有的真归并淹没成 0；分层能露出每档合并率，据此把向量上限收敛到现实预估。
|
||
if dry_run and plan and sample > 0:
|
||
by_band = {"≥0.85": [], "0.80–0.85": [], "0.75–0.80": []}
|
||
for e in sorted(plan, key=lambda x: x[2], reverse=True):
|
||
by_band[_sim_band(e[2])].append(e)
|
||
live_bands = [b for b in by_band if by_band[b]]
|
||
per = max(1, sample // max(1, len(live_bands))) # 每档配额（档内等距取，覆盖该档相似度跨度）
|
||
picks = []
|
||
for b in live_bands:
|
||
es = by_band[b]
|
||
k = min(per, len(es))
|
||
idxs = [round(i * (len(es) - 1) / (k - 1)) for i in range(k)] if k > 1 else [0]
|
||
picks += [es[i] for i in dict.fromkeys(idxs)]
|
||
click.echo(f" —— 抽样 merge_judge 判定（{len(picks)} 对，按相似度档分层）——")
|
||
band_hit = {b: [0, 0] for b in by_band} # 档 → [merge数, 抽样数]
|
||
for lid, loser_id, sim in picks:
|
||
verdict, why = merge_judge(_judge_view(cards[loser_id]), _judge_view(cards[lid]), model)
|
||
bd = _sim_band(sim)
|
||
band_hit[bd][0] += verdict == "merge"
|
||
band_hit[bd][1] += 1
|
||
click.echo(f" [{verdict}] {sim:.3f}({bd}) 输《{cards[loser_id].get('名称')}》→ "
|
||
f"母《{cards[lid].get('名称')}》：{why}")
|
||
# 分档外推：估真实归并 = Σ(该档附着边数 × 该档抽样 merge 率)。样本小，仅供数量级判断。
|
||
est = sum(band.get(b, 0) * (band_hit[b][0] / band_hit[b][1] if band_hit[b][1] else 0) for b in by_band)
|
||
for b in by_band:
|
||
hit, tot = band_hit[b]
|
||
click.echo(f" 档 {b}: 抽样 merge {hit}/{tot}，本档附着边 {band.get(b, 0)}"
|
||
f" → 估归并 ~{band.get(b, 0) * (hit / tot if tot else 0):.0f}")
|
||
click.echo(f" 分档外推真实归并 ~{est:.0f}（{n}→~{n - est:.0f}）；样本小仅供数量级，放量以实跑逐对确认为准")
|
||
elif not dry_run:
|
||
for lid, loser_id, _s in plan[:20]:
|
||
click.echo(f" [归并] 《{cards[loser_id].get('名称')}》并入 #{lid}《{cards[lid].get('名称')}》")
|
||
if len(plan) > 20:
|
||
click.echo(f" …另 {len(plan) - 20} 条归并")
|
||
|
||
|
||
@cli.command()
|
||
@click.option("--work-id", type=int)
|
||
def progress(work_id):
|
||
"""进度统计（审查面）。"""
|
||
with psycopg.connect(DSN) as conn:
|
||
where = " AND t.work_id=%s" if work_id else ""
|
||
args = [TENANT] + ([work_id] if work_id else [])
|
||
rows = conn.execute(f"""
|
||
SELECT w.title, count(*) FILTER (WHERE t.scaffold_status='done') AS s_done,
|
||
count(*) FILTER (WHERE t.pattern_status='done') AS p_done, count(*) AS total,
|
||
(SELECT count(*) FROM muse_knowledge_draft d
|
||
WHERE d.tenant_id=%s AND d.source_type='parse_book' AND d.source_id=t.work_id
|
||
AND d.deleted=FALSE) AS drafts
|
||
FROM example_parse_task t JOIN muse_content_work w ON w.id=t.work_id
|
||
WHERE t.tenant_id=%s AND t.deleted=FALSE{where}
|
||
GROUP BY w.title, t.work_id ORDER BY w.title""", [TENANT] + args).fetchall()
|
||
for r in rows:
|
||
click.echo(f"{r[0]:<12} 脚手架 {r[1]}/{r[3]} 范式 {r[2]}/{r[3]} 草稿卡 {r[4]}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
cli()
|
||
except psycopg.Error as e:
|
||
click.echo(f"[db错误] {type(e).__name__}: {e}", err=True)
|
||
sys.exit(1)
|