backend/catalog.py

"""Backend catalog service."""
from __future__ import annotations

import json
import re
import sqlite3
import threading
import time
from collections import defaultdict
from pathlib import Path
from typing import Any

from sqlmodel import Session, select

from backend.image_policy import clip_min_similarity_catalog, strict_clip_catalog_enabled
from backend.scenarios import (
    play_excluded_from_surfacing,
    scenario_title_fields,
    strip_profile_bucket_text_prefix,
    title_surface_as_bundle,
    title_surface_as_profile_bucket_noise,
)
from backend.scrape_artifacts import clean_scraped_catalog_text, clean_scraped_notes
from models import (
    Alias,
    Asset,
    Definition,
    FetlifeKinkMeta,
    FetlifePictureRef,
    Kink,
    KinkExample,
    KinkLemma,
    KinkScenarioParent,
    SimilarityEdge,
)


def _lemma_name_hash(name: str) -> str:
    """Fingerprint a kink name so the lemma cache invalidates when the name changes."""
    import hashlib
    return hashlib.sha1(name.encode("utf-8")).hexdigest()


def _load_cached_lemma_signatures(
    self,
    candidates: dict[str, dict[str, Any]],
) -> tuple[dict[str, str], list[tuple[str, str, str]]]:
    """Return (cached_lemmas_by_kid, empty_writebacks_list).

    A cached row is only returned when its ``name_hash`` matches the current candidate name —
    stale rows are silently ignored (caller will recompute and overwrite).
    """
    try:
        with self._sqlite() as conn:
            rows = conn.execute("SELECT kink_id, name_hash, signature FROM kinklemma").fetchall()
    except sqlite3.OperationalError:
        return {}, []
    cached: dict[str, str] = {}
    for row in rows:
        kid = row["kink_id"]
        if kid not in candidates:
            continue
        if row["name_hash"] != _lemma_name_hash(str(candidates[kid].get("name", "") or "")):
            continue
        cached[kid] = row["signature"] or ""
    return cached, []


def _persist_cached_lemma_signatures(
    self,
    rows: list[tuple[str, str, str]],
) -> None:
    if not rows:
        return
    try:
        with self._sqlite() as conn:
            conn.executemany(
                "INSERT INTO kinklemma (kink_id, name_hash, signature) VALUES (?, ?, ?) "
                "ON CONFLICT(kink_id) DO UPDATE SET name_hash=excluded.name_hash, signature=excluded.signature",
                rows,
            )
            conn.commit()
    except sqlite3.OperationalError:
        return

_IMAGE_TOKEN_RE = re.compile(r"[a-z0-9]+")
_IMAGE_RELEVANCE_STOPWORDS = {"and", "the", "play", "sex", "in", "with"}
_PRODUCT_FLAG_KEYS = (
    "starter_eligible",
    "starter_tier",
    "starter_reason",
    "shared_eligible",
    "prompt_eligible",
    "canonical_priority",
    "image_trust_state",
    "detail_summary",
)


def _image_relevance_from_tokens(kink_tokens: set[str], text_tokens: set[str], reuse_count: int) -> dict[str, Any]:
    score = 1.0
    reasons: list[str] = []
    if reuse_count > 1:
        penalty = min(0.08 * (reuse_count - 1), 0.55)
        score -= penalty
        reasons.append(f"reused:{reuse_count}")
    if text_tokens and kink_tokens and not (kink_tokens & text_tokens):
        score -= 0.18
        reasons.append("caption_mismatch")
    if len(text_tokens) >= 18:
        score -= 0.18
        reasons.append("tag_dense")
    if not text_tokens:
        score -= 0.08
        reasons.append("no_caption")
    score = max(0.0, round(score, 3))
    if reuse_count > 12:
        return {"score": score, "reason": "attachment_reused_too_often", "trusted": False}
    if score < 0.42:
        return {"score": score, "reason": ",".join(reasons) or "low_relevance", "trusted": False}
    return {"score": score, "reason": ",".join(reasons), "trusted": True}


def _starter_sort_key(item: dict[str, Any]) -> tuple[int, float, float, float, str]:
    return (
        int(item.get("starter_tier_rank", 0) or 0),
        float(item.get("source_backed_popularity", 0.0) or 0.0),
        float(item.get("starter_score", 0.0) or 0.0),
        float(item.get("popularity", 0.0) or 0.0),
        str(item.get("name", "")).lower(),
    )


def _catalog_invalidation_token(self) -> tuple[Any, ...]:
    """Bust catalog cache when CLIP file or policy env changes (no restart required)."""
    p = self.path.parent / "clip_asset_scores.json"
    try:
        st = p.stat()
        mtime_ns = st.st_mtime_ns
    except OSError:
        mtime_ns = -1
    return (
        mtime_ns,
        clip_min_similarity_catalog(),
        strict_clip_catalog_enabled(),
    )


