scripts/dev_reset_local.py

#!/usr/bin/env python3
"""Local dev: optionally free API port(s), then delete all user profiles from SQLite.

Does **not** delete kinks, catalog edges, or FetLife crawl tables — only user-scoped rows.

Examples:
  python scripts/dev_reset_local.py --db data/store.db --dry-run
  python scripts/dev_reset_local.py --db data/store.db --port 8011 --i-am-sure
"""
from __future__ import annotations

import argparse
import os
import sqlite3
import subprocess
import sys
from pathlib import Path

_DEFAULT_PORTS = (8011,)


def _kill_ports(ports: list[int], *, dry_run: bool) -> None:
    for port in ports:
        if dry_run:
            print(f"[dry-run] would attempt to free TCP port {port}")
            continue
        # fuser is common on Linux; lsof fallback for macOS-ish environments.
        r = subprocess.run(
            ["fuser", "-k", f"{port}/tcp"],
            capture_output=True,
            text=True,
        )
        if r.returncode == 0:
            print(f"freed port {port} (fuser)")
            continue
        r2 = subprocess.run(["lsof", f"-ti:{port}"], capture_output=True, text=True)
        pids = [int(pid) for pid in r2.stdout.split() if pid.isdigit()]
        for pid in pids:
            os.kill(pid, 9)
        if pids:
            print(f"freed port {port} (lsof)")
        else:
            print(f"note: no listener freed on port {port} (fuser/lsof)", file=sys.stderr)


def _wipe_all_users(conn: sqlite3.Connection) -> None:
    """Delete user-linked rows then users (order safe for typical FK graphs)."""
    conn.executescript(
        """
        DELETE FROM partnergroupmember;
        DELETE FROM partnergroup;
        DELETE FROM partnerlinkrequest;
        DELETE FROM partnerlink;
        DELETE FROM scenariopreference;
        DELETE FROM promptdismissal;
        DELETE FROM rolepreference;
        DELETE FROM playpreference;
        DELETE FROM userpreference;
        DELETE FROM user;
        """
    )


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--db", type=Path, required=True, help="path to SQLite store")
    ap.add_argument(
        "--port",
        type=int,
        action="append",
        default=[],
        help="TCP port to try to kill (repeatable); default 8011 if none given",
    )
    ap.add_argument("--dry-run", action="store_true", help="print actions only; no DB or port changes")
    ap.add_argument(
        "--i-am-sure",
        action="store_true",
        help="required with --dry-run false to apply destructive DB deletes",
    )
    ap.add_argument(
        "--skip-port-kill",
        action="store_true",
        help="do not run fuser/lsof (use in CI / when only DB wipe is wanted)",
    )
    args = ap.parse_args()

    db = args.db.resolve()
    if not db.is_file():
        print(f"database not found: {db}", file=sys.stderr)
        return 1

    ports = [] if args.skip_port_kill else (args.port if args.port else list(_DEFAULT_PORTS))

    if args.dry_run:
        print(f"[dry-run] db={db}")
        if ports:
            _kill_ports(ports, dry_run=True)
        else:
            print("[dry-run] skip port kill")
        print("[dry-run] would DELETE all user-scoped tables then user rows (see script source)")
        return 0

    if not args.i_am_sure:
        print("Refusing to modify database without --i-am-sure (use --dry-run to preview).", file=sys.stderr)
        return 2

    if ports:
        _kill_ports(ports, dry_run=False)

    conn = sqlite3.connect(db)
    try:
        conn.execute("PRAGMA foreign_keys=OFF")
        _wipe_all_users(conn)
        conn.commit()
    finally:
        conn.execute("PRAGMA foreign_keys=ON")
        conn.close()

    print(f"wrote user wipe to {db}")
    return 0


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