backend/benchmark_pair_sources.py

"""Strategies for sourcing **similar kink pairs** into the normalization / merge benchmark.

Human labeling workflow (``merge`` | ``alias_only`` | ``reject``) starts from **candidate pairs**
drawn from the live catalog. Use :func:`collect_all_pair_records` (or the CLI
``scripts/source_normalization_benchmark_pairs.py``) to emit JSONL, then curate rows into
``tests/fixtures/normalization_benchmark_v1.json`` (or a future ``*_pairs.json``).

Tiers (in priority order for review — high signal first):

1. **merge_signature_same_bucket** — Same ``(cluster, merge_signature(name))`` and at least two
   rows. These are exactly what :func:`backend.kink_merge.propose_merge_clusters` groups; the
   canonical/duplicate split is popularity-driven. Pairs are unordered; cap large buckets with
   ``max_pairs_per_bucket`` and optional RNG sampling.

2. **fingerprint_collision** — Same title **fingerprint** (see ``scripts/find_name_duplicate_clusters.py``
   / ``frontend/name-fingerprint.js``) but **different** ``merge_signature``. Often “soft”
   near-duplicates (e.g. ``blowjob`` vs ``Blow job``): good **negatives** for hard-merge or
   **alias_only** positives — *never* auto-merge on signature key alone.

3. **similarity_edge** — Rows from ``similarityedge`` above a score floor (``catalog``,
   ``fetlife_similar``, …). Catches graph-based similarity that signatures miss; label separately
   because many edges are *related* but not title duplicates.

**Sampling:** After union of tiers, apply ``sample_random`` with a fixed ``seed`` so exports are
reproducible for diffs and partial review batches.

**Metrics note:** Pair-level precision/recall needs frozen labels on ``pair_key``; cluster-level
tests in ``tests/test_normalization_benchmark.py`` stay orthogonal.
"""
from __future__ import annotations

import random
import re
import sqlite3
from collections import defaultdict
from collections.abc import Iterable, Iterator, Sequence
from typing import Any

from backend.kink_merge import merge_graph_eligible, merge_signature


def fingerprint_play_title(raw: str) -> str:
    """Collapse play titles for duplicate-ish grouping (keep in sync with find_name_duplicate_clusters / name-fingerprint.js)."""
    s = (raw or "").strip().lower()
    if not s:
        return ""
    s = re.sub(r"[\u2018\u2019\u201c\u201d\u0060\u00b4]", "'", 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.IGNORECASE)
    s = re.sub(r"\bblow\s+job\b", "blowjob", s, flags=re.IGNORECASE)
    s = re.sub(r"\bblowjobs\b", "blowjob", s)
    s = re.sub(r"\bstrap\s+ons\b", "strapon", s, flags=re.IGNORECASE)
    s = re.sub(r"\bstrap\s+on\b", "strapon", s, flags=re.IGNORECASE)
    s = re.sub(r"\bstrapons\b", "strapon", s)
    s = re.sub(r"[^\w\s]", " ", s)
    return re.sub(r"\s+", "", re.sub(r"\s+", " ", s).strip())


def _row_dict(kink_id: str, name: str, cluster: str, popularity: float) -> dict[str, Any]:
    return {
        "kink_id": kink_id,
        "name": name,
        "cluster": cluster,
        "popularity": float(popularity or 0.0),
    }


def _merge_sig_bucket_key(name: str, cluster: str) -> tuple[str, str] | None:
    if not merge_graph_eligible(name):
        return None
    return (cluster, merge_signature(name))


def _sample_pair_indices(n: int, max_pairs: int, rng: random.Random | None) -> Iterator[tuple[int, int]]:
    """Yield up to ``max_pairs`` unordered pairs (i, j) with i < j from range(n)."""
    total = n * (n - 1) // 2
    if total <= max_pairs:
        for i in range(n):
            for j in range(i + 1, n):
                yield i, j
        return
    if rng is None:
        rng = random.Random(0)  # noqa: S311 — deterministic fallback cap when caller omitted RNG
    seen: set[tuple[int, int]] = set()
    attempts = 0
    max_attempts = max_pairs * 20
    while len(seen) < max_pairs and attempts < max_attempts:
        attempts += 1
        i, j = sorted((rng.randrange(n), rng.randrange(n)))
        if i == j:
            continue
        a, b = (i, j) if i < j else (j, i)
        if (a, b) in seen:
            continue
        seen.add((a, b))
        yield a, b