def _load_clip_asset_scores_file(path: Path) -> dict[str, dict[str, Any]]:
    """Load optional offline CLIP verdicts (see scripts/clip_align_kink_assets.py).

    Each entry may include ``accepted: false`` to drop an image at catalog build time.
    With ``KINK_STRICT_CLIP_IMAGES=1`` and a non-empty file, URLs missing from this map are
    dropped too; see ``backend/image_policy.py``.
    """
    if not path.is_file():
        return {}
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    if not isinstance(raw, dict):
        return {}
    out: dict[str, dict[str, Any]] = {}
    for url, payload in raw.items():
        if not isinstance(url, str) or not isinstance(payload, dict):
            continue
        out[url] = payload
    return out


def _invalidate_catalog_cache(self, *, force: bool = False) -> None:
    now = time.monotonic()
    if not force and now - self._last_cache_invalidation_at < 30.0:
        return
    self._last_cache_invalidation_at = now
    self._catalog_cache = None
    self._catalog_token = None

def _build_catalog_cache(self) -> dict[str, Any]:
    type_rows = self._content_type_map()
    if not type_rows:
        type_rows = self._sync_content_types()
    with Session(self.engine) as session:
        kinks = session.exec(select(Kink).order_by(Kink.name)).all()
        aliases = session.exec(select(Alias)).all()
        definitions = session.exec(select(Definition)).all()
        examples = session.exec(select(KinkExample)).all()
        assets = session.exec(select(Asset)).all()
        picture_refs = session.exec(select(FetlifePictureRef)).all()
        fetlife_meta_rows = session.exec(select(FetlifeKinkMeta)).all()
        scenario_parent_rows = session.exec(select(KinkScenarioParent)).all()
        scenario_bridge_rows = session.exec(
            select(SimilarityEdge).where(SimilarityEdge.similarity_type == "scenario_bridge")
        ).all()

    alias_map: dict[str, list[str]] = defaultdict(list)
    definition_map: dict[str, list[dict[str, str]]] = defaultdict(list)
    example_map: dict[str, list[dict[str, str]]] = defaultdict(list)
    asset_map: dict[str, list[dict[str, Any]]] = defaultdict(list)
    raw_asset_count_by_kink: dict[str, int] = defaultdict(int)
    inbound_counts: dict[str, int] = defaultdict(int)
    outbound_counts: dict[str, int] = defaultdict(int)
    fetlife_meta_map = {row.kink_id: row for row in fetlife_meta_rows}
    parents_by_scenario: dict[str, list[dict[str, Any]]] = defaultdict(list)
    children_by_parent: dict[str, list[dict[str, Any]]] = defaultdict(list)
    seen_scenario_parent_pairs: set[tuple[str, str]] = set()
    for row in scenario_parent_rows:
        pair = (row.scenario_kink_id, row.parent_kink_id)
        seen_scenario_parent_pairs.add(pair)
        payload = {
            "scenario_kink_id": row.scenario_kink_id,
            "parent_kink_id": row.parent_kink_id,
            "score": float(row.score),
            "method": row.method,
            "updated_at": row.updated_at,
        }
        parents_by_scenario[row.scenario_kink_id].append(payload)
        children_by_parent[row.parent_kink_id].append(payload)
    for row in scenario_bridge_rows:
        pair = (row.left_kink_id, row.right_kink_id)
        if pair in seen_scenario_parent_pairs:
            continue
        payload = {
            "scenario_kink_id": row.left_kink_id,
            "parent_kink_id": row.right_kink_id,
            "score": float(row.score),
            "method": row.method,
            "updated_at": "",
        }
        parents_by_scenario[row.left_kink_id].append(payload)
        children_by_parent[row.right_kink_id].append(payload)
    for rows in parents_by_scenario.values():
        rows.sort(key=lambda item: (-float(item["score"]), str(item["parent_kink_id"])))
    for rows in children_by_parent.values():
        rows.sort(key=lambda item: (-float(item["score"]), str(item["scenario_kink_id"])))
    with self._sqlite() as conn:
        for row in conn.execute(
            "SELECT right_kink_id, COUNT(*) AS n FROM similarityedge GROUP BY right_kink_id"
        ).fetchall():
            inbound_counts[row["right_kink_id"]] = int(row["n"])
        for row in conn.execute(
            "SELECT left_kink_id, COUNT(*) AS n FROM similarityedge GROUP BY left_kink_id"
        ).fetchall():
            outbound_counts[row["left_kink_id"]] = int(row["n"])

    for row in aliases:
        alias_map[row.kink_id].append(row.alias)
    for row in definitions:
        definition_map[row.kink_id].append(
            {
                "text": row.text,
                "source_id": row.source_id,
                "source_url": row.source_url,
                "license": row.license,
            }
        )
    for row in examples:
        example_map[row.kink_id].append({"text": row.text, "kind": row.kind})
    fallback_definition_by_name: dict[str, str] = {}
    for kink in kinks:
        normalized_name = self._normalized_play_name(kink.name)
        if not normalized_name:
            continue
        candidate_definition = strip_profile_bucket_text_prefix(kink.short_definition.strip())
        candidate_definition = clean_scraped_catalog_text(kink.name, candidate_definition)
        if not candidate_definition:
            defs = definition_map.get(kink.id, [])
            if defs:
                candidate_definition = strip_profile_bucket_text_prefix(defs[0]["text"].strip())
                candidate_definition = clean_scraped_catalog_text(kink.name, candidate_definition)
        if not candidate_definition:
            continue
        incumbent = fallback_definition_by_name.get(normalized_name, "")
        if not incumbent or len(candidate_definition) < len(incumbent):
            fallback_definition_by_name[normalized_name] = candidate_definition
    picture_ref_by_key: dict[tuple[str, str], FetlifePictureRef] = {}
    reuse_counts: dict[str, int] = {}
    for row in picture_refs:
        picture_ref_by_key[(row.kink_id, row.attachment_id)] = row
        reuse_counts.setdefault(row.attachment_id, 0)
    for row in picture_refs:
        reuse_counts[row.attachment_id] += 1
    kink_image_tokens_by_id = {
        kink.id: {
            token
            for token in _IMAGE_TOKEN_RE.findall(kink.name.lower())
            if token not in _IMAGE_RELEVANCE_STOPWORDS
        }
        for kink in kinks
    }
    picture_text_tokens_by_key = {
        key: set(
            _IMAGE_TOKEN_RE.findall(
                " ".join(
                    part
                    for part in (row.picture_title, row.caption_text)
                    if part
                ).lower()
            )
        )
        for key, row in picture_ref_by_key.items()
    }
    clip_by_url = _load_clip_asset_scores_file(self.path.parent / "clip_asset_scores.json")
    clip_catalog_min = clip_min_similarity_catalog()
    # Prototype: KINK_STRICT_CLIP_IMAGES=1 + non-empty clip file → require CLIP row and accepted==true; tighten FL ref path.
    enforce_clip_row = strict_clip_catalog_enabled() and bool(clip_by_url)
    filtered_image_reasons_by_kink: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    image_quality_by_kink: dict[str, dict[str, float]] = defaultdict(lambda: {"score_total": 0.0, "count": 0.0})
    for row in assets:
        raw_asset_count_by_kink[row.kink_id] += 1
        attachment_id = ""
        if row.source_id == "fetlife_fetish_pages":
            attachment_id = Path(row.asset_url).stem
        picture_ref = picture_ref_by_key.get((row.kink_id, attachment_id)) if attachment_id else None
        reuse_count = reuse_counts.get(attachment_id, 1) if attachment_id else 1
        if picture_ref:
            relevance = _image_relevance_from_tokens(
                kink_image_tokens_by_id[row.kink_id],
                picture_text_tokens_by_key[(row.kink_id, attachment_id)],
                reuse_count,
            )
        elif row.source_id == "fetlife_fetish_pages" and strict_clip_catalog_enabled():
            relevance = {
                "score": 0.28,
                "reason": "fetlife_picture_ref_missing_strict",
                "trusted": False,
            }
        else:
            relevance = {"score": 0.72, "reason": "", "trusted": True}
        if not relevance["trusted"]:
            filtered_image_reasons_by_kink[row.kink_id][str(relevance["reason"])] += 1
            continue
        clip_meta = clip_by_url.get(row.asset_url)
        clip_sim = None
        if enforce_clip_row:
            if clip_meta is None:
                filtered_image_reasons_by_kink[row.kink_id]["clip_missing_row"] += 1
                continue
            if clip_meta.get("accepted") is not True:
                filtered_image_reasons_by_kink[row.kink_id]["clip_not_accepted_strict"] += 1
                continue
        if clip_meta:
            kid = str(clip_meta.get("kink_id", "") or "")
            if kid and kid != row.kink_id:
                filtered_image_reasons_by_kink[row.kink_id]["clip_kink_mismatch"] += 1
                continue
            try:
                clip_sim = float(clip_meta.get("similarity"))
            except (TypeError, ValueError):
                clip_sim = None
            if clip_sim is not None and clip_sim < clip_catalog_min:
                filtered_image_reasons_by_kink[row.kink_id]["clip_below_catalog_min_similarity"] += 1
                continue
            if not enforce_clip_row and clip_meta.get("accepted") is False:
                filtered_image_reasons_by_kink[row.kink_id]["clip_rejected_offline"] += 1
                continue
        image_quality_by_kink[row.kink_id]["score_total"] += float(relevance["score"])
        image_quality_by_kink[row.kink_id]["count"] += 1.0
        entry = {
            "asset_url": row.asset_url,
            "license": row.license,
            "creator": picture_ref.creator if picture_ref and picture_ref.creator else row.creator,
            "caption_text": picture_ref.caption_text if picture_ref else "",
            "picture_page_url": picture_ref.picture_page_url if picture_ref else "",
            "tag_spam_score": max(0, reuse_count - 1),
            "attachment_reuse_count": reuse_count,
            "image_relevance_score": relevance["score"],
            "image_filter_reason": relevance["reason"],
            "click_to_reveal": row.click_to_reveal,
            "is_illustration": row.is_illustration,
            "is_explicit": row.is_explicit,
        }
        if clip_sim is not None:
            entry["clip_similarity"] = clip_sim
        if clip_meta is not None and "accepted" in clip_meta:
            entry["clip_accepted"] = clip_meta["accepted"]
        asset_map[row.kink_id].append(entry)
    for items in asset_map.values():
        items.sort(
            key=lambda item: (
                item.get("tag_spam_score", 0),
                0 if item.get("caption_text") else 1,
                -float(item["clip_similarity"]) if item.get("clip_similarity") is not None else 0.0,
                item.get("asset_url", ""),
            )
        )
    detail_by_id: dict[str, dict[str, Any]] = {}
    summary_by_id: dict[str, dict[str, Any]] = {}
    search_docs: list[dict[str, Any]] = []
    for kink in kinks:
        defs = definition_map.get(kink.id, [])
        raw_notes = kink.notes.strip()
        display_notes = clean_scraped_notes(kink.id, kink.name, raw_notes)
        definition = kink.short_definition.strip() or (defs[0]["text"].strip() if defs else "")
        definition = strip_profile_bucket_text_prefix(definition)
        definition = clean_scraped_catalog_text(kink.name, definition)
        if not definition and kink.cluster == "fetlife_fetish":
            definition = strip_profile_bucket_text_prefix(
                fallback_definition_by_name.get(self._normalized_play_name(kink.name), "")
            )
            definition = clean_scraped_catalog_text(kink.name, definition)
        if not definition:
            from backend.curated_definitions import CURATED_DEFINITIONS
            curated = CURATED_DEFINITIONS.get(kink.id, "").strip()
            if curated:
                definition = curated
        examples_for_kink = example_map.get(kink.id, [])
        example_texts = [
            text
            for row in examples_for_kink
            if row["kind"] == "example"
            for text in [clean_scraped_catalog_text(kink.name, row["text"])]
            if text
        ]
        visual_cues = [
            text
            for row in examples_for_kink
            if row["kind"] == "visual_cue"
            for text in [clean_scraped_catalog_text(kink.name, row["text"])]
            if text
        ]
        summary = self._display_summary(definition, display_notes, example_texts, kink.name)
        explicit_popularity = self._extract_popularity_from_notes(raw_notes)
        source_backed_popularity = fetlife_meta_map.get(kink.id).popularity if kink.id in fetlife_meta_map else 0.0
        popularity = source_backed_popularity or explicit_popularity or (
            inbound_counts.get(kink.id, 0) * 8
            + outbound_counts.get(kink.id, 0) * 4
            + len(example_texts)
            + len(asset_map.get(kink.id, [])) * 3
            + len(defs) * 2
        )
        base_payload = {
            "id": kink.id,
            "name": kink.name,
            "cluster": kink.cluster,
            "definition": definition,
            "summary": summary,
            "notes": display_notes,
        }
        type_row = type_rows.get(kink.id)
        content_kind = type_row["content_kind"] if type_row else self._content_kind(base_payload)
        type_evidence = type_row["evidence"] if type_row else ""
        scenario_links = parents_by_scenario.get(kink.id, [])
        scenario_parent_ids = [row["parent_kink_id"] for row in scenario_links]
        is_scenario = bool(scenario_links)
        scenario_child_count = len(children_by_parent.get(kink.id, []))
        if content_kind == "play":
            scenario_title_score, title_surface_flag = scenario_title_fields(kink.name)
        else:
            scenario_title_score, title_surface_flag = 0.0, False
        title_bundle_flag = title_surface_as_bundle(kink.name)
        profile_bucket_noise_flag = title_surface_as_profile_bucket_noise(kink.name)
        from backend.direction_shapes import lookup_direction_shape
        shape_info = lookup_direction_shape(kink.id, kink.name)
        direction_shape = shape_info.get("shape", "action")
        direction_role_a = shape_info.get("role_a", "")
        direction_role_b = shape_info.get("role_b", "")
        detail = {
            "id": kink.id,
            "name": kink.name,
            "cluster": kink.cluster,
            "content_kind": content_kind,
            "type_evidence": type_evidence,
            "definition": definition,
            "summary": summary,
            "notes": display_notes,
            "risk_level": kink.risk_level,
            "is_extreme": kink.is_extreme,
            "popularity": popularity,
            "source_backed_popularity": source_backed_popularity,
            "fetlife_fetish_id": fetlife_meta_map.get(kink.id).fetish_id if kink.id in fetlife_meta_map else "",
            "has_images": bool(asset_map.get(kink.id)),
            "has_real_images": bool(fetlife_meta_map.get(kink.id).has_real_images) if kink.id in fetlife_meta_map else False,
            "raw_asset_count": int(raw_asset_count_by_kink.get(kink.id, 0)),
            "filtered_asset_count": len(asset_map.get(kink.id, [])),
            "representative_image_count": len(asset_map.get(kink.id, [])),
            "image_relevance_score": (
                round(image_quality_by_kink[kink.id]["score_total"] / image_quality_by_kink[kink.id]["count"], 3)
                if image_quality_by_kink[kink.id]["count"]
                else 0.0
            ),
            "image_filter_reasons": dict(filtered_image_reasons_by_kink.get(kink.id, {})),
            "aliases": alias_map.get(kink.id, []),
            "definitions": defs,
            "examples": example_texts,
            "visual_cues": visual_cues,
            "assets": asset_map.get(kink.id, []),
            "similar": [],
            "similar_count": int(fetlife_meta_map.get(kink.id).similar_count) if kink.id in fetlife_meta_map else 0,
            "is_scenario": is_scenario,
            "scenario_parent_ids": scenario_parent_ids,
            "scenario_parent_links": scenario_links,
            "scenario_child_count": scenario_child_count,
            "scenario_title_score": scenario_title_score,
            "title_surface_as_scenario": title_surface_flag,
            "title_surface_as_bundle": title_bundle_flag,
            "title_surface_as_profile_bucket_noise": profile_bucket_noise_flag,
            "direction_shape": direction_shape,
            "direction_role_a": direction_role_a,
            "direction_role_b": direction_role_b,
        }
        detail.update(self._derived_product_flags(detail))
        if is_scenario or title_surface_flag:
            detail.update(
                {
                    "starter_eligible": False,
                    "starter_tier": "scenario",
                    "shared_eligible": False,
                    "prompt_eligible": False,
                }
            )
        detail["starter_tier_rank"] = self._starter_tier_rank(detail)
        detail["starter_score"] = self._starter_score(detail)
        summary_payload = {
            "id": kink.id,
            "name": kink.name,
            "cluster": kink.cluster,
            "content_kind": content_kind,
            "type_evidence": type_evidence,
            "summary": summary,
            "definition": definition,
            "notes": display_notes,
            "has_images": bool(asset_map.get(kink.id)),
            "has_real_images": bool(fetlife_meta_map.get(kink.id).has_real_images) if kink.id in fetlife_meta_map else False,
            "raw_asset_count": int(raw_asset_count_by_kink.get(kink.id, 0)),
            "filtered_asset_count": len(asset_map.get(kink.id, [])),
            "representative_image_count": len(asset_map.get(kink.id, [])),
            "image_relevance_score": (
                round(image_quality_by_kink[kink.id]["score_total"] / image_quality_by_kink[kink.id]["count"], 3)
                if image_quality_by_kink[kink.id]["count"]
                else 0.0
            ),
            "popularity": popularity,
            "source_backed_popularity": source_backed_popularity,
            "similar_count": int(fetlife_meta_map.get(kink.id).similar_count) if kink.id in fetlife_meta_map else 0,
            "is_scenario": is_scenario,
            "scenario_parent_ids": scenario_parent_ids,
            "scenario_child_count": scenario_child_count,
            "scenario_title_score": scenario_title_score,
            "title_surface_as_scenario": title_surface_flag,
            "title_surface_as_bundle": title_bundle_flag,
            "title_surface_as_profile_bucket_noise": profile_bucket_noise_flag,
            "direction_shape": direction_shape,
            "direction_role_a": direction_role_a,
            "direction_role_b": direction_role_b,
        }
        for key in _PRODUCT_FLAG_KEYS:
            summary_payload[key] = detail[key]
        summary_payload["starter_tier_rank"] = detail["starter_tier_rank"]
        summary_payload["starter_score"] = detail["starter_score"]
        surface_excluded = play_excluded_from_surfacing(summary_payload)
        detail["surface_excluded"] = surface_excluded
        summary_payload["surface_excluded"] = surface_excluded
        detail_by_id[kink.id] = detail
        summary_by_id[kink.id] = summary_payload
        search_docs.append(
            {
                "id": kink.id,
                "name": kink.name,
                "cluster": kink.cluster,
                "content_kind": content_kind,
                "type_evidence": type_evidence,
                "aliases": alias_map.get(kink.id, []),
                "summary": summary,
                "definition": definition,
                "examples": example_texts[:8],
                "visual_cues": visual_cues[:8],
                "popularity": popularity,
                "source_backed_popularity": source_backed_popularity,
                "is_scenario": is_scenario,
                "scenario_parent_ids": scenario_parent_ids,
                "scenario_child_count": scenario_child_count,
                "scenario_title_score": scenario_title_score,
                "title_surface_as_scenario": title_surface_flag,
                "title_surface_as_bundle": title_bundle_flag,
                "title_surface_as_profile_bucket_noise": profile_bucket_noise_flag,
                "starter_eligible": summary_payload["starter_eligible"],
                "starter_tier": summary_payload["starter_tier"],
                "prompt_eligible": summary_payload["prompt_eligible"],
                "shared_eligible": summary_payload["shared_eligible"],
                "image_trust_state": summary_payload["image_trust_state"],
                "detail_summary": summary_payload["detail_summary"],
                "haystack": " ".join(
                    [
                        kink.name,
                        *alias_map.get(kink.id, []),
                        definition,
                        summary,
                        display_notes,
                        *example_texts[:8],
                        *visual_cues[:8],
                    ]
                ).lower(),
            }
        )
    self._mark_runtime_duplicates(detail_by_id, summary_by_id, search_docs)
    self._maybe_rebuild_fts_index(search_docs)
    play_exact_sets: defaultdict[str, set[str]] = defaultdict(set)
    for doc in search_docs:
        if doc["content_kind"] != "play":
            continue
        play_exact_sets[doc["name"].lower()].add(doc["id"])
        for alias in doc["aliases"]:
            play_exact_sets[str(alias).lower()].add(doc["id"])
    play_exact_ids_by_lower = {k: sorted(v) for k, v in play_exact_sets.items()}
    play_summaries = [
        item
        for item in summary_by_id.values()
        if item["content_kind"] == "play" and not item["surface_excluded"]
    ]
    play_summaries.sort(key=lambda item: (-item["popularity"], item["name"].lower()))
    play_details = [detail_by_id[item["id"]] for item in play_summaries]
    starter_summaries = [item for item in play_summaries if item["starter_eligible"]]
    starter_summaries.sort(key=_starter_sort_key, reverse=True)
    return {
        "detail_by_id": detail_by_id,
        "summary_by_id": summary_by_id,
        "search_docs": search_docs,
        "play_exact_ids_by_lower": play_exact_ids_by_lower,
        "play_summaries": play_summaries,
        "play_details": play_details,
        "starter_summaries": starter_summaries,
    }

