scripts/find_name_duplicate_clusters.py

#!/usr/bin/env python3
"""Report clusters of kinks whose titles normalize to the same fingerprint (near-duplicates).

Uses the same rules as frontend/name-fingerprint.js (keep in sync when tuning).

Usage:
  python scripts/find_name_duplicate_clusters.py [--min-size 2] [--limit 40] [--db data/store.db]
"""
from __future__ import annotations

import argparse
import re
import sqlite3
from collections import defaultdict
from pathlib import Path


def fingerprint_play_title(raw: str) -> str:
    s = (raw or "").strip().lower()
    if not s:
        return ""
    s = re.sub(r"[\u2018\u2019\u201c\u201d`´]", "'", s)
    s = re.sub(r"\s+", " ", s)
    s = re.sub(r"\bgood\s+morning\s+", "goodmorning ", s)
    s = re.sub(r"\bdeep\s*throating\b", "deepthroat", s)
    s = re.sub(r"\bdeepthroating\b", "deepthroat", s)
    s = re.sub(r"\bdeep\s*throat\b", "deepthroat", s)
    s = re.sub(r"\bblow\s+jobs\b", "blowjob", s, flags=re.I)
    s = re.sub(r"\bblow\s+job\b", "blowjob", s, flags=re.I)
    s = re.sub(r"\bblowjobs\b", "blowjob", s)
    s = re.sub(r"\bstrap\s+ons\b", "strapon", s, flags=re.I)
    s = re.sub(r"\bstrap\s+on\b", "strapon", s, flags=re.I)
    s = re.sub(r"\bstrapons\b", "strapon", s)
    s = re.sub(r"[^\w\s]", " ", s)
    s = re.sub(r"\s+", " ", s).strip()
    s = re.sub(r"\s+", "", s)
    return s


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--db", type=Path, default=Path(__file__).resolve().parent.parent / "data" / "store.db")
    ap.add_argument("--min-size", type=int, default=2)
    ap.add_argument("--limit", type=int, default=40, help="max clusters to print")
    args = ap.parse_args()

    conn = sqlite3.connect(args.db)
    conn.row_factory = sqlite3.Row
    rows = conn.execute("SELECT id, name FROM kink").fetchall()
    buckets: dict[str, list[tuple[str, str]]] = defaultdict(list)
    for r in rows:
        fp = fingerprint_play_title(r["name"])
        if len(fp) < 4:
            continue
        buckets[fp].append((r["id"], r["name"]))

    clusters = [(fp, v) for fp, v in buckets.items() if len(v) >= args.min_size]
    clusters.sort(key=lambda x: (-len(x[1]), x[0]))

    print(f"Total kinks: {len(rows)}")
    print(f"Clusters (size >= {args.min_size}): {len(clusters)}\n")

    for fp, items in clusters[: args.limit]:
        print(f"[{len(items)}] {fp!r}")
        for kid, nm in sorted(items, key=lambda x: x[1].lower())[:20]:
            print(f"    {kid}\t{nm}")
        if len(items) > 20:
            print(f"    ... +{len(items) - 20} more")
        print()


if __name__ == "__main__":
    main()