Antigravity CLI の会話データベースは、重複を消しても 1 バイトも縮みません

#!/usr/bin/env python3
"""会話データベースの容量を、テーブル単位と重複追記の観点で棚卸しする。
使い方: python3 agdb_audit.py /path/to/conversation.db
読み取り専用で開くため、CLI を起動したままでも実行できる。"""
import sys, sqlite3, os
 
def open_ro(path):
    # mode=ro で開く。書き込み側を止める必要がない。
    return sqlite3.connect(f"file:{path}?mode=ro", uri=True)
 
def table_bytes(cur):
    """dbstat があればテーブル別の実ページ使用量を返す。無ければ None。"""
    try:
        rows = cur.execute(
            "SELECT name, SUM(pgsize) FROM dbstat GROUP BY name ORDER BY 2 DESC"
        ).fetchall()
    except sqlite3.OperationalError:
        return None
    return [(n, b) for n, b in rows if b]
 
def table_bytes_fallback(cur):
    """dbstat 無しの環境向け。行データの実バイト長を列ごとに合算した概算。"""
    out = []
    names = [r[0] for r in cur.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")]
    for n in names:
        cnt = cur.execute(f'SELECT COUNT(*) FROM "{n}"').fetchone()[0]
        if cnt == 0:
            out.append((n, 0, 0))
            continue
        cols = [r[1] for r in cur.execute(f'PRAGMA table_info("{n}")')]
        expr = " + ".join(f'COALESCE(LENGTH(CAST("{c}" AS BLOB)),0)' for c in cols)
        total = cur.execute(f'SELECT SUM({expr}) FROM "{n}"').fetchone()[0] or 0
        out.append((n, total, cnt))
    return sorted(out, key=lambda r: -r[1])
 
def duplicate_report(cur, min_repeat=3, top=8):
    """主キー列を除く全列が一致する行を数える。
    起動のたびに同じ内容が積まれる型の膨張は、ここに露出する。"""
    result = []
    names = [r[0] for r in cur.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")]
    for n in names:
        info = list(cur.execute(f'PRAGMA table_info("{n}")'))
        cols = [r[1] for r in info if not r[5]]        # r[5] が 1 なら主キー列
        if not cols:
            continue
        key = ", ".join(f'"{c}"' for c in cols)
        total = cur.execute(f'SELECT COUNT(*) FROM "{n}"').fetchone()[0]
        if total == 0:
            continue
        distinct = cur.execute(
            f'SELECT COUNT(*) FROM (SELECT DISTINCT {key} FROM "{n}")').fetchone()[0]
        dup = total - distinct
        if dup <= 0:
            continue
        worst = cur.execute(
            f'SELECT COUNT(*) AS c, {key} FROM "{n}" GROUP BY {key} '
            f'HAVING c >= ? ORDER BY c DESC LIMIT ?', (min_repeat, top)).fetchall()
        result.append((n, total, distinct, dup, worst))
    return sorted(result, key=lambda r: -r[3])
 
def main():
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(2)
    path = sys.argv[1]
    if not os.path.exists(path):
        print(f"ファイルがありません: {path}")
        sys.exit(1)
    con = open_ro(path)
    cur = con.cursor()
    psize = cur.execute("PRAGMA page_size").fetchone()[0]
    pages = cur.execute("PRAGMA page_count").fetchone()[0]
    free = cur.execute("PRAGMA freelist_count").fetchone()[0]
    size = os.path.getsize(path)
    print(f"[file] {size:,} bytes / page_size={psize} / page_count={pages:,}")
    print(f"[file] freelist={free:,} ページ = {free * psize:,} bytes")
 
    print("\n[容量の内訳]")
    tb = table_bytes(cur)
    if tb:
        for n, b in tb[:12]:
            print(f"  {b:>12,} bytes  {b * 100.0 / size:5.1f}%  {n}")
    else:
        print("  dbstat が無いため概算（行データの合計長）で表示します")
        for n, b, cnt in table_bytes_fallback(cur)[:12]:
            print(f"  {b:>12,} bytes  {cnt:>9,} rows  {n}")
 
    print("\n[重複追記の検出]")
    rep = duplicate_report(cur)
    if not rep:
        print("  完全一致の重複行はありません")
    for n, total, distinct, dup, worst in rep[:6]:
        print(f"  {n}: {total:,} 行中 {dup:,} 行が重複"
              f"（実質 {distinct:,} 種類 / 重複率 {dup * 100.0 / total:.1f}%）")
        for row in worst[:3]:
            vals = " | ".join(str(v)[:40] for v in row[1:])
            print(f"      x{row[0]:<5} {vals}")
    con.close()
 
if __name__ == "__main__":
    main()