scripts/diagnose_partner_overlap.py

#!/usr/bin/env python3
"""Diagnose why Together \"Matched\" overlap may differ from expectations for a user.

Usage:
  KINK_STORE_PATH=/path/to/store.db python scripts/diagnose_partner_overlap.py [user_id]
  # Fast (no catalog load): SQL only
  python scripts/diagnose_partner_overlap.py vocal-gnu-15 --sql-only

Compares (unless --sql-only):
  - Pairwise GET /users/{id}/overlap/{partner} (backend.shared_play_list)
  - Each partner group's shared_play_list_for_group (what the Together UI uses)

Common reasons for differences:
  - **Solo partner group**: the owner's ``partnergroupmember`` rows may omit the linked
    partner (only a self row for share prefs). Overlap needs ≥2 participants → empty
    Matched for that group, while the partner's group or pairwise overlap is non-empty.
  - Together tab uses the *selected* partner group. With 3+ real members, Matched is
    the intersection across *everyone*, not just you + one partner.
  - The green \"Match\" badge only appears for love/like pairs (not curious); rows can
    still list curious + curious.

Requires KINK_STORE_PATH or default data/store_slim.db under repo (if present).
"""
from __future__ import annotations

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


def _repo_root() -> Path:
    return Path(__file__).resolve().parents[1]


def _default_db() -> Path:
    env = os.environ.get("KINK_STORE_PATH", "").strip()
    if env:
        return Path(env)
    return _repo_root() / "data" / "store_slim.db"


def _sql_group_snapshot(store: Path, user_id: str) -> dict:
    """Participant counts from SQLite only (matches backend _partner_group_dict logic)."""
    conn = sqlite3.connect(store)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    partners = [
        r[0]
        for r in cur.execute(
            "SELECT CASE WHEN left_user_id = ? THEN right_user_id ELSE left_user_id END "
            "FROM partnerlink WHERE left_user_id = ? OR right_user_id = ?",
            (user_id, user_id, user_id),
        ).fetchall()
    ]
    groups_out: list[dict] = []
    # Mirror list_partner_groups: owned groups first, then guest groups (same viewer).
    seen: set[str] = set()
    rows: list[sqlite3.Row] = []
    for r in cur.execute(
        "SELECT id, owner_user_id, name FROM partnergroup WHERE owner_user_id = ? ORDER BY name, id",
        (user_id,),
    ):
        rows.append(r)
        seen.add(r["id"])
    for r in cur.execute(
        """
        SELECT g.id, g.owner_user_id, g.name FROM partnergroup g
        JOIN partnergroupmember m ON m.group_id = g.id AND m.member_user_id = ?
        WHERE g.owner_user_id != ?
        ORDER BY g.name, g.id
        """,
        (user_id, user_id),
    ):
        if r["id"] not in seen:
            rows.append(r)
            seen.add(r["id"])
    for g in rows:
        gid = g["id"]
        owner_id = g["owner_user_id"]
        mems = cur.execute(
            "SELECT member_user_id, share_full_list FROM partnergroupmember WHERE group_id = ?",
            (gid,),
        ).fetchall()
        member_ids = sorted(
            {m["member_user_id"] for m in mems if m["member_user_id"] != owner_id}
        )
        participant_ids: list[str] = []
        for uid in (owner_id, *member_ids):
            if uid not in participant_ids:
                participant_ids.append(uid)
        groups_out.append(
            {
                "id": gid,
                "name": g["name"],
                "owner_user_id": owner_id,
                "member_ids": member_ids,
                "participant_ids": participant_ids,
                "participant_count": len(participant_ids),
                "overlap_will_be_empty": len(participant_ids) < 2,
            }
        )
    conn.close()
    return {"user_id": user_id, "partners": sorted(partners), "groups": groups_out}


def main() -> int:
    parser = argparse.ArgumentParser(description="Diagnose partner / group overlap for one user.")
    parser.add_argument("user_id", nargs="?", default="vocal-gnu-15", help="Public user id (default: vocal-gnu-15)")
    parser.add_argument(
        "--store-path",
        type=Path,
        default=None,
        help="SQLite store path (overrides KINK_STORE_PATH)",
    )
    parser.add_argument(
        "--sql-only",
        action="store_true",
        help="Only query SQLite (instant). Detects solo participant groups without loading catalog.",
    )
    args = parser.parse_args()
    store = args.store_path or _default_db()
    if not store.is_file():
        print(f"ERROR: database not found: {store}", file=sys.stderr)
        print("Set KINK_STORE_PATH or pass --store-path to your production store.", file=sys.stderr)
        return 2

    if args.sql_only:
        snap = _sql_group_snapshot(store, args.user_id)
        print(json.dumps(snap, indent=2))
        return 0

    os.environ.setdefault("KINK_SKIP_HEAVY_WARM", "1")
    sys.path.insert(0, str(_repo_root()))
    from backend.core import Backend  # noqa: E402

    backend = Backend(store)
    uid = args.user_id
    user = backend.get_user(uid)
    if not user:
        print(json.dumps({"error": "unknown_user", "user_id": uid, "store": str(store)}, indent=2))
        return 1

    partners = user.get("partners") or []
    groups = user.get("partner_groups") or []

    def direct_ids(ov: dict) -> list[str]:
        return sorted(k["id"] for k in ov.get("direct_matches") or [])

    report: dict = {
        "user_id": uid,
        "store": str(store),
        "partners": partners,
        "groups": [],
        "pairwise_vs_groups": [],
        "sql_snapshot": _sql_group_snapshot(store, uid),
    }

    for g in groups:
        gid = g["id"]
        pids = g.get("participant_ids") or []
        ov = backend.shared_play_list_for_group(uid, gid)
        report["groups"].append(
            {
                "id": gid,
                "name": g.get("name"),
                "owner_user_id": g.get("owner_user_id"),
                "participant_ids": pids,
                "participant_count": len(pids),
                "direct_match_count": len(ov.get("direct_matches") or []),
                "direct_match_ids": direct_ids(ov),
            }
        )

    for partner_id in partners:
        pair_ov = backend.shared_play_list(uid, partner_id)
        pair_set = set(direct_ids(pair_ov))
        row = {
            "partner_id": partner_id,
            "pairwise_direct_match_ids": sorted(pair_set),
            "per_group": [],
        }
        for g in groups:
            gid = g["id"]
            pids = set(g.get("participant_ids") or [])
            if partner_id not in pids or uid not in pids:
                continue
            gov = backend.shared_play_list_for_group(uid, gid)
            gset = set(direct_ids(gov))
            row["per_group"].append(
                {
                    "group_id": gid,
                    "group_name": g.get("name"),
                    "participant_ids": g.get("participant_ids"),
                    "only_in_pairwise_not_group": sorted(pair_set - gset),
                    "only_in_group_not_pairwise": sorted(gset - pair_set),
                    "same": pair_set == gset,
                }
            )
        report["pairwise_vs_groups"].append(row)

    print(json.dumps(report, indent=2))
    return 0


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