169 lines
8.9 KiB
Python
169 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
||
"""parse-book skill：终态样张导出（零 LLM，纯库读渲染 markdown）。
|
||
|
||
跑序末环「export 样张呈报」的固化实现（批9 收尾落地）：
|
||
- patterns：五书范式卡样张——每书每型抽实例数最多的代表卡（实例多=跨章生长充分，
|
||
在 review 全量审核未跑时作为质量代理指标），combat 型加倍抽样给创始人拍板
|
||
「题材绑定：打标签 vs 改写泛化」提供判断材料。
|
||
- upgrade：单书升格实体卡样张——按出场章数取头部实体（主角/核心配角自然浮上），
|
||
重点渲染追加类字段的 [窗N] 生长轨迹与别名归并，这是升格管线的核心卖点。
|
||
|
||
用法：
|
||
export-patterns [--top 1] [--combat-top 2] [--out 路径]
|
||
export-upgrade --work-id N [--top 8] [--out 路径]
|
||
"""
|
||
import json
|
||
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 # noqa: E402
|
||
|
||
|
||
def render_val(v, limit=6):
|
||
"""字段值渲染：list 逐条列出（追加类字段的窗号轨迹），超长截断标注省略数。"""
|
||
if isinstance(v, list):
|
||
shown = v[:limit]
|
||
tail = f"\n - …（另 {len(v) - limit} 条略）" if len(v) > limit else ""
|
||
return "".join(f"\n - {x}" for x in shown) + tail
|
||
return str(v)
|
||
|
||
|
||
@click.group()
|
||
def cli():
|
||
pass
|
||
|
||
|
||
@cli.command("export-patterns")
|
||
@click.option("--top", type=int, default=1, show_default=True, help="每书每型抽几张代表卡")
|
||
@click.option("--combat-top", type=int, default=2, show_default=True, help="combat 型抽样数（拍板材料加倍）")
|
||
@click.option("--out", default="", help="输出文件（默认打印到 stdout）")
|
||
def export_patterns(top, combat_top, out):
|
||
"""五书范式卡终态样张：统计面貌 + 每型实例数 top 代表卡。"""
|
||
lines = []
|
||
with psycopg.connect(DSN) as conn:
|
||
works = conn.execute(
|
||
"""SELECT DISTINCT draft_payload->'出处'->>'书名' FROM muse_knowledge_draft
|
||
WHERE source_type='parse_book' AND deleted=FALSE ORDER BY 1""").fetchall()
|
||
# 全局统计头：五书×五型分布，给创始人 30 秒看清全库面貌
|
||
lines.append("# 范式卡终态样张（批9 五书收官）\n")
|
||
lines.append("> 抽样规则：每书每型取**实例数最多**的代表卡（实例多=跨章复现充分）；"
|
||
"combat 型加倍抽样，供「题材绑定：打标签 vs 改写泛化」拍板。\n>")
|
||
lines.append("> 口径说明：卡按「出处.书名」归书统计；跨书判重归并后实例可能来自多本书"
|
||
"（带书名标注的实例即是）——这是设计特性：同一范式跨书复现，实例累积是迁移性证据。\n")
|
||
lines.append("## 全库面貌\n")
|
||
lines.append("| 书 | 总卡数 | " + " | ".join(["craft", "trope", "scene_pattern", "emotion", "combat"]) + " | 多实例卡 |")
|
||
lines.append("|---|---|---|---|---|---|---|---|")
|
||
for (title,) in works:
|
||
st = dict(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""", (title,)).fetchall())
|
||
multi = conn.execute(
|
||
"""SELECT count(*) FROM muse_knowledge_draft
|
||
WHERE source_type='parse_book' AND deleted=FALSE
|
||
AND draft_payload->'出处'->>'书名'=%s
|
||
AND jsonb_array_length(draft_payload->'实例') > 1""", (title,)).fetchone()[0]
|
||
total = sum(st.values())
|
||
lines.append(f"| {title} | {total} | " + " | ".join(
|
||
str(st.get(t, 0)) for t in ["craft", "trope", "scene_pattern", "emotion", "combat"])
|
||
+ f" | {multi} |")
|
||
# 逐书逐型代表卡
|
||
for (title,) in works:
|
||
lines.append(f"\n## 《{title}》\n")
|
||
for ttype in ["craft", "trope", "scene_pattern", "emotion", "combat"]:
|
||
k = combat_top if ttype == "combat" else top
|
||
rows = conn.execute(
|
||
"""SELECT draft_payload FROM muse_knowledge_draft
|
||
WHERE source_type='parse_book' AND deleted=FALSE
|
||
AND draft_payload->'出处'->>'书名'=%s AND draft_payload->>'型'=%s
|
||
ORDER BY jsonb_array_length(draft_payload->'实例') DESC LIMIT %s""",
|
||
(title, ttype, k)).fetchall()
|
||
for (p,) in rows:
|
||
inst = p.get("实例") or []
|
||
lines.append(f"### {p.get('名称')}（{ttype}·实例 {len(inst)} 条）\n")
|
||
lines.append(f"**摘要**：{p.get('一句话摘要', '')}\n")
|
||
for fk, fv in (p.get("字段") or {}).items():
|
||
lines.append(f"- **{fk}**：{render_val(fv, limit=4)}")
|
||
for i in inst[:4]:
|
||
# 实例渲染成可读行：dict 形态（章/定位/书）拼成「第N章·定位」，跨书实例标书名
|
||
if isinstance(i, dict):
|
||
bk = f"《{i['书']}》" if i.get("书") else ""
|
||
lines.append(f"- 实例：{bk}第{i.get('章', '?')}章·{i.get('定位', '')}")
|
||
else:
|
||
lines.append(f"- 实例：{i}")
|
||
if len(inst) > 4:
|
||
lines.append(f"- 实例：…（另 {len(inst) - 4} 条略）")
|
||
lines.append("")
|
||
_emit(lines, out)
|
||
|
||
|
||
@cli.command("export-upgrade")
|
||
@click.option("--work-id", type=int, required=True)
|
||
@click.option("--top", type=int, default=8, show_default=True, help="按出场章数取头部实体数")
|
||
@click.option("--out", default="", help="输出文件（默认打印到 stdout）")
|
||
def export_upgrade(work_id, top, out):
|
||
"""单书升格实体卡终态样张：统计面貌 + 头部实体全字段生长轨迹 + 各型代表。"""
|
||
lines = []
|
||
with psycopg.connect(DSN) as conn:
|
||
title = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
|
||
cards = conn.execute(
|
||
"""SELECT d.id, d.draft_payload FROM muse_knowledge_draft d
|
||
JOIN example_upgrade_card_state cs ON cs.draft_id=d.id
|
||
WHERE d.work_id=%s AND d.deleted=FALSE""", (work_id,)).fetchall()
|
||
st = {}
|
||
for _, p in cards:
|
||
st[p.get("type", "?")] = st.get(p.get("type", "?"), 0) + 1
|
||
n_alias = conn.execute(
|
||
"SELECT count(*) FROM example_upgrade_alias WHERE work_id=%s AND deleted=FALSE",
|
||
(work_id,)).fetchone()[0]
|
||
lines.append(f"# 《{title}》升格实体卡终态样张（{len(cards)} 张）\n")
|
||
lines.append(f"> 型分布：{st}｜别名归并 {n_alias} 条。")
|
||
lines.append("> 字段三类演进：底色覆写（保终态+审计）/ 演进追加（带 [窗N] 时间线）/ 别名累积。\n")
|
||
# 头部实体：出场章数排序（主角/核心配角自然浮上），展示完整生长轨迹
|
||
ranked = sorted(cards, key=lambda r: -len(r[1].get("出场章") or []))
|
||
picked = ranked[:top]
|
||
# 各型代表补位：头部多为 character，其他型（faction/item/setting…）各补 1 张最厚的
|
||
seen_ids = {r[0] for r in picked}
|
||
for ttype in sorted({p.get("type") for _, p in cards} - {"character", None}):
|
||
cand = [r for r in ranked if r[1].get("type") == ttype and r[0] not in seen_ids]
|
||
if cand:
|
||
picked.append(cand[0])
|
||
seen_ids.add(cand[0][0])
|
||
for did, p in picked:
|
||
app = p.get("出场章") or []
|
||
als = [a for (a,) in conn.execute(
|
||
"""SELECT alias FROM example_upgrade_alias
|
||
WHERE work_id=%s AND canonical_name=%s AND deleted=FALSE""",
|
||
(work_id, p.get("名称"))).fetchall()]
|
||
span = f"{min(app)}–{max(app)} 章共 {len(app)} 次" if app else "无"
|
||
lines.append(f"## {p.get('名称')}（{p.get('type')}·出场 {span}）\n")
|
||
if als:
|
||
lines.append(f"**别名**：{', '.join(als)}\n")
|
||
lines.append(f"**摘要**：{p.get('一句话摘要', '')}\n")
|
||
for fk, fv in (p.get("字段") or {}).items():
|
||
lines.append(f"- **{fk}**：{render_val(fv, limit=8)}")
|
||
lines.append("")
|
||
_emit(lines, out)
|
||
|
||
|
||
def _emit(lines, out):
|
||
"""输出落盘或打印；落盘时报字数便于评估呈报体量。"""
|
||
text = "\n".join(lines)
|
||
if out:
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(text)
|
||
click.echo(f"已导出 {out}（{len(text)} 字）")
|
||
else:
|
||
click.echo(text)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
cli()
|
||
except psycopg.Error as e:
|
||
click.echo(f"[错误] {type(e).__name__}: {e}", err=True)
|
||
sys.exit(1)
|