def _mark_runtime_duplicates(
    self,
    detail_by_id: dict[str, dict[str, Any]],
    summary_by_id: dict[str, dict[str, Any]],
    search_docs: list[dict[str, Any]],
) -> None:
    """Group play kinks by (cluster, merge_signature) and (cluster, compact_form); within each
    group with >= 2 members, keep the canonical (highest popularity, has-definition, shorter name)
    and mark every other member ``surface_excluded=True``.

    This dedupe is purely runtime — the SQLite catalog is not modified — so it lands on the next
    deploy without needing a Hub dataset rebuild.
    """
    from backend.kink_merge import compact_form, lemma_signature, merge_signature

    candidates: dict[str, dict[str, Any]] = {}
    for kid, summary in summary_by_id.items():
        if summary.get("content_kind") != "play":
            continue
        if summary.get("surface_excluded"):
            continue
        candidates[kid] = summary

    sig_groups: defaultdict[tuple[str, str], list[str]] = defaultdict(list)
    compact_groups: defaultdict[tuple[str, str], list[str]] = defaultdict(list)
    # Lemma-based grouping (spaCy ``en_core_web_sm`` lemmatization + POS-aware stopword filter).
    # Catches morphological siblings ("Cuddling" / "Cuddles" / "Cuddly", "After sex cuddles" /
    # "Cuddling after sex") that ``merge_signature``'s sorted-token-set misses. F1 0.80 on the
    # hand-labeled set vs F1 0.29 for ``merge_signature`` alone — see scripts/merge_dedup_eval.py.
    lemma_groups: defaultdict[tuple[str, str], list[str]] = defaultdict(list)
    cached_lemmas, lemma_writebacks = _load_cached_lemma_signatures(self, candidates)
    for kid, summary in candidates.items():
        cluster = str(summary.get("cluster", "") or "")
        name = str(summary.get("name", "") or "")
        sig = merge_signature(name)
        if sig:
            sig_groups[(cluster, sig)].append(kid)
        cf = compact_form(name)
        if len(cf) >= 8:
            compact_groups[(cluster, cf)].append(kid)
        if kid in cached_lemmas:
            lemma_sig = cached_lemmas[kid]
        else:
            lemma_sig = lemma_signature(name)
            lemma_writebacks.append((kid, _lemma_name_hash(name), lemma_sig))
        if lemma_sig:
            lemma_groups[(cluster, lemma_sig)].append(kid)
    _persist_cached_lemma_signatures(self, lemma_writebacks)

    def _canonical_key(kid: str) -> tuple[int, int, float, int, str]:
        """Sort key — higher is better. Strict preference: has_definition first, then well-cased
        Title-style name (so ``Anal Creampie`` wins over ``analcreampie``), then popularity, then
        shorter name, then id stability."""
        summary = candidates[kid]
        detail = detail_by_id.get(kid, {})
        name = str(summary.get("name", "") or "")
        has_def = 1 if (detail.get("definition") or "").strip() else 0
        # Heuristic: a name with at least one uppercase letter is better-cased than all-lowercase.
        well_cased = 1 if any(ch.isupper() for ch in name) else 0
        pop = float(summary.get("popularity", 0.0) or 0.0)
        # Ties on the rest: prefer SHORTER name (negate length so larger key = shorter name).
        return (has_def, well_cased, pop, -len(name), kid)

    duplicate_ids: set[str] = set()
    for groups in (sig_groups, compact_groups, lemma_groups):
        for ids in groups.values():
            if len(ids) < 2:
                continue
            canonical = max(ids, key=_canonical_key)
            for kid in ids:
                if kid != canonical:
                    duplicate_ids.add(kid)

    if not duplicate_ids:
        return
    for kid in duplicate_ids:
        if kid in detail_by_id:
            detail_by_id[kid]["surface_excluded"] = True
            detail_by_id[kid]["surface_excluded_reason"] = "runtime_duplicate"
        if kid in summary_by_id:
            summary_by_id[kid]["surface_excluded"] = True
            summary_by_id[kid]["surface_excluded_reason"] = "runtime_duplicate"
    duplicate_set = duplicate_ids
    for doc in search_docs:
        if doc.get("id") in duplicate_set:
            doc["surface_excluded"] = True


