scripts/db_shrink_archive.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python
"""一次性瘦身 sessions.db —— 清理重复归档 + 重建 FTS + VACUUM.
背景（2026-10-03）:
* ``_context_govern`` 每个工具步都会把「已移出 LLM 视图」的 TOOL 消息重新
归档一遍，而 ``async_archive_messages`` 此前不做去重 → 实测
messages_archive 60,578 行里只有 3,977 行唯一（context_prune 54,879 行
去重后仅 1,295 行，42× 重复）。
* FTS5 只建了 AFTER INSERT 触发器，没有 DELETE → messages_fts_data 涨到
29MB（活跃消息正文仅 4.5MB），docsize 190,411 行里 186,419 行是孤儿。
两者叠加让 sessions.db 从「应该的 ~18MB」涨到 119.9MB。
代码侧的修复已在本次提交完成（写入端去重 + 补 FTS DELETE 触发器 + agent 层
只归档首次移出的消息）；本脚本用于**清理已经产生的存量数据**。
用法::
# 只看会做什么，不改任何东西（默认，推荐先跑一次）
python scripts/db_shrink_archive.py
# 实际执行（会自动先备份到 <db>.bak-<时间戳>）
python scripts/db_shrink_archive.py --apply
# 指定数据库（默认自动探测 ~/.scout/sessions.db）
python scripts/db_shrink_archive.py --db D:/.scout/sessions.db --apply
安全:
* ``--apply`` 前强制做整文件备份（shutil.copy2），失败即中止。
* 全程在单个事务里做 DELETE，中断则整体回滚。
* VACUUM 需要额外磁盘空间（约等于库大小）。
"""
from __future__ import annotations
import argparse
import os
import shutil
import sqlite3
import sys
import time
from pathlib import Path
def _human(n: float) -> str:
for unit in ("B", "KB", "MB", "GB"):
if n < 1024 or unit == "GB":
return f"{n:.1f}{unit}"
n /= 1024
return f"{n:.1f}GB"
def default_db_path() -> Path | None:
"""探测运行时数据目录里的 sessions.db（与 launcher 的 data_dir() 一致）."""
try:
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from scout.core.paths import data_dir # type: ignore
p = Path(data_dir()) / "sessions.db"
if p.exists():
return p
except Exception:
pass
for cand in (
Path.home() / ".scout" / "sessions.db",
Path("D:/.scout/sessions.db"),
Path("data/scout.db"),
):
if cand.exists():
return cand
return None
def has_column(conn: sqlite3.Connection, table: str, col: str) -> bool:
return col in {r[1] for r in conn.execute(f"PRAGMA table_info({table})")}
def dedup_group_expr(conn: sqlite3.Connection) -> str:
"""去重分组表达式 —— 新库用 content_hash，存量库退回 content 全文."""
if has_column(conn, "messages_archive", "content_hash"):
return "session_id, role, COALESCE(content_hash, content)"
return "session_id, role, content"
def analyze(conn: sqlite3.Connection) -> dict:
cur = conn.execute("PRAGMA page_size")
page_size = cur.fetchone()[0]
cur = conn.execute("PRAGMA page_count")
page_count = cur.fetchone()[0]
grp = dedup_group_expr(conn)
total = conn.execute("SELECT COUNT(*) FROM messages_archive").fetchone()[0]
uniq = conn.execute(
"SELECT COUNT(*) FROM ("
f" SELECT MIN(id) FROM messages_archive GROUP BY {grp}"
")"
).fetchone()[0]
msgs = conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
try:
docsize = conn.execute("SELECT COUNT(*) FROM messages_fts_docsize").fetchone()[0]
except Exception:
docsize = -1
return {
"size": page_size * page_count,
"archive_total": total,
"archive_uniq": uniq,
"archive_dup": total - uniq,
"messages": msgs,
"fts_docsize": docsize,
}
def main() -> int:
ap = argparse.ArgumentParser(description="清理 sessions.db 的重复归档并重建 FTS")
ap.add_argument("--db", default=None, help="sessions.db 路径（默认自动探测）")
ap.add_argument("--apply", action="store_true", help="实际执行；不带则只做 dry-run")
ap.add_argument("--no-backup", action="store_true", help="跳过备份（不推荐）")
ap.add_argument("--keep-fts", action="store_true", help="跳过 FTS 重建")
args = ap.parse_args()
db = Path(args.db) if args.db else default_db_path()
if db is None or not db.exists():
print("找不到 sessions.db，请用 --db 指定路径")
return 1
print(f"数据库: {db}")
conn = sqlite3.connect(str(db))
before = analyze(conn)
print("\n── 当前状态 ──────────────────────────────")
print(f" 文件体积 : {_human(before['size'])}")
print(f" 归档总行数 : {before['archive_total']}")
print(f" 去重后唯一行 : {before['archive_uniq']}")
print(f" 重复行数 : {before['archive_dup']} "
f"(占 {before['archive_dup'] / max(before['archive_total'], 1) * 100:.1f}%)")
print(f" 活跃消息行 : {before['messages']}")
print(f" FTS docsize 行 : {before['fts_docsize']}")
if not args.apply:
dup = before["archive_dup"]
print(f"\n[dry-run] 将删除约 {dup} 行重复归档、重建 FTS 并 VACUUM。")
print("[dry-run] 实际执行请加 --apply（会自动备份）。")
conn.close()
return 0
# ── 备份 ──────────────────────────────────────────────
if not args.no_backup:
bak = db.with_suffix(db.suffix + f".bak-{time.strftime('%Y%m%d-%H%M%S')}")
print(f"\n备份到: {bak}")
conn.close()
try:
shutil.copy2(db, bak)
except Exception as e:
print(f"备份失败，已中止: {e}")
return 1
conn = sqlite3.connect(str(db))
else:
print("\n[warn] 已跳过备份")
# ── 清理 ──────────────────────────────────────────────
print("\n删除重复归档 ...")
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("BEGIN")
try:
conn.execute(
"DELETE FROM messages_archive WHERE id NOT IN ("
f" SELECT MIN(id) FROM messages_archive GROUP BY {dedup_group_expr(conn)}"
")"
)
deleted = conn.execute("SELECT changes()").fetchone()[0]
conn.commit()
except Exception as e:
conn.rollback()
print(f"删除失败，已回滚: {e}")
conn.close()
return 1
print(f" 已删除 {deleted} 行")
if not args.keep_fts:
print("重建 FTS 索引 ...")
try:
conn.execute("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')")
conn.commit()
except Exception as e:
print(f" FTS 重建失败（可忽略）: {e}")
print("VACUUM ...")
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
try:
conn.execute("VACUUM")
except Exception as e:
print(f" VACUUM 失败（可忽略）: {e}")
after = analyze(conn)
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
conn.close()
real_after = os.path.getsize(db)
print("\n── 完成 ──────────────────────────────────")
print(f" 归档行 : {before['archive_total']} → {after['archive_total']}")
print(f" FTS docsize 行 : {before['fts_docsize']} → {after['fts_docsize']}")
print(f" 逻辑体积 : {_human(before['size'])} → {_human(after['size'])}")
print(f" 实际文件 : {_human(os.path.getsize(db)) if False else _human(real_after)} "
f"(省下 {_human(max(before['size'] - real_after, 0))})")
return 0
if __name__ == "__main__":
raise SystemExit(main())