scripts/clip_threshold_report.py

#!/usr/bin/env python3
"""
Read-only report: how many CLIP-scored assets pass at various min_similarity cutoffs.

Uses ``data/clip_asset_scores.json`` (same file as ``clip_align_kink_assets.py`` output).

Example::

    python scripts/clip_threshold_report.py
    python scripts/clip_threshold_report.py --db data/store.db

With ``--db``, also reports how many ``Asset`` URLs are missing from the scores file (coverage gap).
"""
from __future__ import annotations

import argparse
import json
import statistics
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


def main() -> int:
    ap = argparse.ArgumentParser(description="CLIP similarity distribution & coverage report")
    ap.add_argument("--scores", type=Path, default=ROOT / "data" / "clip_asset_scores.json")
    ap.add_argument("--db", type=Path, default=None, help="Optional SQLite DB to compare Asset URLs vs scores file")
    args = ap.parse_args()

    path = args.scores.resolve()
    if not path.is_file():
        print(f"No scores file at {path}", file=sys.stderr)
        return 1

    raw = json.loads(path.read_text(encoding="utf-8"))
    sims_accepted: list[float] = []
    sims_all: list[float] = []
    n_false = 0
    for payload in raw.values():
        if not isinstance(payload, dict):
            continue
        try:
            s = float(payload.get("similarity", 0.0))
        except (TypeError, ValueError):
            continue
        sims_all.append(s)
        if payload.get("accepted") is True:
            sims_accepted.append(s)
        elif payload.get("accepted") is False:
            n_false += 1

    print(f"Scores file: {path}")
    print(f"  entries: {len(raw)}  accepted_true: {len(sims_accepted)}  accepted_false: {n_false}")
    if sims_all:
        print(f"  similarity (all rows): min={min(sims_all):.4f} median={statistics.median(sims_all):.4f} max={max(sims_all):.4f}")
    thresholds = [0.18, 0.20, 0.22, 0.24, 0.26, 0.28, 0.30]
    print("  hypothetical passes if min_similarity were set to T (count of rows with sim >= T):")
    for t in thresholds:
        n = sum(1 for s in sims_all if s >= t)
        pct = 100.0 * n / len(sims_all) if sims_all else 0.0
        print(f"    T={t:.2f}  →  {n:6d}  ({pct:5.1f}% of scored rows)")

    if args.db:
        from sqlmodel import Session, create_engine, select

        from models import Asset

        engine = create_engine(f"sqlite:///{args.db.resolve()}")
        with Session(engine) as session:
            urls = [r.asset_url for r in session.exec(select(Asset)).all()]
        in_file = sum(1 for u in urls if u in raw)
        missing = len(urls) - in_file
        print(f"\nDB {args.db}: {len(urls)} asset URLs; {in_file} in scores file; {missing} missing from file")
        if urls:
            print(f"  coverage: {100.0 * in_file / len(urls):.1f}%")
        if missing and missing <= 25:
            for u in urls:
                if u not in raw:
                    print(f"    missing: {u}")

    return 0


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