def _rebuild_fts_index(self, search_docs: list[dict[str, Any]]) -> None:
    with self._sqlite() as conn:
        conn.execute("DELETE FROM play_fts")
        conn.executemany(
            "INSERT INTO play_fts (kink_id, content_kind, name, aliases, definition, examples) VALUES (?, ?, ?, ?, ?, ?)",
            [
                (
                    doc["id"],
                    doc["content_kind"],
                    doc["name"],
                    " ".join(doc["aliases"]),
                    doc["definition"],
                    " ".join(doc["examples"]),
                )
                for doc in search_docs
            ],
        )
        conn.commit()

def _maybe_rebuild_fts_index(self, search_docs: list[dict[str, Any]]) -> None:
    try:
        with self._sqlite() as conn:
            row = conn.execute("SELECT COUNT(*) AS n FROM play_fts").fetchone()
            existing_count = int(row["n"]) if row else 0
        if existing_count == 0 or existing_count < max(1000, len(search_docs) // 2):
            self._rebuild_fts_index(search_docs)
    except sqlite3.OperationalError:
        return

_CATALOG_PICKLE_VERSION = 1


def _catalog_pickle_path(self) -> Path:
    return self.path.parent / f".{self.path.name}.catalog_v{_CATALOG_PICKLE_VERSION}.pkl"


def _catalog_pickle_key(self, token: tuple[Any, ...]) -> tuple[Any, ...]:
    """A content-derived key that survives side-table writes (e.g. kinklemma) but invalidates
    when the underlying catalog changes. Uses SQL counters instead of file mtime because
    _mark_runtime_duplicates writes back to kinklemma during the same build, bumping mtime.

    Deliberately excludes ``token`` (CLIP env policy): the cached catalog dict's structure
    does not depend on those env vars, and including them would cause spurious misses when
    the build-host env differs from the runtime env (e.g. baking a pickle locally for HF).
    """
    import sys as _sys
    try:
        with self._sqlite() as conn:
            kink_count = int(conn.execute("SELECT COUNT(*) FROM kink").fetchone()[0])
            edge_count = int(conn.execute("SELECT COUNT(*) FROM similarityedge").fetchone()[0])
            max_kid = conn.execute("SELECT MAX(id) FROM kink").fetchone()[0] or ""
    except sqlite3.OperationalError:
        return ()
    return (
        _CATALOG_PICKLE_VERSION,
        _sys.version_info[:2],
        kink_count,
        edge_count,
        max_kid,
    )


def _try_load_pickled_catalog(self, token: tuple[Any, ...]) -> dict[str, Any] | None:
    path = _catalog_pickle_path(self)
    if not path.is_file():
        print(f"[kink_cli pickle] miss path_missing path={path}", flush=True)
        return None
    import pickle
    try:
        with path.open("rb") as f:
            header = pickle.load(f)
            expected = _catalog_pickle_key(self, token)
            if header != expected:
                print(f"[kink_cli pickle] miss key_mismatch path={path} disk={header!r} expected={expected!r}", flush=True)
                return None
            cache = pickle.load(f)
            print(f"[kink_cli pickle] hit path={path} size={path.stat().st_size}", flush=True)
            return cache
    except (OSError, pickle.UnpicklingError, EOFError, ValueError, AttributeError, ImportError) as exc:
        print(f"[kink_cli pickle] miss exception path={path} exc={exc!r}", flush=True)
        return None


def _save_pickled_catalog(self, cache: dict[str, Any], token: tuple[Any, ...]) -> None:
    path = _catalog_pickle_path(self)
    tmp = path.with_suffix(path.suffix + ".tmp")
    import pickle
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        with tmp.open("wb") as f:
            pickle.dump(_catalog_pickle_key(self, token), f, protocol=pickle.HIGHEST_PROTOCOL)
            pickle.dump(cache, f, protocol=pickle.HIGHEST_PROTOCOL)
        tmp.replace(path)
        print(f"[kink_cli pickle] saved path={path} size={path.stat().st_size}", flush=True)
    except OSError as exc:
        print(f"[kink_cli pickle] save_failed path={path} exc={exc!r}", flush=True)
        try:
            tmp.unlink()
        except OSError:
            pass


def _catalog(self) -> dict[str, Any]:
    token = _catalog_invalidation_token(self)
    if self._catalog_cache is not None and getattr(self, "_catalog_token", None) == token:
        return self._catalog_cache
    if hasattr(self, "_catalog_ready") and not self._catalog_ready.is_set():
        self._catalog_ready.wait()
        error = getattr(self, "_catalog_warm_error", None)
        if error is not None:
            raise RuntimeError("Catalog warm-up failed") from error
        token = _catalog_invalidation_token(self)
        if self._catalog_cache is not None and getattr(self, "_catalog_token", None) == token:
            return self._catalog_cache
    lock = getattr(self, "_catalog_build_lock", None)
    if lock is None:
        lock = threading.Lock()
        self._catalog_build_lock = lock
    with lock:
        token = _catalog_invalidation_token(self)
        if self._catalog_cache is not None and getattr(self, "_catalog_token", None) == token:
            return self._catalog_cache
        cached = _try_load_pickled_catalog(self, token)
        if cached is not None:
            self._catalog_cache = cached
            self._catalog_token = token
            return self._catalog_cache
        self._catalog_cache = self._build_catalog_cache()
        self._catalog_token = _catalog_invalidation_token(self)
        _save_pickled_catalog(self, self._catalog_cache, self._catalog_token)
        return self._catalog_cache

def catalog_is_ready(self) -> bool:
    return self._catalog_cache is not None

def warm_up_catalog(self) -> None:
    """Build catalog cache in background thread on startup."""
    if self._catalog_cache is not None:
        return
    thread = getattr(self, "_catalog_warm_thread", None)
    if thread is not None and thread.is_alive():
        return
    self._catalog_ready = threading.Event()
    self._catalog_warm_error = None

    def _build():
        try:
            lock = getattr(self, "_catalog_build_lock", None)
            if lock is None:
                lock = threading.Lock()
                self._catalog_build_lock = lock
            with lock:
                token = _catalog_invalidation_token(self)
                if self._catalog_cache is None or getattr(self, "_catalog_token", None) != token:
                    self._catalog_cache = self._build_catalog_cache()
                    self._catalog_token = _catalog_invalidation_token(self)
        except BaseException as exc:
            self._catalog_warm_error = exc
        finally:
            self._catalog_ready.set()

    thread = threading.Thread(target=_build, daemon=True)
    self._catalog_warm_thread = thread
    thread.start()

def refresh_cache(self) -> None:
    self._invalidate_catalog_cache(force=True)
    self._catalog()