scripts/inspect_scenario_title_surface.py

#!/usr/bin/env python3
"""Offline inspection: histogram of scenario_title_score for catalog plays.

Helps tune ``SCENARIO_TITLE_SURFACE_THRESHOLD`` in ``backend/scenarios.py``.

Usage:
  KINK_SKIP_HEAVY_WARM=1 python scripts/inspect_scenario_title_surface.py --db data/store_slim.db
  KINK_SKIP_HEAVY_WARM=1 python scripts/inspect_scenario_title_surface.py --db data/store_slim.db --near 0.38 --sample 25
"""
from __future__ import annotations

import argparse
import os
from collections import Counter
from pathlib import Path


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--db", type=Path, required=True, help="path to SQLite store (e.g. data/store_slim.db)")
    ap.add_argument(
        "--bin",
        type=float,
        default=0.05,
        help="histogram bucket width in score units (default 0.05)",
    )
    ap.add_argument(
        "--near",
        type=float,
        default=0.0,
        help="if > 0, print up to --sample rows with score in [near-bin, near+bin]",
    )
    ap.add_argument("--sample", type=int, default=20, help="max rows to print for --near")
    args = ap.parse_args()
    db = args.db.resolve()
    if not db.is_file():
        raise SystemExit(f"database not found: {db}")

    os.environ.setdefault("KINK_SKIP_HEAVY_WARM", "1")
    from backend.core import Backend
    from backend.scenarios import SCENARIO_TITLE_SURFACE_THRESHOLD, scenario_title_fields

    b = Backend(db)
    b._catalog()
    plays = [k for k in b._catalog()["detail_by_id"].values() if b._content_kind(k) == "play"]
    rows: list[tuple[float, bool, bool, str, str]] = []
    for k in plays:
        score, surf = scenario_title_fields(str(k.get("name", "")))
        rows.append((score, surf, bool(k.get("is_scenario")), str(k.get("id", "")), str(k.get("name", ""))))

    print(f"plays: {len(rows)}  threshold: {SCENARIO_TITLE_SURFACE_THRESHOLD}")
    surf_n = sum(1 for r in rows if r[1])
    linked_n = sum(1 for r in rows if r[2])
    print(f"title_surface_as_scenario (computed): {surf_n}  is_scenario (DB): {linked_n}")

    w = max(args.bin, 1e-6)
    hist: Counter[str] = Counter()
    for score, _, _, _, _ in rows:
        bkt = int(score / w) * w
        hist[f"{bkt:.3f}-{bkt + w:.3f}"] += 1
    print("\nhistogram (score range -> count):")
    for label in sorted(hist.keys(), key=lambda s: float(s.split("-")[0])):
        print(f"  {label}: {hist[label]}")

    if args.near > 0:
        lo, hi = args.near - w, args.near + w
        near = [(s, sid, name) for s, _, _, sid, name in rows if lo <= s <= hi]
        near.sort(key=lambda t: t[0], reverse=True)
        print(f"\nsample scores in [{lo:.4f}, {hi:.4f}] (up to {args.sample}):")
        for s, sid, name in near[: args.sample]:
            print(f"  {s:.4f}\t{sid}\t{name}")

    return 0


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