def collect_merge_signature_pairs(
    rows: Sequence[tuple[str, str, str, float]],
    *,
    max_pairs_per_bucket: int = 40,
    rng: random.Random | None = None,
) -> list[dict[str, Any]]:
    """Pairs that share the same merge-signature bucket (same cluster)."""
    buckets: dict[tuple[str, str], list[tuple[str, str, str, float]]] = defaultdict(list)
    for kid, name, cluster, pop in rows:
        key = _merge_sig_bucket_key(name, cluster)
        if key is None:
            continue
        buckets[key].append((kid, name, cluster, float(pop or 0.0)))

    out: list[dict[str, Any]] = []
    for (cluster, sig), group in sorted(buckets.items()):
        if len(group) < 2:
            continue
        group.sort(key=lambda t: t[0])
        for i, j in _sample_pair_indices(len(group), max_pairs_per_bucket, rng):
            a, b = group[i], group[j]
            lo, hi = sorted((a, b), key=lambda t: t[0])
            out.append(
                {
                    "strategy": "merge_signature_same_bucket",
                    "cluster": cluster,
                    "merge_signature": sig,
                    "fingerprint": fingerprint_play_title(lo[1]),
                    "left": _row_dict(lo[0], lo[1], lo[2], lo[3]),
                    "right": _row_dict(hi[0], hi[1], hi[2], hi[3]),
                    "pair_key": f"{lo[0]}|{hi[0]}",
                },
            )
    return out


def collect_fingerprint_collision_pairs(
    rows: Sequence[tuple[str, str, str, float]],
    *,
    min_fingerprint_len: int = 4,
    max_pairs_per_fingerprint: int = 30,
    rng: random.Random | None = None,
) -> list[dict[str, Any]]:
    """Same collapsed fingerprint, different merge_signature (subset / alias / reject review)."""
    by_fp: dict[str, list[tuple[str, str, str, float]]] = defaultdict(list)
    for kid, name, cluster, pop in rows:
        fp = fingerprint_play_title(name)
        if len(fp) < min_fingerprint_len:
            continue
        by_fp[fp].append((kid, name, cluster, float(pop or 0.0)))

    out: list[dict[str, Any]] = []
    for fp, group in sorted(by_fp.items(), key=lambda x: (-len(x[1]), x[0])):
        if len(group) < 2:
            continue
        sigs: dict[str, list[tuple[str, str, str, float]]] = defaultdict(list)
        for item in group:
            sigs[merge_signature(item[1])].append(item)
        if len(sigs) < 2:
            continue
        flat: list[tuple[str, str, str, float]] = []
        for g in sigs.values():
            flat.extend(g)
        candidates = [
            (i, j)
            for i in range(len(flat))
            for j in range(i + 1, len(flat))
            if merge_signature(flat[i][1]) != merge_signature(flat[j][1])
        ]
        if not candidates:
            continue
        if len(candidates) > max_pairs_per_fingerprint:
            r = rng or random.Random(0)  # noqa: S311
            r.shuffle(candidates)
            candidates = candidates[:max_pairs_per_fingerprint]
        for i, j in candidates:
            a, b = flat[i], flat[j]
            lo, hi = sorted((a, b), key=lambda t: t[0])
            out.append(
                {
                    "strategy": "fingerprint_collision",
                    "fingerprint": fp,
                    "merge_signature_left": merge_signature(lo[1]),
                    "merge_signature_right": merge_signature(hi[1]),
                    "left": _row_dict(lo[0], lo[1], lo[2], lo[3]),
                    "right": _row_dict(hi[0], hi[1], hi[2], hi[3]),
                    "pair_key": f"{lo[0]}|{hi[0]}",
                },
            )
    return out


