119 lines
7.0 KiB
Python
119 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
||
"""parse-book skill：升格全量重抽准备——清一本书的升格产出与串行状态（2026-07-18 批9c 收尾）。
|
||
|
||
为什么需要它（锚点「全量重抽的管线坑」第 1 条）：升格是**串行有状态**管线，直接重跑会被
|
||
旧状态卡死——窗全是 done 不再处理、预扫/判重底册装着旧代卡把新观察并进废卡、旧审计行会被
|
||
撤销机制错还原到已软删的旧卡上、别名表全行唯一索引让新别名被"冲突即跳过"静默吞掉。
|
||
|
||
清理边界（逐表核实过约束后定的，2026-07-18）：
|
||
· muse_knowledge_draft 升格卡 → **软删**（deleted=TRUE，红线：知识数据绝不物删，可回滚）
|
||
· example_upgrade_window → 保留窗行（窗参数未变、from_chapter 锚幂等），仅重置 status=pending
|
||
· example_upgrade_alias → **硬删**该书行——uk(tenant_id,work_id,alias) 是全行唯一索引（非部分索引），
|
||
软删旧行会挡住重抽新 INSERT（ON CONFLICT DO NOTHING 静默吞）→ 别名判重空转
|
||
· example_upgrade_presence → 硬删该书行（运行状态留档，重抽全量重生成；undo 机制既有做法即硬删）
|
||
· example_upgrade_card_state → 硬删该书行（卡水位挂旧卡 draft_id，新卡新水位）
|
||
· example_upgrade_audit → 硬删该书行——undo_window 的还原 JOIN **不过滤 d.deleted**，
|
||
留着旧审计行，重抽期间任何同窗撤销都会把旧代软删卡错还原（污染）
|
||
· example_knowledge_embedding → **不动**：旧卡向量行靠判重召回 SQL 的 JOIN d.deleted=FALSE 天然排除，
|
||
软删状态保留可回溯；重抽边抽边嵌写新卡向量
|
||
|
||
用法：
|
||
reset_upgrade_work.py --work-id N 预览（只读统计，不写库）
|
||
reset_upgrade_work.py --work-id N --execute 真清（单事务，全清或全不清）
|
||
"""
|
||
import sys
|
||
import pathlib
|
||
|
||
import click
|
||
import psycopg
|
||
|
||
# 复用管线的连接与租户常量（与 parse_upgrade 同源）
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
||
from parse_llm import DSN, TENANT # noqa: E402
|
||
from upgrade_work_lock import UpgradeWorkLockUnavailable, upgrade_work_lock # noqa: E402
|
||
|
||
SOURCE_TYPE = "upgrade_book"
|
||
|
||
|
||
def _reset(work_id, execute):
|
||
"""在调用方已持有同书锁时预览或执行重抽清理。"""
|
||
|
||
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 count(*) FROM muse_knowledge_draft
|
||
WHERE tenant_id=%s AND work_id=%s AND source_type=%s AND deleted=FALSE""",
|
||
(TENANT, work_id, SOURCE_TYPE)).fetchone()[0]
|
||
wins = conn.execute(
|
||
"""SELECT count(*), count(*) FILTER (WHERE status='done') FROM example_upgrade_window
|
||
WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE""",
|
||
(TENANT, work_id)).fetchone()
|
||
ali = conn.execute("SELECT count(*) FROM example_upgrade_alias WHERE tenant_id=%s AND work_id=%s",
|
||
(TENANT, work_id)).fetchone()[0]
|
||
pres = conn.execute("SELECT count(*) FROM example_upgrade_presence WHERE tenant_id=%s AND work_id=%s",
|
||
(TENANT, work_id)).fetchone()[0]
|
||
# card_state 无 work 冗余错？——有 work_id 列（建表即有）；audit 无 work_id，靠 JOIN 卡定位
|
||
cs = conn.execute("SELECT count(*) FROM example_upgrade_card_state WHERE tenant_id=%s AND work_id=%s",
|
||
(TENANT, work_id)).fetchone()[0]
|
||
aud = conn.execute(
|
||
"""SELECT count(*) FROM example_upgrade_audit a
|
||
WHERE a.tenant_id=%s AND a.draft_id IN
|
||
(SELECT id FROM muse_knowledge_draft WHERE work_id=%s AND source_type=%s)""",
|
||
(TENANT, work_id, SOURCE_TYPE)).fetchone()[0]
|
||
click.echo(f"《{title}》work={work_id} 重抽准备{'（执行）' if execute else '（预览）'}：")
|
||
click.echo(f" 软删活升格卡 {cards} ｜ 重置窗 {wins[0]}(其中done {wins[1]}) ｜ "
|
||
f"硬删 别名{ali} 留档{pres} 卡水位{cs} 审计{aud}")
|
||
if not execute:
|
||
click.echo(" （预览模式未写库；加 --execute 真清）")
|
||
return
|
||
# 单事务执行：全清或全不清（中途失败自动回滚，不留半清状态）
|
||
conn.execute(
|
||
"""UPDATE muse_knowledge_draft SET deleted=TRUE, updater='upgrade-reset'
|
||
WHERE tenant_id=%s AND work_id=%s AND source_type=%s AND deleted=FALSE""",
|
||
(TENANT, work_id, SOURCE_TYPE))
|
||
conn.execute(
|
||
"""UPDATE example_upgrade_window SET status='pending', error_message=NULL,
|
||
updater='upgrade-reset' WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE""",
|
||
(TENANT, work_id))
|
||
# 审计先删（靠卡 JOIN 定位，卡还查得到——虽然刚软删，JOIN 不看 deleted）
|
||
conn.execute(
|
||
"""DELETE FROM example_upgrade_audit
|
||
WHERE tenant_id=%s AND draft_id IN
|
||
(SELECT id FROM muse_knowledge_draft WHERE work_id=%s AND source_type=%s)""",
|
||
(TENANT, work_id, SOURCE_TYPE))
|
||
conn.execute("DELETE FROM example_upgrade_alias WHERE tenant_id=%s AND work_id=%s",
|
||
(TENANT, work_id))
|
||
conn.execute("DELETE FROM example_upgrade_presence WHERE tenant_id=%s AND work_id=%s",
|
||
(TENANT, work_id))
|
||
conn.execute("DELETE FROM example_upgrade_card_state WHERE tenant_id=%s AND work_id=%s",
|
||
(TENANT, work_id))
|
||
conn.commit()
|
||
# 清后复核：底册应为空、窗应全 pending
|
||
left = conn.execute(
|
||
"""SELECT count(*) FROM muse_knowledge_draft
|
||
WHERE tenant_id=%s AND work_id=%s AND source_type=%s AND deleted=FALSE""",
|
||
(TENANT, work_id, SOURCE_TYPE)).fetchone()[0]
|
||
pend = conn.execute(
|
||
"""SELECT count(*) FILTER (WHERE status='pending'), count(*) FROM example_upgrade_window
|
||
WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE""", (TENANT, work_id)).fetchone()
|
||
click.echo(f" ✅ 清后复核：活升格卡 {left}（应0）｜ 窗 pending {pend[0]}/{pend[1]}（应全 pending）")
|
||
|
||
|
||
@click.command()
|
||
@click.option("--work-id", type=int, required=True)
|
||
@click.option("--execute", is_flag=True, help="真执行（默认只预览统计）")
|
||
def main(work_id, execute):
|
||
"""先获取同书会话锁，再预览或执行升格全量重抽准备。"""
|
||
|
||
try:
|
||
with upgrade_work_lock(DSN, TENANT, work_id):
|
||
return _reset(work_id, execute)
|
||
except UpgradeWorkLockUnavailable as exc:
|
||
raise click.ClickException(str(exc)) from exc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|