#!/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())