def collect_similarity_edge_pairs(
    conn: sqlite3.Connection,
    *,
    min_score: float = 0.82,
    similarity_types: Sequence[str] | None = None,
    max_pairs: int = 50_000,
) -> list[dict[str, Any]]:
    """Pairs linked in ``similarityedge`` (graph similarity, not signature identity)."""
    types = tuple(similarity_types) if similarity_types else ("catalog", "fetlife_similar")
    placeholders = ",".join("?" * len(types))
    q = f"""
    SELECT s.similarity_type, s.score, s.method,
           l.id AS lid, l.name AS lname, l.cluster AS lcluster, COALESCE(ml.popularity, 0) AS lpop,
           r.id AS rid, r.name AS rname, r.cluster AS rcluster, COALESCE(mr.popularity, 0) AS rpop
    FROM similarityedge s
    JOIN kink l ON l.id = s.left_kink_id
    JOIN kink r ON r.id = s.right_kink_id
    LEFT JOIN fetlifekinkmeta ml ON ml.kink_id = l.id
    LEFT JOIN fetlifekinkmeta mr ON mr.kink_id = r.id
    WHERE s.score >= ?
      AND s.similarity_type IN ({placeholders})
      AND s.left_kink_id < s.right_kink_id
    """  # noqa: S608 — IN list is only `?` placeholders; values bound via parameters
    cur = conn.execute(q, (min_score, *types))
    out: list[dict[str, Any]] = []
    for row in cur.fetchall():
        if len(out) >= max_pairs:
            break
        lid, rid = str(row["lid"]), str(row["rid"])
        lname, rname = str(row["lname"]), str(row["rname"])
        lcl, rcl = str(row["lcluster"]), str(row["rcluster"])
        lpop, rpop = float(row["lpop"] or 0), float(row["rpop"] or 0)
        if lid > rid:
            lid, rid, lname, rname, lcl, rcl, lpop, rpop = rid, lid, rname, lname, rcl, lcl, rpop, lpop
        lo = (lid, lname, lcl, lpop)
        hi = (rid, rname, rcl, rpop)
        out.append(
            {
                "strategy": "similarity_edge",
                "similarity_type": str(row["similarity_type"]),
                "score": float(row["score"]),
                "method": str(row["method"] or ""),
                "merge_signature_left": merge_signature(lo[1]),
                "merge_signature_right": merge_signature(hi[1]),
                "fingerprint_left": fingerprint_play_title(lo[1]),
                "fingerprint_right": fingerprint_play_title(hi[1]),
                "left": _row_dict(lo[0], lo[1], lo[2], lo[3]),
                "right": _row_dict(hi[0], hi[1], hi[2], hi[3]),
                "pair_key": f"{lo[0]}|{hi[0]}",
            },
        )
    return out


def dedupe_pair_records(records: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
    """Keep first occurrence of each ``pair_key`` (merge tiers may overlap graph edges)."""
    seen: set[str] = set()
    out: list[dict[str, Any]] = []
    for rec in records:
        key = rec.get("pair_key")
        if not key or key in seen:
            continue
        seen.add(key)
        out.append(rec)
    return out


def collect_all_pair_records(
    rows: Sequence[tuple[str, str, str, float]],
    conn: sqlite3.Connection | None,
    *,
    strategies: Sequence[str],
    max_pairs_per_merge_bucket: int = 40,
    max_pairs_per_fingerprint: int = 30,
    similarity_min_score: float = 0.82,
    similarity_max_pairs: int = 50_000,
    rng: random.Random | None = None,
) -> list[dict[str, Any]]:
    """Union requested strategies in order (merge → fingerprint → similarity)."""
    want = {str(s).strip() for s in strategies}
    merged: list[dict[str, Any]] = []
    if "merge_signature" in want:
        merged.extend(
            collect_merge_signature_pairs(
                rows,
                max_pairs_per_bucket=max_pairs_per_merge_bucket,
                rng=rng,
            ),
        )
    if "fingerprint_collision" in want:
        merged.extend(
            collect_fingerprint_collision_pairs(
                rows,
                max_pairs_per_fingerprint=max_pairs_per_fingerprint,
                rng=rng,
            ),
        )
    if "similarity_edge" in want:
        if conn is None:
            raise ValueError("similarity_edge strategy requires an open sqlite connection")
        merged.extend(
            collect_similarity_edge_pairs(
                conn,
                min_score=similarity_min_score,
                max_pairs=similarity_max_pairs,
            ),
        )
    return merged


def sample_pair_records(
    records: Sequence[dict[str, Any]],
    n: int,
    *,
    seed: int,
) -> list[dict[str, Any]]:
    if n <= 0 or len(records) <= n:
        return list(records)
    rng = random.Random(seed)  # noqa: S311 — benchmark export reproducibility
    items = list(records)
    rng.shuffle(items)
    return items[:n]