scripts/build_slim_store.py

#!/usr/bin/env python3
"""
Copy the SQLite store, compact derived similarity state, rebuild indexes, and VACUUM.

This mirrors the post-merge maintenance in wipe_users_and_merge_title_duplicates.py:
  - prune duplicate self-edges and duplicate (left, right, type) rows
  - replace ``catalog`` edges with a capped mirror of ``fetlife_similar`` (no sklearn TF-IDF)
  - optionally rebuild ``fetlife_collaborative`` edges (off by default; a full rebuild can
    change edge counts vs an older snapshot and grow the file)
  - rebuild FTS and catalog caches
  - VACUUM + PRAGMA optimize

Requires the same environment as the app (project venv). Large databases may take several minutes.

Usage:
  python scripts/build_slim_store.py [--source data/store.db] [--dest data/store_slim.db] [--force] [--rebuild-collab]
"""
from __future__ import annotations

import argparse
import importlib.util
import sys
from pathlib import Path


def _load_wipe_helpers(root: Path):
    path = root / "scripts" / "wipe_users_and_merge_title_duplicates.py"
    spec = importlib.util.spec_from_file_location("wipe_helpers", path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _backup_sqlite(src: Path, dst: Path) -> None:
    import sqlite3

    dst.parent.mkdir(parents=True, exist_ok=True)
    if dst.exists():
        dst.unlink()
    s = sqlite3.connect(src)
    d = sqlite3.connect(dst)
    try:
        s.backup(d)
    finally:
        d.close()
        s.close()


def _file_mb(p: Path) -> float:
    return p.stat().st_size / (1024 * 1024)


def main() -> int:
    root = Path(__file__).resolve().parent.parent
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--source", type=Path, default=root / "data" / "store.db")
    ap.add_argument("--dest", type=Path, default=root / "data" / "store_slim.db")
    ap.add_argument("--force", action="store_true", help="Overwrite --dest if it exists.")
    ap.add_argument(
        "--rebuild-collab",
        action="store_true",
        help="Delete and recompute fetlife_collaborative edges from fetlifeuserfetish (slow; can differ from prior snapshot).",
    )
    args = ap.parse_args()
    src = args.source.resolve()
    dst = args.dest.resolve()

    if not src.is_file():
        print(f"Source database not found: {src}", file=sys.stderr)
        return 1
    if dst.exists() and not args.force:
        print(f"Refusing to overwrite {dst} (use --force).", file=sys.stderr)
        return 1

    import sqlite3

    print(f"Source: {src} ({_file_mb(src):.1f} MiB)", flush=True)
    print(f"Backup → {dst} …", flush=True)
    _backup_sqlite(src, dst)
    print(f"Copied: {_file_mb(dst):.1f} MiB", flush=True)

    helpers = _load_wipe_helpers(root)
    conn = sqlite3.connect(dst)
    conn.row_factory = sqlite3.Row
    try:
        helpers.prune_similarity_edges(conn)
        helpers.repopulate_catalog_edges_from_fetlife_similar(conn)
        conn.commit()
    finally:
        conn.close()

    sys.path.insert(0, str(root))
    from backend.core import Backend

    print("Refreshing catalog / FTS …", flush=True)
    backend = Backend(dst)
    try:
        if args.rebuild_collab:
            print("Rebuilding FetLife collaborative edges …", flush=True)
            backend.rebuild_fetlife_collaborative_edges()
        backend.refresh_cache()
    finally:
        backend.engine.dispose()

    conn = sqlite3.connect(dst)
    try:
        conn.execute("VACUUM")
        conn.execute("PRAGMA optimize")
        conn.commit()
    finally:
        conn.close()

    wal = Path(str(dst) + "-wal")
    shm = Path(str(dst) + "-shm")
    for extra in (wal, shm):
        if extra.exists():
            try:
                extra.unlink()
            except OSError:
                pass

    print(f"Done: {dst} ({_file_mb(dst):.1f} MiB)", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())